storage_service.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.
**/

//storage_service.c

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/stat.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <time.h>
#include <sys/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 "storage_service.h"
#include "storage_func.h"
#include "storage_sync.h"
#include "storage_global.h"
#include "fdfs_base64.h"
#include "hash.h"

pthread_mutex_t g_storage_thread_lock;
int g_storage_thread_count = 0;

static pthread_mutex_t path_index_thread_lock;
static pthread_mutex_t stat_count_thread_lock;

int storage_service_init()
{
	int result;

	if ((result=init_pthread_lock(&g_storage_thread_lock)) != 0)
	{
		return result;
	}

	if ((result=init_pthread_lock(&path_index_thread_lock)) != 0)
	{
		return result;
	}

	if ((result=init_pthread_lock(&stat_count_thread_lock)) != 0)
	{
		return result;
	}

	return result;
}

void storage_service_destroy()
{
	pthread_mutex_destroy(&g_storage_thread_lock);
	pthread_mutex_destroy(&path_index_thread_lock);
	pthread_mutex_destroy(&stat_count_thread_lock);
}

static int storage_gen_filename(StorageClientInfo *pClientInfo, \
		const int file_size, \
		const char *szFormattedExt, const int ext_name_len, \
		const time_t timestamp, char *filename, int *filename_len)
{
	int result;
	int r;
	char buff[sizeof(int) * 4];
	char encoded[sizeof(int) * 6 + 1];
	char szStorageIp[IP_ADDRESS_SIZE];
	int n;
	int len;
	in_addr_t server_ip;

	server_ip = getSockIpaddr(pClientInfo->sock, \
			szStorageIp, IP_ADDRESS_SIZE);
	r = rand();

	int2buff(server_ip, buff);
	int2buff(timestamp, buff+sizeof(int));
	int2buff(file_size, buff+sizeof(int)*2);
	int2buff(r, buff+sizeof(int)*3);

	base64_encode_ex(buff, sizeof(int) * 4, encoded, filename_len, false);

	if (g_file_distribute_path_mode == FDFS_FILE_DIST_PATH_ROUND_ROBIN)
	{
		len = sprintf(buff, STORAGE_DATA_DIR_FORMAT"/", \
				g_dist_path_index_high);
		len += sprintf(buff + len, STORAGE_DATA_DIR_FORMAT"/", \
				g_dist_path_index_low);

		if (++g_dist_write_file_count >= g_file_distribute_rotate_count)
		{
			g_dist_write_file_count = 0;
	
			if ((result=pthread_mutex_lock( \
					&path_index_thread_lock)) != 0)
			{
				logError("file: "__FILE__", line: %d, " \
					"call pthread_mutex_lock fail, " \
					"errno: %d, error info: %s", \
					__LINE__, result, strerror(result));
			}

			++g_dist_path_index_low;
			if (g_dist_path_index_low >= g_subdir_count_per_path)
			{  //rotate
				g_dist_path_index_high++;
				if (g_dist_path_index_high >= \
					g_subdir_count_per_path)  //rotate
				{
					g_dist_path_index_high = 0;
				}
				g_dist_path_index_low = 0;
			}

			if (++g_stat_change_count%STORAGE_SYNC_STAT_FILE_FREQ==0)
			{
				if (storage_write_to_stat_file() != 0)
				{
				}
			}

			if ((result=pthread_mutex_unlock( \
					&path_index_thread_lock)) != 0)
			{
				logError("file: "__FILE__", line: %d, " \
					"call pthread_mutex_unlock fail, " \
					"errno: %d, error info: %s", \
					__LINE__, result, strerror(result));
			}
		}
	}  //random
	else
	{
		n = PJWHash(encoded, *filename_len) % (1 << 16);

		len = sprintf(buff,STORAGE_DATA_DIR_FORMAT"/",((n >> 8)&0xFF) \
				% g_subdir_count_per_path);
		len += sprintf(buff+len, STORAGE_DATA_DIR_FORMAT"/",(n & 0xFF)\
				 % g_subdir_count_per_path);
	}

	memcpy(filename, buff, len);
	memcpy(filename+len, encoded, *filename_len);
	memcpy(filename+len+(*filename_len), szFormattedExt, ext_name_len);
	*filename_len += len + ext_name_len;
	*(filename + (*filename_len)) = '\0';

	return 0;
}

static int storage_sort_metadata_buff(char *meta_buff, const int meta_size)
{
	FDFSMetaData *meta_list;
	int meta_count;
	int meta_bytes;
	int result;

	meta_list = fdfs_split_metadata(meta_buff, &meta_count, &result);
	if (meta_list == NULL)
	{
		return result;
	}

	/*
	{
	int i;
	//printf("meta_count=%d\n", meta_count);
	for (i=0; i<meta_count; i++)
	{
		//printf("%d. %s=%s\n", i+1, meta_list[i].name, meta_list[i].value);
	}
	}
	*/

	qsort((void *)meta_list, meta_count, sizeof(FDFSMetaData), \
		metadata_cmp_by_name);

	fdfs_pack_metadata(meta_list, meta_count, meta_buff, &meta_bytes);
	free(meta_list);

	return 0;
}

