tracker_mem.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/stat.h>
#include <sys/socket.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 <fcntl.h>
#include <pthread.h>
#include "fdfs_define.h"
#include "logger.h"
#include "fdfs_global.h"
#include "tracker_global.h"
#include "tracker_mem.h"
#include "shared_func.h"

static pthread_mutex_t mem_thread_lock;

#define STORAGE_GROUPS_LIST_FILENAME	"storage_groups.dat"
#define STORAGE_SERVERS_LIST_FILENAME	"storage_servers.dat"
#define STORAGE_SYNC_TIMESTAMP_FILENAME	"storage_sync_timestamp.dat"
#define STORAGE_DATA_FIELD_SEPERATOR	','

#define TRACKER_MEM_ALLOC_ONCE	2

static int tracker_malloc_storage_path_mbs(FDFSStorageDetail *pStorage, \
		const int store_path_count)
{
	int alloc_bytes;

	if (store_path_count <= 0)
	{
		return 0;
	}

	alloc_bytes = sizeof(int64_t) * store_path_count;

	pStorage->path_total_mbs = (int64_t *)malloc(alloc_bytes);
	if (pStorage->path_total_mbs == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail, " \
			"errno: %d, error info: %s", \
			__LINE__, alloc_bytes, errno, strerror(errno));
		return errno != 0 ? errno : ENOMEM;
	}

	pStorage->path_free_mbs = (int64_t *)malloc(alloc_bytes);
	if (pStorage->path_free_mbs == NULL)
	{
		free(pStorage->path_total_mbs);
		pStorage->path_total_mbs = NULL;

		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail, " \
			"errno: %d, error info: %s", \
			__LINE__, alloc_bytes, errno, strerror(errno));
		return errno != 0 ? errno : ENOMEM;
	}

	memset(pStorage->path_total_mbs, 0, alloc_bytes);
	memset(pStorage->path_free_mbs, 0, alloc_bytes);

	return 0;
}

static int tracker_realloc_storage_path_mbs(FDFSStorageDetail *pStorage, \
		const int old_store_path_count, const int new_store_path_count)
{
	int alloc_bytes;
	int copy_bytes;
	int64_t *new_path_total_mbs;
	int64_t *new_path_free_mbs;

	if (new_store_path_count <= 0)
	{
		return EINVAL;
	}

	alloc_bytes = sizeof(int64_t) * new_store_path_count;

	new_path_total_mbs = (int64_t *)malloc(alloc_bytes);
	if (new_path_total_mbs == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail, " \
			"errno: %d, error info: %s", \
			__LINE__, alloc_bytes, errno, strerror(errno));
		return errno != 0 ? errno : ENOMEM;
	}

	new_path_free_mbs = (int64_t *)malloc(alloc_bytes);
	if (new_path_free_mbs == NULL)
	{
		free(new_path_total_mbs);

		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail, " \
			"errno: %d, error info: %s", \
			__LINE__, alloc_bytes, errno, strerror(errno));
		return errno != 0 ? errno : ENOMEM;
	}

	memset(new_path_total_mbs, 0, alloc_bytes);
	memset(new_path_free_mbs, 0, alloc_bytes);

	if (old_store_path_count == 0)
	{
		pStorage->path_total_mbs = new_path_total_mbs;
		pStorage->path_free_mbs = new_path_free_mbs;

		return 0;
	}

	copy_bytes = (old_store_path_count < new_store_path_count ? \
		old_store_path_count : new_store_path_count) * sizeof(int64_t);
	memcpy(new_path_total_mbs, pStorage->path_total_mbs, copy_bytes);
	memcpy(new_path_free_mbs, pStorage->path_free_mbs, copy_bytes);

	free(pStorage->path_total_mbs);
	free(pStorage->path_free_mbs);

	pStorage->path_total_mbs = new_path_total_mbs;
	pStorage->path_free_mbs = new_path_free_mbs;

	return 0;
}

