storage_client.c

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

/**
* Copyright (C) 2008 Happy Fish / YuQing
*
* FastDFS may be copied only under the terms of the GNU General
* Public License V3, which may be found in the FastDFS source kit.
* Please visit the FastDFS Home Page http://www.csource.org/ for more detail.
**/


#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <time.h>
#include <unistd.h>
#include "fdfs_define.h"
#include "logger.h"
#include "fdfs_global.h"
#include "sockopt.h"
#include "shared_func.h"
#include "tracker_types.h"
#include "tracker_proto.h"
#include "tracker_client.h"
#include "storage_client.h"
#include "client_global.h"
#include "fdfs_base64.h"

#define FDFS_SPLIT_GROUP_NAME_AND_FILENAME(file_id) \
	char new_file_id[FDFS_GROUP_NAME_MAX_LEN + 64]; \
	char *group_name; \
	char *filename; \
	char *pSeperator; \
	\
	snprintf(new_file_id, sizeof(new_file_id), "%s", file_id); \
	pSeperator = strchr(new_file_id, FDFS_FILE_ID_SEPERATOR); \
	if (pSeperator == NULL) \
	{ \
		return EINVAL; \
	} \
	\
	*pSeperator = '\0'; \
	group_name = new_file_id; \
	filename =  pSeperator + 1; \

#define storage_get_read_connection(pTrackerServer, \
		ppStorageServer, group_name, filename, \
		pNewStorage, new_connection) \
	storage_get_connection(pTrackerServer, \
		ppStorageServer, TRACKER_PROTO_CMD_SERVICE_QUERY_FETCH, \
		group_name, filename, pNewStorage, new_connection)

#define storage_get_update_connection(pTrackerServer, \
		ppStorageServer, group_name, filename, \
		pNewStorage, new_connection) \
	storage_get_connection(pTrackerServer, \
		ppStorageServer, TRACKER_PROTO_CMD_SERVICE_QUERY_UPDATE, \
		group_name, filename, pNewStorage, new_connection)

static int storage_get_connection(TrackerServerInfo *pTrackerServer, \
		TrackerServerInfo **ppStorageServer, const byte cmd, \
		const char *group_name, const char *filename, \
		TrackerServerInfo *pNewStorage, bool *new_connection)
{
	int result;
	if (*ppStorageServer == NULL)
	{
		if (cmd == TRACKER_PROTO_CMD_SERVICE_QUERY_FETCH)
		{
			result = tracker_query_storage_fetch(pTrackerServer, \
		                pNewStorage, group_name, filename);
		}
		else
		{
			result = tracker_query_storage_update(pTrackerServer, \
		                pNewStorage, group_name, filename);
		}

		if (result != 0)
		{
			return result;
		}

		if ((result=tracker_connect_server(pNewStorage)) != 0)
		{
			return result;
		}

		*ppStorageServer = pNewStorage;
		*new_connection = true;
	}
	else
	{
		if ((*ppStorageServer)->sock >= 0)
		{
			*new_connection = false;
		}
		else
		{
			if ((result=tracker_connect_server(*ppStorageServer)) != 0)
			{
				return result;
			}

			*new_connection = true;
		}
	}

	return 0;
}

static int storage_get_write_connection(TrackerServerInfo *pTrackerServer, \
		TrackerServerInfo **ppStorageServer, 
		TrackerServerInfo *pNewStorage, int *store_path_index, \
		bool *new_connection)
{
	int result;
	if (*ppStorageServer == NULL)
	{
		if ((result=tracker_query_storage_store(pTrackerServer, \
		                pNewStorage, store_path_index)) != 0)
		{
			return result;
		}

		if ((result=tracker_connect_server(pNewStorage)) != 0)
		{
			return result;
		}

		*ppStorageServer = pNewStorage;
		*new_connection = true;
	}
	else
	{
		if ((*ppStorageServer)->sock >= 0)
		{
			*new_connection = false;
		}
		else
		{
			result = tracker_connect_server(*ppStorageServer);
			if (result != 0)
			{
				return result;
			}

			*new_connection = true;
		}
	}

	return 0;
}

int storage_get_metadata1(TrackerServerInfo *pTrackerServer, \
		TrackerServerInfo *pStorageServer,  \
		const char *file_id, \
		FDFSMetaData **meta_list, int *meta_count)
{
	FDFS_SPLIT_GROUP_NAME_AND_FILENAME(file_id)

	return storage_get_metadata(pTrackerServer, pStorageServer, \
			group_name, filename, meta_list, meta_count);
}

