storage_service.c

文件系统源代码!!!!! 文件系统源代码

				"client ip: %s, in request pkg, " \
				"file size: "INT64_PRINTF_FORMAT \
				" != remain bytes: "INT64_PRINTF_FORMAT"", \
				__LINE__, pClientInfo->ip_addr, file_bytes, \
				nInPackLen - (2*FDFS_PROTO_PKG_LEN_SIZE + \
				FDFS_GROUP_NAME_MAX_LEN + filename_len));
			resp.status = EPIPE;
			break;
		}

		if ((resp.status=tcprecvdata(pClientInfo->sock, filename, \
			filename_len, g_network_timeout)) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip: %s, recv filename fail, " \
				"errno: %d, error info: %s.", \
				__LINE__, pClientInfo->ip_addr, \
				resp.status, strerror(resp.status));
			break;
		}
		*(filename + filename_len) = '\0';
		if ((resp.status=storage_split_filename(filename, \
			&filename_len, true_filename, &pBasePath)) != 0)
		{
			break;
		}
		if ((resp.status=fdfs_check_data_filename(true_filename, \
			filename_len)) != 0)
		{
			break;
		}

		snprintf(full_filename, sizeof(full_filename), \
				"%s/data/%s", pBasePath, true_filename);

		if ((proto_cmd == STORAGE_PROTO_CMD_SYNC_CREATE_FILE) && \
			fileExists(full_filename))
		{
			logWarning("file: "__FILE__", line: %d, " \
				"cmd=%d, client ip: %s, data file: %s " \
				"already exists, ignore it", \
				__LINE__, \
				proto_cmd, \
				pClientInfo->ip_addr, full_filename);

			if ((resp.status=tcpdiscard(pClientInfo->sock, \
					file_bytes)) != 0)
			{
				logError("file: "__FILE__", line: %d, " \
					"client ip: %s, discard buff fail, " \
					"buff size: "INT64_PRINTF_FORMAT", " \
					"errno: %d, error info: %s.", \
					__LINE__, pClientInfo->ip_addr, \
					file_bytes, \
					resp.status, strerror(resp.status));
				break;
			}
		}
		//else if ((resp.status=recv_file_serialized(pClientInfo->sock, 
		else if ((resp.status=tcprecvfile(pClientInfo->sock, 
				full_filename, file_bytes, \
				g_fsync_after_written_bytes)) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip: %s, recv file buff fail, " \
				"file size: "INT64_PRINTF_FORMAT \
				", errno: %d, error info: %s.", \
				__LINE__, pClientInfo->ip_addr, \
				file_bytes, resp.status, strerror(resp.status));
			break;
		}

		if (proto_cmd == STORAGE_PROTO_CMD_SYNC_CREATE_FILE)
		{
			resp.status = storage_binlog_write(*timestamp, \
				STORAGE_OP_TYPE_REPLICA_CREATE_FILE, \
				filename);
		}
		else
		{
			resp.status = storage_binlog_write(*timestamp, \
				STORAGE_OP_TYPE_REPLICA_UPDATE_FILE, \
				filename);
		}

		break;
	}

	resp.cmd = STORAGE_PROTO_CMD_RESP;
	if ((result=tcpsenddata(pClientInfo->sock, \
		&resp, sizeof(resp), g_network_timeout)) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"client ip: %s, send data fail, " \
			"errno: %d, error info: %s", \
			__LINE__, pClientInfo->ip_addr, \
			result, strerror(result));
		return result;
	}

	return resp.status;
}