static int tracker_realloc_group_path_mbs(FDFSGroupInfo *pGroup, \
		const int new_store_path_count)
{
	FDFSStorageDetail *pStorage;
	FDFSStorageDetail *pEnd;
	int result;

	pEnd = pGroup->all_servers + pGroup->alloc_size;
	for (pStorage=pGroup->all_servers; pStorage<pEnd; pStorage++)
	{
		if ((result=tracker_realloc_storage_path_mbs(pStorage, \
			pGroup->store_path_count, new_store_path_count)) != 0)
		{
			return result;
		}
	}

	pGroup->store_path_count = new_store_path_count;

	return 0;
}

static int tracker_malloc_group_path_mbs(FDFSGroupInfo *pGroup)
{
	FDFSStorageDetail *pStorage;
	FDFSStorageDetail *pEnd;
	int result;

	pEnd = pGroup->all_servers + pGroup->alloc_size;
	for (pStorage=pGroup->all_servers; pStorage<pEnd; pStorage++)
	{
		if ((result=tracker_malloc_storage_path_mbs(pStorage, \
				pGroup->store_path_count)) != 0)
		{
			return result;
		}
	}

	return 0;
}

static int tracker_malloc_all_group_path_mbs()
{
	FDFSGroupInfo *pGroup;
	FDFSGroupInfo *pEnd;
	int result;

	pEnd = g_groups.groups + g_groups.alloc_size;
	for (pGroup=g_groups.groups; pGroup<pEnd; pGroup++)
	{
		if (pGroup->store_path_count == 0)
		{
			continue;
		}

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

	return 0;
}

static int tracker_load_groups(const char *data_path)
{
#define STORAGE_DATA_GROUP_FIELDS	4

	FILE *fp;
	char szLine[256];
	char *fields[STORAGE_DATA_GROUP_FIELDS];
	int result;
	int col_count;
	TrackerClientInfo clientInfo;
	bool bInserted;

	if ((fp=fopen(STORAGE_GROUPS_LIST_FILENAME, "r")) == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"open file \"%s/%s\" fail, " \
			"errno: %d, error info: %s", \
			__LINE__, data_path, STORAGE_GROUPS_LIST_FILENAME, \
			errno, strerror(errno));
		return errno != 0 ? errno : ENOENT;
	}

	result = 0;
	while (fgets(szLine, sizeof(szLine), fp) != NULL)
	{
		if (*szLine == '\0')
		{
			continue;
		}

		col_count = splitEx(szLine, STORAGE_DATA_FIELD_SEPERATOR, \
			fields, STORAGE_DATA_GROUP_FIELDS);
		if (col_count != STORAGE_DATA_GROUP_FIELDS && \
			col_count != STORAGE_DATA_GROUP_FIELDS - 2)
		{
			logError("file: "__FILE__", line: %d, " \
				"the format of the file \"%s/%s\" is invalid", \
				__LINE__, data_path, \
				STORAGE_GROUPS_LIST_FILENAME);
			result = errno != 0 ? errno : EINVAL;
			break;
		}
	
		memset(&clientInfo, 0, sizeof(TrackerClientInfo));
		snprintf(clientInfo.group_name, sizeof(clientInfo.group_name),\
				"%s", trim(fields[0]));
		if ((result=tracker_mem_add_group(&clientInfo, \
				false, &bInserted)) != 0)
		{
			break;
		}

		if (!bInserted)
		{
			logError("file: "__FILE__", line: %d, " \
				"in the file \"%s/%s\", " \
				"group \"%s\" is duplicate", \
				__LINE__, data_path, \
				STORAGE_GROUPS_LIST_FILENAME, \
				clientInfo.group_name);
			result = errno != 0 ? errno : EEXIST;
			break;
		}

		clientInfo.pGroup->storage_port = atoi(trim(fields[1]));
		if (col_count == STORAGE_DATA_GROUP_FIELDS - 2)
		{  //version < V1.12
			clientInfo.pGroup->store_path_count = 0;
			clientInfo.pGroup->subdir_count_per_path = 0;
		}
		else
		{
			clientInfo.pGroup->store_path_count = \
				atoi(trim(fields[2]));
			clientInfo.pGroup->subdir_count_per_path = \
				atoi(trim(fields[3]));
		}
	}

	fclose(fp);
	return result;
}