static int storage_save_file(StorageClientInfo *pClientInfo, \
		const int store_path_index, const int file_size, \
		const char *file_ext_name,  \
		char *meta_buff, const int meta_size, \
		char *filename, int *filename_len)
{
#define FILE_TIMESTAMP_ADVANCED_SECS	30

	int result;
	int i;
	char *pStorePath;
	char full_filename[MAX_PATH_SIZE+64];
	char meta_filename[MAX_PATH_SIZE+64];
	char szFormattedExt[FDFS_FILE_EXT_NAME_MAX_LEN + 2];
	char *p;
	int ext_name_len;
	int pad_len;
	time_t start_time;
	time_t end_time;

	ext_name_len = strlen(file_ext_name);
	if (ext_name_len == 0)
	{
		pad_len = FDFS_FILE_EXT_NAME_MAX_LEN + 1;
	}
	else
	{
		pad_len = FDFS_FILE_EXT_NAME_MAX_LEN - ext_name_len;
	}

	p = szFormattedExt;
	for (i=0; i<pad_len; i++)
	{
		*p++ = '0' + (int)(10.0 * (double)rand() / RAND_MAX);
	}

	if (ext_name_len > 0)
	{
		*p++ = '.';
		memcpy(p, file_ext_name, ext_name_len);
		p += ext_name_len;
	}
	*p = '\0';

	pStorePath = g_store_paths[store_path_index];
	start_time = time(NULL);
	for (i=0; i<10; i++)
	{
		if ((result=storage_gen_filename(pClientInfo, file_size, \
				szFormattedExt, FDFS_FILE_EXT_NAME_MAX_LEN+1, \
				start_time + FILE_TIMESTAMP_ADVANCED_SECS, \
				filename, filename_len)) != 0)
		{
			return result;
		}

		sprintf(full_filename, "%s/data/%s", pStorePath, filename);
		if (!fileExists(full_filename))
		{
			break;
		}

		*full_filename = '\0';
	}

	if (*full_filename == '\0')
	{
		logError("file: "__FILE__", line: %d, " \
			"Can't generate uniq filename", __LINE__);
		*filename = '\0';
		*filename_len = 0;
		return ENOENT;
	}

	//if ((result=recv_file_serialized(pClientInfo->sock, 
	if ((result=tcprecvfile(pClientInfo->sock, 
		full_filename, file_size, g_fsync_after_written_bytes)) != 0)
	{
		*filename = '\0';
		*filename_len = 0;
		return result;
	}

	if (meta_size > 0)
	{
		if ((result=storage_sort_metadata_buff(meta_buff, \
				meta_size)) != 0)
		{
			*filename = '\0';
			*filename_len = 0;
			unlink(full_filename);
			return result;
		}

		sprintf(meta_filename, "%s"STORAGE_META_FILE_EXT, \
				full_filename);
		if ((result=writeToFile(meta_filename, meta_buff, \
				meta_size)) != 0)
		{
			*filename = '\0';
			*filename_len = 0;
			unlink(full_filename);
			return result;
		}
	}
	else
	{
		*meta_filename = '\0';
	}

	end_time = time(NULL);
	if (end_time - start_time > FILE_TIMESTAMP_ADVANCED_SECS)
	{  //need to rename filename
		char new_full_filename[MAX_PATH_SIZE+64];
		char new_meta_filename[MAX_PATH_SIZE+64];
		char new_filename[64];
		int new_filename_len;

		for (i=0; i<10; i++)
		{
			if ((result=storage_gen_filename(pClientInfo,file_size,\
				szFormattedExt, FDFS_FILE_EXT_NAME_MAX_LEN+1, \
				end_time, new_filename, &new_filename_len))!=0)
			{
				return 0;
			}

			sprintf(new_full_filename, "%s/data/%s", \
				pStorePath, new_filename);
			if (!fileExists(new_full_filename))
			{
				break;
			}

			*new_full_filename = '\0';
		}

		if (*full_filename == '\0')
		{
			logWarning("file: "__FILE__", line: %d, " \
				"Can't generate uniq filename", __LINE__);
			return 0;
		}

		if (rename(full_filename, new_full_filename) != 0)
		{
			logWarning("file: "__FILE__", line: %d, " \
				"rename %s to %s fail, " \
				"errno: %d, error info: %s", __LINE__, \
				full_filename, new_full_filename, \
				errno, strerror(errno));
			return 0;
		}

		if (*meta_filename != '\0')
		{
			sprintf(new_meta_filename, "%s"STORAGE_META_FILE_EXT, \
					new_full_filename);
			if (rename(meta_filename, new_meta_filename) != 0)
			{
				logError("file: "__FILE__", line: %d, " \
					"rename %s to %s fail, " \
					"errno: %d, error info: %s", __LINE__,\
					meta_filename, new_meta_filename, \
					errno, strerror(errno));

				unlink(new_full_filename);
				*filename = '\0';
				*filename_len = 0;
				return errno != 0 ? errno : EPERM;
			}
		}

		*filename_len = new_filename_len;
		memcpy(filename, new_filename, new_filename_len+1);
	}

	*filename_len=sprintf(full_filename, "%c"STORAGE_DATA_DIR_FORMAT"/%s", \
			STORAGE_STORE_PATH_PREFIX_CHAR, \
			store_path_index, filename);
	memcpy(filename, full_filename, (*filename_len) + 1);

	return 0;
}