/**
pkg format:
Header
FDFS_GROUP_NAME_MAX_LEN bytes: group_name
filename
**/
static int storage_get_metadata(StorageClientInfo *pClientInfo, \
				const int64_t nInPackLen)
{
	TrackerHeader resp;
	int result;
	char in_buff[FDFS_GROUP_NAME_MAX_LEN + 64];
	char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
	char full_filename[MAX_PATH_SIZE+sizeof(in_buff)+64];
	char true_filename[64];
	char *filename;
	char *pBasePath;
	char *file_buff;
	int filename_len;
	int64_t file_bytes;

	memset(&resp, 0, sizeof(resp));
	file_buff = NULL;
	file_bytes = 0;
	while (1)
	{
		if (nInPackLen <= FDFS_GROUP_NAME_MAX_LEN)
		{
			logError("file: "__FILE__", line: %d, " \
				"cmd=%d, client ip: %s, package size "INT64_PRINTF_FORMAT" " \
				"is not correct, " \
				"expect length > %d", \
				__LINE__, \
				STORAGE_PROTO_CMD_UPLOAD_FILE, \
				pClientInfo->ip_addr,  \
				nInPackLen, FDFS_GROUP_NAME_MAX_LEN);
			resp.status = EINVAL;
			break;
		}

		if (nInPackLen >= sizeof(in_buff))
		{
			logError("file: "__FILE__", line: %d, " \
				"cmd=%d, client ip: %s, package size "INT64_PRINTF_FORMAT" " \
				"is too large, " \
				"expect length should < %d", \
				__LINE__, \
				STORAGE_PROTO_CMD_UPLOAD_FILE, \
				pClientInfo->ip_addr,  \
				nInPackLen, sizeof(in_buff));
			resp.status = EINVAL;
			break;
		}

		if ((resp.status=tcprecvdata(pClientInfo->sock, in_buff, \
			nInPackLen, g_network_timeout)) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, recv data fail, " \
				"errno: %d, error info: %s.", \
				__LINE__, pClientInfo->ip_addr, \
				resp.status, strerror(resp.status));
			break;
		}

		memcpy(group_name, in_buff, FDFS_GROUP_NAME_MAX_LEN);
		group_name[FDFS_GROUP_NAME_MAX_LEN] = '\0';
		if (strcmp(group_name, g_group_name) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, group_name: %s " \
				"not correct, should be: %s", \
				__LINE__, pClientInfo->ip_addr, \
				group_name, g_group_name);
			resp.status = EINVAL;
			break;
		}

		*(in_buff + nInPackLen) = '\0';
		filename = in_buff + FDFS_GROUP_NAME_MAX_LEN;
		filename_len = nInPackLen - FDFS_GROUP_NAME_MAX_LEN;
		if ((resp.status=storage_split_filename(filename, \
			&filename_len, true_filename, &pBasePath)) != 0)
		{
			break;
		}
		if ((resp.status=fdfs_check_data_filename(true_filename, \
			filename_len)) != 0)
		{
			break;
		}

		sprintf(full_filename, "%s/data/%s", pBasePath, true_filename);
		if (!fileExists(full_filename))
		{
			resp.status = ENOENT;
			break;
		}

		strcat(full_filename, STORAGE_META_FILE_EXT);
		resp.status = getFileContent(full_filename, \
				&file_buff, &file_bytes);
		if (resp.status == ENOENT)
		{
			resp.status = 0;
		}
		break;
	}

	resp.cmd = STORAGE_PROTO_CMD_RESP;
	long2buff(file_bytes, resp.pkg_len);
	if ((result=tcpsenddata(pClientInfo->sock, \
		&resp, sizeof(resp), g_network_timeout)) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"client ip: %s, send data fail, " \
			"errno: %d, error info: %s", \
			__LINE__, pClientInfo->ip_addr, \
			result, strerror(result));

		if (file_buff != NULL)
		{
			free(file_buff);
		}
		return result;
	}

	if (resp.status != 0)
	{
		if (file_buff != NULL)
		{
			free(file_buff);
		}

		return resp.status;
	}


	if (file_buff != NULL)
	{
		result = tcpsenddata(pClientInfo->sock, \
			file_buff, file_bytes, g_network_timeout);
		free(file_buff);
		if(result != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip: %s, send data fail, " \
				"errno: %d, error info: %s", \
				__LINE__, pClientInfo->ip_addr, \
				result, strerror(result));

			return result;
		}
	}

	return resp.status;
}