static int tracker_locate_storage_sync_server(FDFSStorageSync *pStorageSyncs, \
		const int nStorageSyncCount, const bool bLoadFromFile)
{
	FDFSStorageSync *pSyncServer;
	FDFSStorageSync *pSyncEnd;

	pSyncEnd = pStorageSyncs + nStorageSyncCount;
	for (pSyncServer=pStorageSyncs; pSyncServer<pSyncEnd; pSyncServer++)
	{
		pSyncServer->pStorage->psync_src_server = \
			tracker_mem_get_storage(pSyncServer->pGroup, \
			pSyncServer->sync_src_ip_addr);
		if (pSyncServer->pStorage->psync_src_server == NULL)
		{
			char buff[MAX_PATH_SIZE+64];
			if (bLoadFromFile)
			{
				snprintf(buff, sizeof(buff), \
					"in the file \"%s/data/%s\", ", \
					g_base_path, \
					STORAGE_SERVERS_LIST_FILENAME);
			}
			else
			{
				buff[0] = '\0';
			}

			logError("file: "__FILE__", line: %d, " \
				"%sgroup_name: %s, storage server \"%s:%d\" " \
				"does not exist", \
				__LINE__, buff, \
				pSyncServer->pGroup->group_name, \
				pSyncServer->sync_src_ip_addr, \
				pSyncServer->pGroup->storage_port);

			return ENOENT;
		}
	}

	return 0;
}