int storage_get_metadata(TrackerServerInfo *pTrackerServer, \
			TrackerServerInfo *pStorageServer,  \
			const char *group_name, const char *filename, \
			FDFSMetaData **meta_list, \
			int *meta_count)
{
	TrackerHeader header;
	int result;
	TrackerServerInfo storageServer;
	char out_buff[sizeof(TrackerHeader)+FDFS_GROUP_NAME_MAX_LEN+64];
	int64_t in_bytes;
	int filename_len;
	char *file_buff;
	int64_t file_size;
	bool new_connection;

	file_buff = NULL;
	*meta_list = NULL;
	*meta_count = 0;

	if ((result=storage_get_update_connection(pTrackerServer, \
		&pStorageServer, group_name, filename, \
		&storageServer, &new_connection)) != 0)
	{
		return result;
	}

	while (1)
	{
	/**
	send pkg format:
	FDFS_GROUP_NAME_MAX_LEN bytes: group_name
	remain bytes: filename
	**/

	memset(out_buff, 0, sizeof(out_buff));
	snprintf(out_buff + sizeof(TrackerHeader), sizeof(out_buff) - \
		sizeof(TrackerHeader),  "%s", group_name);
	filename_len = snprintf(out_buff + sizeof(TrackerHeader) + \
			FDFS_GROUP_NAME_MAX_LEN, \
			sizeof(out_buff) - sizeof(TrackerHeader) - \
			FDFS_GROUP_NAME_MAX_LEN,  "%s", filename);

	long2buff(FDFS_GROUP_NAME_MAX_LEN + filename_len, header.pkg_len);
	header.cmd = STORAGE_PROTO_CMD_GET_METADATA;
	header.status = 0;
	memcpy(out_buff, &header, sizeof(TrackerHeader));

	if ((result=tcpsenddata(pStorageServer->sock, out_buff, \
			sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN + \
			filename_len, g_network_timeout)) != 0)
	{
		logError("send data to storage server %s:%d fail, " \
			"errno: %d, error info: %s", \
			pStorageServer->ip_addr, \
			pStorageServer->port, \
			result, strerror(result));

		break;
	}

	if ((result=fdfs_recv_response(pStorageServer, \
		&file_buff, 0, &in_bytes)) != 0)
	{
		break;
	}

	file_size = in_bytes;
	if (file_size == 0)
	{
		break;
	}

	file_buff[in_bytes] = '\0';
	*meta_list = fdfs_split_metadata(file_buff, meta_count, &result);
	break;
	}

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

	if (new_connection)
	{
		fdfs_quit(pStorageServer);
		tracker_disconnect_server(pStorageServer);
	}

	return result;
}

int storage_delete_file1(TrackerServerInfo *pTrackerServer, \
			TrackerServerInfo *pStorageServer, \
			const char *file_id)
{
	FDFS_SPLIT_GROUP_NAME_AND_FILENAME(file_id)

	return storage_delete_file(pTrackerServer, \
			pStorageServer, group_name, filename);
}

int storage_delete_file(TrackerServerInfo *pTrackerServer, \
			TrackerServerInfo *pStorageServer, \
			const char *group_name, const char *filename)
{
	TrackerHeader header;
	int result;
	TrackerServerInfo storageServer;
	char out_buff[sizeof(TrackerHeader)+FDFS_GROUP_NAME_MAX_LEN+64];
	char in_buff[1];
	char *pBuff;
	int64_t in_bytes;
	int filename_len;
	bool new_connection;

	if ((result=storage_get_update_connection(pTrackerServer, \
		&pStorageServer, group_name, filename, \
		&storageServer, &new_connection)) != 0)
	{
		return result;
	}

	while (1)
	{
	/**
	send pkg format:
	FDFS_GROUP_NAME_MAX_LEN bytes: group_name
	remain bytes: filename
	**/

	memset(out_buff, 0, sizeof(out_buff));
	snprintf(out_buff + sizeof(TrackerHeader), sizeof(out_buff) - \
		sizeof(TrackerHeader),  "%s", group_name);
	filename_len = snprintf(out_buff + sizeof(TrackerHeader) + \
			FDFS_GROUP_NAME_MAX_LEN, \
			sizeof(out_buff) - sizeof(TrackerHeader) - \
			FDFS_GROUP_NAME_MAX_LEN,  "%s", filename);

	long2buff(FDFS_GROUP_NAME_MAX_LEN + filename_len, header.pkg_len);
	header.cmd = STORAGE_PROTO_CMD_DELETE_FILE;
	header.status = 0;
	memcpy(out_buff, &header, sizeof(TrackerHeader));

	if ((result=tcpsenddata(pStorageServer->sock, out_buff, \
		sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN + \
		filename_len, g_network_timeout)) != 0)
	{
		logError("send data to storage server %s:%d fail, " \
			"errno: %d, error info: %s", \
			pStorageServer->ip_addr, \
			pStorageServer->port, \
			result, strerror(result));
		break;
	}

	pBuff = in_buff;
	if ((result=fdfs_recv_response(pStorageServer, \
		&pBuff, 0, &in_bytes)) != 0)
	{
		break;
	}

	break;
	}

	if (new_connection)
	{
		fdfs_quit(pStorageServer);
		tracker_disconnect_server(pStorageServer);
	}

	return result;
}