/**
pkg format:
Header
FDFS_GROUP_NAME_MAX_LEN bytes: group_name
filename
**/
static int storage_download_file(StorageClientInfo *pClientInfo, \
				const int64_t nInPackLen)
{
	TrackerHeader resp;
	int result;
	char in_buff[FDFS_GROUP_NAME_MAX_LEN + 64];
	char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
	char full_filename[MAX_PATH_SIZE+sizeof(in_buff)+16];
	char true_filename[64];
	char *pBasePath;
	char *filename;
	int filename_len;
	int64_t file_bytes;
	struct stat stat_buf;

	memset(&resp, 0, sizeof(resp));
	file_bytes = 0;
	while (1)
	{
		if (nInPackLen <= FDFS_GROUP_NAME_MAX_LEN)
		{
			logError("file: "__FILE__", line: %d, " \
				"cmd=%d, client ip: %s, package size "INT64_PRINTF_FORMAT" " \
				"is not correct, " \
				"expect length > %d", \
				__LINE__, \
				STORAGE_PROTO_CMD_UPLOAD_FILE, \
				pClientInfo->ip_addr,  \
				nInPackLen, FDFS_GROUP_NAME_MAX_LEN);
			resp.status = EINVAL;
			break;
		}

		if (nInPackLen >= sizeof(in_buff))
		{
			logError("file: "__FILE__", line: %d, " \
				"cmd=%d, client ip: %s, package size "INT64_PRINTF_FORMAT" " \
				"is too large, " \
				"expect length should < %d", \
				__LINE__, \
				STORAGE_PROTO_CMD_UPLOAD_FILE, \
				pClientInfo->ip_addr,  \
				nInPackLen, sizeof(in_buff));
			resp.status = EINVAL;
			break;
		}

		if ((resp.status=tcprecvdata(pClientInfo->sock, in_buff, \
				nInPackLen, g_network_timeout)) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, recv data fail, " \
				"errno: %d, error info: %s.", \
				__LINE__, pClientInfo->ip_addr, \
				resp.status, strerror(resp.status));
			break;
		}

		memcpy(group_name, in_buff, FDFS_GROUP_NAME_MAX_LEN);
		group_name[FDFS_GROUP_NAME_MAX_LEN] = '\0';
		if (strcmp(group_name, g_group_name) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, group_name: %s " \
				"not correct, should be: %s", \
				__LINE__, pClientInfo->ip_addr, \
				group_name, g_group_name);
			resp.status = EINVAL;
			break;
		}

		*(in_buff + nInPackLen) = '\0';
		filename = in_buff + FDFS_GROUP_NAME_MAX_LEN;
		filename_len = nInPackLen - FDFS_GROUP_NAME_MAX_LEN;
		if ((resp.status=storage_split_filename(filename, \
			&filename_len, true_filename, &pBasePath)) != 0)
		{
			break;
		}

		if ((resp.status=fdfs_check_data_filename(true_filename, \
			filename_len)) != 0)
		{
			break;
		}

		sprintf(full_filename, "%s/data/%s", pBasePath, true_filename);
		if (stat(full_filename, &stat_buf) == 0)
		{
			if (!S_ISREG(stat_buf.st_mode))
			{
				logError("file: "__FILE__", line: %d, " \
					"%s is not a regular file", \
					__LINE__, full_filename);
				resp.status = EISDIR;
				break;
			}

			resp.status = 0;
			file_bytes = stat_buf.st_size;
		}
		else
		{
			resp.status = errno != 0 ? errno : ENOENT;

			logError("file: "__FILE__", line: %d, " \
				"call stat fail, file: %s, "\
				"error no: %d, error info: %s", \
				__LINE__, full_filename, \
				errno, strerror(errno));
		}

		break;
	}

	resp.cmd = STORAGE_PROTO_CMD_RESP;
	long2buff(file_bytes, resp.pkg_len);

	if ((result=tcpsenddata(pClientInfo->sock, \
		&resp, sizeof(resp), g_network_timeout)) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"client ip: %s, send data fail, " \
			"errno: %d, error info: %s", \
			__LINE__, pClientInfo->ip_addr, \
			result, strerror(result));
		return result;
	}

	if (resp.status != 0)
	{
		return resp.status;
	}

	result = tcpsendfile(pClientInfo->sock, \
		full_filename, file_bytes);
	if(result != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"client ip: %s, send file fail, " \
			"errno: %d, error info: %s", \
			__LINE__, pClientInfo->ip_addr, \
			result, strerror(result));

		return result;
	}

	return resp.status;
}