static int storage_do_set_metadata(StorageClientInfo *pClientInfo, \
		const char *full_meta_filename, char *meta_buff, \
		const int meta_bytes, const char op_flag, char *sync_flag)
{
	FDFSMetaData *old_meta_list;
	FDFSMetaData *new_meta_list;
	FDFSMetaData *all_meta_list;
	FDFSMetaData *pOldMeta;
	FDFSMetaData *pNewMeta;
	FDFSMetaData *pAllMeta;
	FDFSMetaData *pOldMetaEnd;
	FDFSMetaData *pNewMetaEnd;
	char *file_buff;
	char *all_meta_buff;
	int64_t file_bytes;
	int old_meta_count;
	int new_meta_count;
	int all_meta_bytes;
	int result;

	*sync_flag = '\0';
	if (op_flag == STORAGE_SET_METADATA_FLAG_OVERWRITE)
	{
		if (meta_bytes == 0)
		{
			if (!fileExists(full_meta_filename))
			{
				return 0;
			}

			if (unlink(full_meta_filename) != 0)
			{
				logError("file: "__FILE__", line: %d, " \
					"client ip: %s, " \
					"delete meta file %s fail," \
					"errno: %d, error info: %s", \
					__LINE__, \
					pClientInfo->ip_addr, \
					full_meta_filename, \
					errno, strerror(errno));
				return errno != 0 ? errno : EACCES;
			}

			*sync_flag = STORAGE_OP_TYPE_SOURCE_DELETE_FILE;
			return 0;
		}

		if ((result=storage_sort_metadata_buff(meta_buff, \
				meta_bytes)) != 0)
		{
			return result;
		}

		if (fileExists(full_meta_filename))
		{
			*sync_flag = STORAGE_OP_TYPE_SOURCE_UPDATE_FILE;
		}
		else
		{
			*sync_flag = STORAGE_OP_TYPE_SOURCE_CREATE_FILE;
		}
		return writeToFile(full_meta_filename, meta_buff, meta_bytes);
	}

	result = getFileContent(full_meta_filename, &file_buff, &file_bytes);
	if (result == ENOENT)
	{
		if (meta_bytes == 0)
		{
			return 0;
		}

		if ((result=storage_sort_metadata_buff(meta_buff, \
				meta_bytes)) != 0)
		{
			return result;
		}

		*sync_flag = STORAGE_OP_TYPE_SOURCE_CREATE_FILE;
		return writeToFile(full_meta_filename, meta_buff, meta_bytes);
	}
	else if (result != 0)
	{
		return result;
	}

	old_meta_list = fdfs_split_metadata(file_buff, &old_meta_count, &result);
	if (old_meta_list == NULL)
	{
		free(file_buff);
		return result;
	}

	new_meta_list = fdfs_split_metadata(meta_buff, &new_meta_count, &result);
	if (new_meta_list == NULL)
	{
		free(file_buff);
		free(old_meta_list);
		return result;
	}

	all_meta_list = (FDFSMetaData *)malloc(sizeof(FDFSMetaData) * \
				(old_meta_count + new_meta_count));
	if (all_meta_list == NULL)
	{
		free(file_buff);
		free(old_meta_list);
		free(new_meta_list);

		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail", __LINE__, sizeof(FDFSMetaData) \
			 * (old_meta_count + new_meta_count));
		return ENOMEM;
	}

	qsort((void *)new_meta_list, new_meta_count, sizeof(FDFSMetaData), \
		metadata_cmp_by_name);

	pOldMetaEnd = old_meta_list + old_meta_count;
	pNewMetaEnd = new_meta_list + new_meta_count;
	pOldMeta = old_meta_list;
	pNewMeta = new_meta_list;
	pAllMeta = all_meta_list;
	while (pOldMeta < pOldMetaEnd && pNewMeta < pNewMetaEnd)
	{
		result = strcmp(pOldMeta->name, pNewMeta->name);
		if (result < 0)
		{
			memcpy(pAllMeta, pOldMeta, sizeof(FDFSMetaData));
			pOldMeta++;
		}
		else if (result == 0)
		{
			memcpy(pAllMeta, pNewMeta, sizeof(FDFSMetaData));
			pOldMeta++;
			pNewMeta++;
		}
		else  //result > 0
		{
			memcpy(pAllMeta, pNewMeta, sizeof(FDFSMetaData));
			pNewMeta++;
		}

		pAllMeta++;
	}