static int tracker_load_storages(const char *data_path)
{
#define STORAGE_DATA_SERVER_FIELDS	17

	FILE *fp;
	char szLine[256];
	char *fields[STORAGE_DATA_SERVER_FIELDS];
	char *psync_src_ip_addr;
	FDFSStorageSync *pStorageSyncs;
	int nStorageSyncSize;
	int nStorageSyncCount;
	int cols;
	int result;
	TrackerClientInfo clientInfo;
	bool bInserted;

	if ((fp=fopen(STORAGE_SERVERS_LIST_FILENAME, "r")) == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"open file \"%s/%s\" fail, " \
			"errno: %d, error info: %s", \
			__LINE__, data_path, STORAGE_SERVERS_LIST_FILENAME, \
			errno, strerror(errno));
		return errno != 0 ? errno : ENOENT;
	}

	nStorageSyncSize = 0;
	nStorageSyncCount = 0;
	pStorageSyncs = NULL;
	result = 0;
	while (fgets(szLine, sizeof(szLine), fp) != NULL)
	{
		if (*szLine == '\0')
		{
			continue;
		}

		if ((cols=splitEx(szLine, STORAGE_DATA_FIELD_SEPERATOR, \
			fields, STORAGE_DATA_SERVER_FIELDS)) != \
				STORAGE_DATA_SERVER_FIELDS)
		{
			logError("file: "__FILE__", line: %d, " \
				"the format of the file \"%s/%s\" is invalid" \
				", colums: %d != expect colums: %d", \
				__LINE__, data_path, \
				STORAGE_SERVERS_LIST_FILENAME, \
				cols, STORAGE_DATA_SERVER_FIELDS);
			result = errno != 0 ? errno : EINVAL;
			break;
		}
	
		memset(&clientInfo, 0, sizeof(TrackerClientInfo));
		snprintf(clientInfo.group_name, sizeof(clientInfo.group_name),\
				"%s", trim(fields[0]));
		snprintf(clientInfo.ip_addr, sizeof(clientInfo.ip_addr),\
				"%s", trim(fields[1]));
		if ((clientInfo.pGroup=tracker_mem_get_group( \
				clientInfo.group_name)) == NULL)
		{
			logError("file: "__FILE__", line: %d, " \
				"in the file \"%s/%s\", " \
				"group \"%s\" is not found", \
				__LINE__, data_path, \
				STORAGE_SERVERS_LIST_FILENAME, \
				clientInfo.group_name);
			result = errno != 0 ? errno : ENOENT;
			break;
		}

		if ((result=tracker_mem_add_storage(&clientInfo, \
				false, &bInserted)) != 0)
		{
			break;
		}

		if (!bInserted)
		{
			logError("file: "__FILE__", line: %d, " \
				"in the file \"%s/%s\", " \
				"storage \"%s\" is duplicate", \
				__LINE__, data_path, \
				STORAGE_SERVERS_LIST_FILENAME, \
				clientInfo.ip_addr);
			result = errno != 0 ? errno : EEXIST;
			break;
		}
		
		clientInfo.pStorage->status = atoi(trim_left(fields[2]));
		if (!((clientInfo.pStorage->status == \
				FDFS_STORAGE_STATUS_WAIT_SYNC) || \
			(clientInfo.pStorage->status == \
				FDFS_STORAGE_STATUS_SYNCING) || \
			(clientInfo.pStorage->status == \
				FDFS_STORAGE_STATUS_INIT)))
		{
			clientInfo.pStorage->status = \
				FDFS_STORAGE_STATUS_OFFLINE;
		}

		psync_src_ip_addr = trim(fields[3]);
		clientInfo.pStorage->sync_until_timestamp = atoi( \
					trim_left(fields[4]));
		clientInfo.pStorage->stat.total_upload_count = strtoll( \
					trim_left(fields[5]), NULL, 10);
		clientInfo.pStorage->stat.success_upload_count = strtoll( \
					trim_left(fields[6]), NULL, 10);
		clientInfo.pStorage->stat.total_set_meta_count = strtoll( \
					trim_left(fields[7]), NULL, 10);
		clientInfo.pStorage->stat.success_set_meta_count = strtoll( \
					trim_left(fields[8]), NULL, 10);
		clientInfo.pStorage->stat.total_delete_count = strtoll( \
					trim_left(fields[9]), NULL, 10);
		clientInfo.pStorage->stat.success_delete_count = strtoll( \
					trim_left(fields[10]), NULL, 10);
		clientInfo.pStorage->stat.total_download_count = strtoll( \
					trim_left(fields[11]), NULL, 10);
		clientInfo.pStorage->stat.success_download_count = strtoll( \
					trim_left(fields[12]), NULL, 10);
		clientInfo.pStorage->stat.total_get_meta_count = strtoll( \
					trim_left(fields[13]), NULL, 10);
		clientInfo.pStorage->stat.success_get_meta_count = strtoll( \
					trim_left(fields[14]), NULL, 10);
		clientInfo.pStorage->stat.last_source_update = atoi( \
					trim_left(fields[15]));
		clientInfo.pStorage->stat.last_sync_update = atoi( \
					trim_left(fields[16]));
		if (*psync_src_ip_addr == '\0')
		{
			continue;
		}

		if (nStorageSyncSize <= nStorageSyncCount)
		{
			nStorageSyncSize += 8;
			pStorageSyncs = (FDFSStorageSync *)realloc( \
				pStorageSyncs, \
				sizeof(FDFSStorageSync) * nStorageSyncSize);
			if (pStorageSyncs == NULL)
			{
				result = errno != 0 ? errno : ENOMEM;
				logError("file: "__FILE__", line: %d, " \
					"realloc %d bytes fail", __LINE__, \
					sizeof(FDFSStorageSync) * \
					nStorageSyncSize);
				break;
			}
		}

		pStorageSyncs[nStorageSyncCount].pGroup = clientInfo.pGroup;
		pStorageSyncs[nStorageSyncCount].pStorage = clientInfo.pStorage;
		snprintf(pStorageSyncs[nStorageSyncCount].sync_src_ip_addr, \
			IP_ADDRESS_SIZE, "%s", psync_src_ip_addr);
		nStorageSyncCount++;

	}

	fclose(fp);

	if (pStorageSyncs == NULL)
	{
		return result;
	}

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

	result = tracker_locate_storage_sync_server(pStorageSyncs, \
			nStorageSyncCount, true);
	free(pStorageSyncs);
	return result;
}