int storage_do_download_file1(TrackerServerInfo *pTrackerServer, \
		TrackerServerInfo *pStorageServer, \
		const int download_type, const char *file_id, \
		char **file_buff, void *arg, int64_t *file_size)
{
	FDFS_SPLIT_GROUP_NAME_AND_FILENAME(file_id)

	return storage_do_download_file(pTrackerServer, pStorageServer, \
		download_type, group_name, filename, \
		file_buff, arg, file_size);
}

int storage_do_download_file(TrackerServerInfo *pTrackerServer, \
		TrackerServerInfo *pStorageServer, const int download_type, \
		const char *group_name, const char *remote_filename, \
		char **file_buff, void *arg, int64_t *file_size)
{
	TrackerHeader header;
	int result;
	TrackerServerInfo storageServer;
	char out_buff[sizeof(TrackerHeader)+FDFS_GROUP_NAME_MAX_LEN+64];
	int64_t in_bytes;
	int filename_len;
	bool new_connection;

	*file_size = 0;
	if ((result=storage_get_read_connection(pTrackerServer, \
		&pStorageServer, group_name, remote_filename, \
		&storageServer, &new_connection)) != 0)
	{
		return result;
	}

	while (1)
	{
	/**
	send pkg format:
	FDFS_GROUP_NAME_MAX_LEN bytes: group_name
	remain bytes: filename
	**/

	memset(out_buff, 0, sizeof(out_buff));
	snprintf(out_buff + sizeof(TrackerHeader), sizeof(out_buff) - \
		sizeof(TrackerHeader),  "%s", group_name);
	filename_len = snprintf(out_buff + sizeof(TrackerHeader) + \
			FDFS_GROUP_NAME_MAX_LEN, \
			sizeof(out_buff) - sizeof(TrackerHeader) - \
			FDFS_GROUP_NAME_MAX_LEN,  "%s", remote_filename);

	long2buff(FDFS_GROUP_NAME_MAX_LEN + filename_len, header.pkg_len);
	header.cmd = STORAGE_PROTO_CMD_DOWNLOAD_FILE;
	header.status = 0;
	memcpy(out_buff, &header, sizeof(TrackerHeader));

	if ((result=tcpsenddata(pStorageServer->sock, out_buff, \
		sizeof(TrackerHeader) + FDFS_GROUP_NAME_MAX_LEN + \
		filename_len, g_network_timeout)) != 0)
	{
		logError("send data to storage server %s:%d fail, " \
			"errno: %d, error info: %s", \
			pStorageServer->ip_addr, \
			pStorageServer->port, \
			result, strerror(result));
		break;
	}

	if (download_type == FDFS_DOWNLOAD_TO_FILE)
	{
		if ((result=fdfs_recv_header(pStorageServer, \
			&in_bytes)) != 0)
		{
			break;
		}

		if ((result=tcprecvfile(pStorageServer->sock, \
				*file_buff, in_bytes, 0)) != 0)
		{
			break;
		}
	}
	else if (download_type == FDFS_DOWNLOAD_TO_BUFF)
	{
		*file_buff = NULL;
		if ((result=fdfs_recv_response(pStorageServer, \
			file_buff, 0, &in_bytes)) != 0)
		{
			break;
		}
	}
	else
	{
		DownloadCallback callback;
		char buff[2048];
		int recv_bytes;
		int64_t remain_bytes;

		if ((result=fdfs_recv_header(pStorageServer, \
			&in_bytes)) != 0)
		{
			break;
		}

		callback = (DownloadCallback)*file_buff;
		remain_bytes = in_bytes;
		while (remain_bytes > 0)
		{
			if (remain_bytes > sizeof(buff))
			{
				recv_bytes = sizeof(buff);
			}
			else
			{
				recv_bytes = remain_bytes;
			}

			if ((result=tcprecvdata(pStorageServer->sock, buff, \
				recv_bytes, g_network_timeout)) != 0)
			{
				logError("recv data from storage server " \
					"%s:%d fail, " \
					"errno: %d, error info: %s", \
					pStorageServer->ip_addr, \
					pStorageServer->port, \
					result, strerror(result));
				break;
			}

			result = callback(arg, in_bytes, buff, recv_bytes);
			if (result != 0)
			{
				logError("call callback function fail, " \
					"error code: %d", result);
				break;
			}

			remain_bytes -= recv_bytes;
		}

		if (remain_bytes != 0)
		{
			break;
		}
	}

	*file_size = in_bytes;
	break;
	}

	if (new_connection)
	{
		fdfs_quit(pStorageServer);
		tracker_disconnect_server(pStorageServer);
	}

	return result;
}