storage_service.c

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

	while (pOldMeta < pOldMetaEnd)
	{
		memcpy(pAllMeta, pOldMeta, sizeof(FDFSMetaData));
		pOldMeta++;
		pAllMeta++;
	}

	while (pNewMeta < pNewMetaEnd)
	{
		memcpy(pAllMeta, pNewMeta, sizeof(FDFSMetaData));
		pNewMeta++;
		pAllMeta++;
	}

	free(file_buff);
	free(old_meta_list);
	free(new_meta_list);

	all_meta_buff = fdfs_pack_metadata(all_meta_list, \
			pAllMeta - all_meta_list, NULL, &all_meta_bytes);
	free(all_meta_list);
	if (all_meta_buff == NULL)
	{
		return ENOMEM;
	}

	*sync_flag = STORAGE_OP_TYPE_SOURCE_UPDATE_FILE;
	result = writeToFile(full_meta_filename, all_meta_buff, all_meta_bytes);
	free(all_meta_buff);

	return result;
}

/**
8 bytes: filename length
8 bytes: meta data size
1 bytes: operation flag, 
     'O' for overwrite all old metadata
     'M' for merge, insert when the meta item not exist, otherwise update it
FDFS_GROUP_NAME_MAX_LEN bytes: group_name
filename
meta data bytes: each meta data seperated by \x01,
		 name and value seperated by \x02
**/
static int storage_set_metadata(StorageClientInfo *pClientInfo, \
				const int64_t nInPackLen)
{
	TrackerHeader resp;
	char *in_buff;
	char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
	char filename[64];
	char true_filename[64];
	char meta_filename[64+sizeof(STORAGE_META_FILE_EXT)];
	char full_filename[MAX_PATH_SIZE + 64 + sizeof(STORAGE_META_FILE_EXT)];
	char op_flag;
	char sync_flag;
	char *meta_buff;
	int meta_bytes;
	char *pBasePath;
	int filename_len;
	int true_filename_len;
	int result;

	memset(&resp, 0, sizeof(resp));
	in_buff = NULL;
	while (1)
	{
		if (nInPackLen <= 2 * FDFS_PROTO_PKG_LEN_SIZE + 1 + \
					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_SET_METADATA, \
				pClientInfo->ip_addr,  \
				nInPackLen, 2 * FDFS_PROTO_PKG_LEN_SIZE + 1 \
				+ FDFS_GROUP_NAME_MAX_LEN);
			resp.status = EINVAL;
			break;
		}

		in_buff = (char *)malloc(nInPackLen + 1);
		if (in_buff == NULL)
		{
			resp.status = ENOMEM;

			logError("file: "__FILE__", line: %d, " \
				"malloc "INT64_PRINTF_FORMAT" bytes fail", \
				__LINE__, nInPackLen + 1);
			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;
		}

		*(in_buff + nInPackLen) = '\0';
		filename_len = buff2long(in_buff);
		meta_bytes = buff2long(in_buff + FDFS_PROTO_PKG_LEN_SIZE);
		if (filename_len <= 0 || filename_len >= sizeof(filename))
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, invalid filename length: %d", \
				__LINE__, pClientInfo->ip_addr, filename_len);
			resp.status = EINVAL;
			break;
		}

		op_flag = *(in_buff + 2 * FDFS_PROTO_PKG_LEN_SIZE);
		if (op_flag != STORAGE_SET_METADATA_FLAG_OVERWRITE && \
			op_flag != STORAGE_SET_METADATA_FLAG_MERGE)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, " \
				"invalid operation flag: 0x%02X", \
				__LINE__, pClientInfo->ip_addr, op_flag);
			resp.status = EINVAL;
			break;
		}

		if (meta_bytes < 0 || meta_bytes != nInPackLen - \
				(2 * FDFS_PROTO_PKG_LEN_SIZE + 1 + \
				FDFS_GROUP_NAME_MAX_LEN + filename_len))
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, invalid meta bytes: %d", \
				__LINE__, pClientInfo->ip_addr, meta_bytes);
			resp.status = EINVAL;
			break;
		}

		memcpy(group_name, in_buff + 2*FDFS_PROTO_PKG_LEN_SIZE+1, \
			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;
		}

		memcpy(filename, in_buff + 2 * FDFS_PROTO_PKG_LEN_SIZE + 1 + \
			FDFS_GROUP_NAME_MAX_LEN, filename_len);
		*(filename + filename_len) = '\0';
		true_filename_len = filename_len;
		if ((resp.status=storage_split_filename(filename, \
			&true_filename_len, true_filename, &pBasePath)) != 0)
		{
			break;
		}
		if ((resp.status=fdfs_check_data_filename(true_filename, \
			true_filename_len)) != 0)
		{
			break;
		}

		meta_buff = in_buff + 2 * FDFS_PROTO_PKG_LEN_SIZE + 1 + \
				FDFS_GROUP_NAME_MAX_LEN + filename_len;
		*(meta_buff + meta_bytes) = '\0';

		sprintf(full_filename, "%s/data/%s",pBasePath,true_filename);
		if (!fileExists(full_filename))
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, filename: %s not exist", \
				__LINE__, pClientInfo->ip_addr, true_filename);
			resp.status = ENOENT;
			break;
		}

		sprintf(meta_filename,"%s"STORAGE_META_FILE_EXT,true_filename);
		strcat(full_filename, STORAGE_META_FILE_EXT);

		resp.status = storage_do_set_metadata(pClientInfo, \
			full_filename, meta_buff, meta_bytes, \
			op_flag, &sync_flag);
		if (resp.status != 0)
		{
			break;
		}

		if (sync_flag != '\0')
		{
			sprintf(meta_filename,"%s"STORAGE_META_FILE_EXT,filename);
			resp.status = storage_binlog_write( \
					time(NULL), sync_flag, meta_filename);
		}

		break;
	}

	resp.cmd = STORAGE_PROTO_CMD_RESP;

	if (in_buff != NULL)
	{
		free(in_buff);
	}

	if ((result=tcpsenddata(pClientInfo->sock, (void *)&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;
}

/**
1 byte: store path index
8 bytes: meta data bytes
8 bytes: file size 
FDFS_FILE_EXT_NAME_MAX_LEN bytes: file ext name, do not include dot (.)
meta data bytes: each meta data seperated by \x01,
		 name and value seperated by \x02
file size bytes: file content
**/
static int storage_upload_file(StorageClientInfo *pClientInfo, \
				const int64_t nInPackLen)
{
	TrackerHeader resp;
	int out_len;
	char in_buff[1+2*FDFS_PROTO_PKG_LEN_SIZE+FDFS_FILE_EXT_NAME_MAX_LEN+1];
	char meta_buff[64 * 1024];
	char *pMetaData;
	char *file_ext_name;
	char out_buff[128];
	char filename[128];
	int meta_bytes;
	int64_t file_bytes;
	int filename_len;
	int result;
	int store_path_index;

	memset(&resp, 0, sizeof(resp));
	pMetaData = meta_buff;
	filename[0] = '\0';
	filename_len = 0;
	while (1)
	{
		if (nInPackLen < 1 + 2 * FDFS_PROTO_PKG_LEN_SIZE + 
				FDFS_FILE_EXT_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, 1 + 2 * FDFS_PROTO_PKG_LEN_SIZE + \
				FDFS_FILE_EXT_NAME_MAX_LEN);
			resp.status = EINVAL;
			break;
		}

		if ((resp.status=tcprecvdata(pClientInfo->sock, in_buff, \
			1+2*FDFS_PROTO_PKG_LEN_SIZE+FDFS_FILE_EXT_NAME_MAX_LEN,\
			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;
		}

		store_path_index = *in_buff;
		if (store_path_index < 0 || store_path_index >= g_path_count)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip: %s, store_path_index: %d " \
				"is invalid", __LINE__, \
				pClientInfo->ip_addr, store_path_index);
			resp.status = EINVAL;
			break;
		}

		meta_bytes = buff2long(in_buff+1);
		file_bytes = buff2long(in_buff + 1+FDFS_PROTO_PKG_LEN_SIZE);
		if (meta_bytes < 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip:%s, pkg length is not correct, " \
				"invalid meta bytes: %d", \
				__LINE__, pClientInfo->ip_addr, \
				meta_bytes);
			resp.status = EINVAL;
			break;
		}

		if (file_bytes < 0 || file_bytes != nInPackLen - \
			(1 + 2 * FDFS_PROTO_PKG_LEN_SIZE + \
			FDFS_FILE_EXT_NAME_MAX_LEN + meta_bytes))
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip: %s, pkg length is not correct, " \
				"invalid file bytes: "INT64_PRINTF_FORMAT"", \
				__LINE__, pClientInfo->ip_addr, \
				file_bytes);
			resp.status = EINVAL;
			break;
		}

		file_ext_name = in_buff + 1 + 2 * FDFS_PROTO_PKG_LEN_SIZE;
		file_ext_name[FDFS_FILE_EXT_NAME_MAX_LEN] = '\0';

		if (meta_bytes > 0)
		{
			if (meta_bytes > sizeof(meta_buff))
			{
				pMetaData = (char *)malloc(meta_bytes + 1);
				if (pMetaData == NULL)
				{
					resp.status = ENOMEM;

					logError("file: "__FILE__", line: %d, " \
						"malloc %d bytes fail", \
						__LINE__, meta_bytes + 1);
					break;
				}
			}

			if ((resp.status=tcprecvdata(pClientInfo->sock, \
				pMetaData, meta_bytes, 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;
			}

			*(pMetaData + meta_bytes) = '\0';
		}
		else
		{
			*pMetaData = '\0';
		}

		resp.status = storage_save_file(pClientInfo, store_path_index,\
			file_bytes, file_ext_name, pMetaData, meta_bytes, \
			filename, &filename_len);

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

		resp.status = storage_binlog_write(time(NULL), \
				STORAGE_OP_TYPE_SOURCE_CREATE_FILE, filename);
		if (resp.status != 0)
		{
			break;
		}

		if (meta_bytes > 0)
		{
			char meta_filename[64];
			sprintf(meta_filename, "%s"STORAGE_META_FILE_EXT, \
				filename);
			resp.status = storage_binlog_write(time(NULL), \
					STORAGE_OP_TYPE_SOURCE_CREATE_FILE, \
					meta_filename);
		}

		break;
	}

	resp.cmd = STORAGE_PROTO_CMD_RESP;
	if (resp.status == 0)
	{
		out_len = FDFS_GROUP_NAME_MAX_LEN + filename_len;
		long2buff(out_len, resp.pkg_len);

		memcpy(out_buff, &resp, sizeof(resp));
		memcpy(out_buff+sizeof(resp), g_group_name, \
			FDFS_GROUP_NAME_MAX_LEN);
		memcpy(out_buff+sizeof(resp)+FDFS_GROUP_NAME_MAX_LEN, \
			filename, filename_len);
	}
	else
	{
		out_len = 0;
		long2buff(out_len, resp.pkg_len);
		memcpy(out_buff, &resp, sizeof(resp));
	}

	if (pMetaData != meta_buff && pMetaData != NULL)
	{
		free(pMetaData);
	}

	if ((result=tcpsenddata(pClientInfo->sock, out_buff, \
		sizeof(resp) + out_len, 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;
}

/**
8 bytes: filename bytes
8 bytes: file size
4 bytes: source op timestamp
FDFS_GROUP_NAME_MAX_LEN bytes: group_name
filename bytes : filename
file size bytes: file content
**/
static int storage_sync_copy_file(StorageClientInfo *pClientInfo, \
		const int64_t nInPackLen, const char proto_cmd, int *timestamp)
{
	TrackerHeader resp;
	char in_buff[2 * FDFS_PROTO_PKG_LEN_SIZE + \
			4 + FDFS_GROUP_NAME_MAX_LEN + 1];
	char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
	char true_filename[64];
	char filename[64];
	char full_filename[MAX_PATH_SIZE];
	char *pBasePath;
	int filename_len;
	int64_t file_bytes;
	int result;

	memset(&resp, 0, sizeof(resp));
	while (1)
	{
		if (nInPackLen <= 2 * FDFS_PROTO_PKG_LEN_SIZE + \
					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__, \
				proto_cmd, \
				pClientInfo->ip_addr,  nInPackLen, \
				2 * FDFS_PROTO_PKG_LEN_SIZE + \
					4 + FDFS_GROUP_NAME_MAX_LEN);
			resp.status = EINVAL;
			break;
		}

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

		filename_len = buff2long(in_buff);
		file_bytes = buff2long(in_buff + FDFS_PROTO_PKG_LEN_SIZE);
		if (filename_len < 0 || filename_len >= sizeof(filename))
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip: %s, in request pkg, " \
				"filename length: %d is invalid, " \
				"which < 0 or >= %d", \
				__LINE__, pClientInfo->ip_addr, \
				filename_len,  sizeof(filename));
			resp.status = EPIPE;
			break;
		}

		if (file_bytes < 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip: %s, in request pkg, " \
				"file size: "INT64_PRINTF_FORMAT" is invalid, "\
				"which < 0", __LINE__, \
				pClientInfo->ip_addr, file_bytes);
			resp.status = EPIPE;
			break;
		}

		*timestamp = buff2int(in_buff + 2 * FDFS_PROTO_PKG_LEN_SIZE);
		memcpy(group_name, in_buff + 2 * FDFS_PROTO_PKG_LEN_SIZE + 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;
		}

		if (file_bytes != nInPackLen - (2*FDFS_PROTO_PKG_LEN_SIZE + \
				4 + FDFS_GROUP_NAME_MAX_LEN + filename_len))
		{
			logError("file: "__FILE__", line: %d, " \