/**
pkg format:
Header
4 bytes: source delete timestamp
FDFS_GROUP_NAME_MAX_LEN bytes: group_name
filename
**/
static int storage_sync_delete_file(StorageClientInfo *pClientInfo, \
		const int64_t nInPackLen, int *timestamp)
{
	TrackerHeader resp;
	char in_buff[FDFS_GROUP_NAME_MAX_LEN + 64];
	char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
	char full_filename[MAX_PATH_SIZE+sizeof(in_buff)];
	char true_filename[64];
	char *pBasePath;
	char *filename;
	int filename_len;
	int result;

	memset(&resp, 0, sizeof(resp));
	while (1)
	{
		if (nInPackLen <= 4 + FDFS_GROUP_NAME_MAX_LEN)
		{
			logError("file: "__FILE__", line: %d, " \
				"cmd=%d, client ip: %s, package size " \
				INT64_PRINTF_FORMAT" is not correct, " \
				"expect length <= %d", \
				__LINE__, \
				STORAGE_PROTO_CMD_SYNC_DELETE_FILE, \
				pClientInfo->ip_addr,  \
				nInPackLen, 4 + FDFS_GROUP_NAME_MAX_LEN);
			resp.status = EINVAL;
			break;
		}

		if (nInPackLen >= sizeof(in_buff))
		{
			logError("file: "__FILE__", line: %d, " \
				"cmd=%d, client ip: %s, package size " \
				INT64_PRINTF_FORMAT" is too large, " \
				"expect length should < %d", \
				__LINE__, \
				STORAGE_PROTO_CMD_SYNC_DELETE_FILE, \
				pClientInfo->ip_addr,  \
				nInPackLen, sizeof(in_buff));
			resp.status = EINVAL;
			break;
		}

		if ((resp.status=tcprecvdata(pClientInfo->sock, in_buff, \
			nInPackLen, g_network_timeout)) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, recv data fail, " \
				"errno: %d, error info: %s.", \
				__LINE__, pClientInfo->ip_addr, \
				resp.status, strerror(resp.status));
			break;
		}

		*timestamp = buff2int(in_buff);
		memcpy(group_name, in_buff + 4, FDFS_GROUP_NAME_MAX_LEN);
		group_name[FDFS_GROUP_NAME_MAX_LEN] = '\0';
		if (strcmp(group_name, g_group_name) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, group_name: %s " \
				"not correct, should be: %s", \
				__LINE__, pClientInfo->ip_addr, \
				group_name, g_group_name);
			resp.status = EINVAL;
			break;
		}

		*(in_buff + nInPackLen) = '\0';
		filename = in_buff + 4 + FDFS_GROUP_NAME_MAX_LEN;
		filename_len = nInPackLen - (4 + FDFS_GROUP_NAME_MAX_LEN);
		if ((resp.status=storage_split_filename(filename, \
			&filename_len, true_filename, &pBasePath)) != 0)
		{
			break;
		}
		if ((resp.status=fdfs_check_data_filename(true_filename, \
			filename_len)) != 0)
		{
			break;
		}

		sprintf(full_filename, "%s/data/%s", pBasePath, true_filename);
		if (unlink(full_filename) != 0)
		{
			if (errno == ENOENT)
			{
				logWarning("file: "__FILE__", line: %d, " \
					"cmd=%d, client ip: %s, " \
					"file %s not exist, " \
					"maybe delete later?", \
					__LINE__, \
					STORAGE_PROTO_CMD_SYNC_DELETE_FILE, \
					pClientInfo->ip_addr, full_filename);
			}
			else
			{
				logError("file: "__FILE__", line: %d, " \
					"client ip: %s, delete file %s fail," \
					"errno: %d, error info: %s", \
					__LINE__, pClientInfo->ip_addr, \
					full_filename, errno, strerror(errno));