static int tracker_load_sync_timestamps(const char *data_path)
{
#define STORAGE_SYNC_TIME_MAX_FIELDS	2 + FDFS_MAX_SERVERS_EACH_GROUP

	FILE *fp;
	char szLine[512];
	char group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
	char previous_group_name[FDFS_GROUP_NAME_MAX_LEN + 1];
	char src_ip_addr[IP_ADDRESS_SIZE];
	char *fields[STORAGE_SYNC_TIME_MAX_FIELDS];
	FDFSGroupInfo *pGroup;
	FDFSGroupInfo *pEnd;
	int cols;
	int src_index;
	int dest_index;
	int curr_synced_timestamp;
	int result;

	if (!fileExists(STORAGE_SYNC_TIMESTAMP_FILENAME))
	{
		return 0;
	}

	if ((fp=fopen(STORAGE_SYNC_TIMESTAMP_FILENAME, "r")) == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"open file \"%s/%s\" fail, " \
			"errno: %d, error info: %s", \
			__LINE__, data_path, STORAGE_SYNC_TIMESTAMP_FILENAME, \
			errno, strerror(errno));
		return errno != 0 ? errno : ENOENT;
	}

	src_index = 0;
	pGroup = NULL;
	*previous_group_name = '\0';
	result = 0;
	while (fgets(szLine, sizeof(szLine), fp) != NULL)
	{
		if (*szLine == '\0' || *szLine == '\n')
		{
			continue;
		}

		if ((cols=splitEx(szLine, STORAGE_DATA_FIELD_SEPERATOR, \
			fields, STORAGE_SYNC_TIME_MAX_FIELDS)) <= 2)
		{
			logError("file: "__FILE__", line: %d, " \
				"the format of the file \"%s/%s\" is invalid" \
				", colums: %d <= 2", \
				__LINE__, data_path, \
				STORAGE_SYNC_TIMESTAMP_FILENAME, cols);
			result = errno != 0 ? errno : EINVAL;
			break;
		}
	
		snprintf(group_name, sizeof(group_name), \
				"%s", trim(fields[0]));
		snprintf(src_ip_addr, sizeof(src_ip_addr), \
				"%s", trim(fields[1]));
		if (strcmp(group_name, previous_group_name) != 0 || \
			pGroup == NULL)
		{
			if ((pGroup=tracker_mem_get_group(group_name)) == NULL)
			{
				logError("file: "__FILE__", line: %d, " \
					"in the file \"%s/%s\", " \
					"group \"%s\" is not found", \
					__LINE__, data_path, \
					STORAGE_SYNC_TIMESTAMP_FILENAME, \
					group_name);
				result = errno != 0 ? errno : ENOENT;
				break;
			}

			strcpy(previous_group_name, group_name);
			src_index = 0;
		}
		
		if (src_index >= pGroup->count)
		{
			logError("file: "__FILE__", line: %d, " \
				"the format of the file \"%s/%s\" is invalid" \
				", group: %s, row count:%d > server count:%d",\
				__LINE__, data_path, \
				STORAGE_SYNC_TIMESTAMP_FILENAME, \
				group_name, src_index+1, pGroup->count);
			result = errno != 0 ? errno : EINVAL;
			break;
		}

		if (strcmp(pGroup->all_servers[src_index].ip_addr, \
			src_ip_addr) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"in data file: \"%s/%s\", " \
				"group: %s, src server ip: %s != %s",\
				__LINE__, data_path, \
				STORAGE_SYNC_TIMESTAMP_FILENAME, \
				group_name, src_ip_addr, \
				pGroup->all_servers[src_index].ip_addr);
			result = errno != 0 ? errno : EINVAL;
			break;
		}

		if (cols > pGroup->count + 2)
		{
			logError("file: "__FILE__", line: %d, " \
				"the format of the file \"%s/%s\" is invalid" \
				", group_name: %s, colums: %d > %d", \