DFS增加批量上传文件功能（服务器端C源代码）

话说使用dfs在生产环境中还算稳定，也没出什么大的乱子。鉴于我们业务的需要，前段时间准备开始自己增加一个批量上传文件的功能。后联系到dfs的作者，和他聊了一下，在他的帮助下，完成了批量文件上传功能。因为dfs是开源的，和作者沟通后，把代码放出来，供大家各取所需。

因为本人以前一直是ms的忠实用户，而且是使用net的为生计的。所以在linux上开发还是黄花闺女上花轿-头一回。所以在某些不足的地方还请大家谅解。话说初学者的我因为dfs的关系还拜师了，不管他答不答应，反正我是叫师傅了,以后我就正式师从happyfish100。（希望他不要嫌我太笨）。

讲了那么多废话，我就把代码拿出来。

/**
* modify by seapeak.xu
* 2010-01-31 pm
* add deal files for betching upload files.
*/
static int storage_deal_file_for_betch(
StorageClientInfo *pClientInfo,
const int64_t fileBufferSize,
GroupArray *pGroupArray,
const int store_path_index,
const char *file_ext_name,
char *filename,
int *filename_len,
int *create_flag)
{
int result;
char full_filename[MAX_PATH_SIZE + 64];
unsigned int file_hash_codes[4];
int i;
do
{
logDebug("begin into deal file function.");
char fileExtName[FDFS_FILE_EXT_NAME_MAX_LEN + 2];
int ext_name_len = strlen(file_ext_name);
if (ext_name_len == 0)
{
logInfo("file is not having ext name.");
result = 1;
break;
}

if(ext_name_len > FDFS_FILE_EXT_NAME_MAX_LEN)
{
logInfo("file ext name size is too long,size is :%d.",ext_name_len);
result = 1;
break;
}

logDebug("begin deal ext name '.'.");
if(file_ext_name[0] == '.')
{
memcpy(fileExtName,file_ext_name,ext_name_len);
}
else
{
fileExtName[0] = '.';
memcpy(fileExtName+1,file_ext_name,ext_name_len);
}
fileExtName[1+ext_name_len] = '\0';

logDebug("files ext name is %s.",fileExtName);

logDebug("begin anaylze file full name.");
char *pStorePath = g_store_paths[store_path_index];
time_t start_time = time(NULL);
for (i = 0; i < 10; i++) {
if ((result = storage_gen_filename(pClientInfo, fileBufferSize,
fileExtName, FDFS_FILE_EXT_NAME_MAX_LEN + 1, start_time
+ FILE_TIMESTAMP_ADVANCED_SECS, filename, filename_len))
!= 0) {
logDebug("deal file full name is error.steps is %d.",i);
return result;
}

sprintf(full_filename, "%s/data/%s", pStorePath, filename);
if (!fileExists(full_filename)) {
break;
}
*full_filename = '\0';
}
logDebug("file full name is %s.",full_filename);
logDebug("analyze file full name is over.");

logDebug("begin save file.");
logDebug("file buffer size is %d.",fileBufferSize);
if ((result =tcprecvfile_ex(pClientInfo->sock,full_filename, fileBufferSize,
g_fsync_after_written_bytes,file_hash_codes,g_network_timeout)) != 0)
{
logDebug("save file fail.");
*filename = '\0';
*filename_len = 0;
return result;
}
logDebug("save file success.");
logDebug("save file over.");

logDebug("begin write sync file.");

logDebug("begin analyze sync file name.");
char src_filename[MAX_PATH_SIZE];
int length = sprintf(src_filename, "%c"
STORAGE_DATA_DIR_FORMAT"/%s", STORAGE_STORE_PATH_PREFIX_CHAR,
store_path_index, filename);
logDebug("sync file name length is %d.",length);
logDebug("sync file name is %s.",src_filename);
logDebug("begin write sync file name  into sync file.");
result = storage_binlog_write(time(NULL),
STORAGE_OP_TYPE_SOURCE_CREATE_FILE, src_filename);
logDebug("write into sync file is over.state is %d.",result);
logDebug("go out from deal function.");
if (result != 0)
{
logDebug("write sync file is error.error number is %d.",result);
logDebug("go out from deal function.");
break;
}
}while(0);
return result;

}

/**
* 2010-01-27pm
* modify by seapeak.xu
* batch upload files
*/
static int storage_batch_upload_file(StorageClientInfo *pClientInfo,
GroupArray *pGroupArray, const int64_t nInPackLen, int *create_flag,int64_t *filesCount)
{
TrackerHeader resp;
memset(&resp,0,sizeof(resp));
int filesNameLength[*filesCount];
int i;
int out_len;
int result;

//	//header size + files name path size(every file is 8 byte) + files name path buffer size
//	char out_buff[sizeof(resp) + FDFS_PROTO_PKG_LEN_SIZE * *filesCount + *filesCount * 128 ];
do
{
logDebug("go in the batch upload files function");
logDebug("pkg length is %d.",nInPackLen);
if(nInPackLen < 1 + 2 * FDFS_PROTO_PKG_LEN_SIZE)//must have storeindex(1 byte),header buffer,and so on
{
logError("file: "__FILE__", line: %d, "
"cmd=%d, client ip: %s, package size "
INT64_PRINTF_FORMAT" is not correct, "
"expect length > %d.in the batch upload files.",
__LINE__, STORAGE_PROTO_CMD_UPLOAD_FILE,
pClientInfo->ip_addr, nInPackLen,
1 + 2 * FDFS_PROTO_PKG_LEN_SIZE);
resp.status = EINVAL;
break;
}

logDebug("begin recv file count data.");
char fileCountBuffer[FDFS_PROTO_PKG_LEN_SIZE];

if ((resp.status = tcprecvdata_nb(pClientInfo->sock, fileCountBuffer,
FDFS_PROTO_PKG_LEN_SIZE,g_network_timeout)) != 0)
{
logError("file: "__FILE__", line: %d, "
"client ip:%s, recv files count buffer fail, "
"errno: %d, error info: %s. in the batch upload files.", __LINE__, pClientInfo->ip_addr,
resp.status, strerror(resp.status));
break;
}

logDebug("recv file count data over.");
*filesCount = buff2long(fileCountBuffer);
logDebug("file count is %d.",*filesCount);

logDebug("begin recv header data.");
char headerBuffer[1 + FDFS_PROTO_PKG_LEN_SIZE * (*filesCount + 2)];
if ((resp.status = tcprecvdata_nb(pClientInfo->sock, headerBuffer,
1 + FDFS_PROTO_PKG_LEN_SIZE * (*filesCount + 2),
g_network_timeout)) != 0) {
logError("file: "__FILE__", line: %d, "
"client ip:%s, recv header buffer fail, "
"errno: %d, error info: %s. in the batch upload files.", __LINE__, pClientInfo->ip_addr,
resp.status, strerror(resp.status));
break;
}
logInfo("recv header data over.");

int store_path_index = *headerBuffer;
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;
}

logInfo("get store path index over. store path index is %d",store_path_index);

logDebug("begin analyze header data for every file size.");
int64_t fileBufferSize[*filesCount];
int64_t filesTotalBufferSize;
int64_t filesTotalExtNameBufferSize;

int64_t tempFilesTotalBufferSize = 0;
for(i = 0; i<*filesCount;i++)
{
fileBufferSize[i] = buff2long(headerBuffer + 1 +(i * FDFS_PROTO_PKG_LEN_SIZE));
tempFilesTotalBufferSize += fileBufferSize[i];
logDebug("%dth files size is %d.",i+1,fileBufferSize[i]);
}
logDebug("begin analyze file ext name buffer size.");
filesTotalExtNameBufferSize = buff2long(headerBuffer + 1 + *filesCount * FDFS_PROTO_PKG_LEN_SIZE);
logDebug("file total ext name size is %d.",filesTotalExtNameBufferSize);
logDebug("begin analyze files total buffer size.");
filesTotalBufferSize = buff2long(headerBuffer + 1 + (*filesCount + 1)*FDFS_PROTO_PKG_LEN_SIZE);
logDebug("files total buffer size is %d.",filesTotalBufferSize);

if(tempFilesTotalBufferSize != filesTotalBufferSize)
{
logError("file: "__FILE__", line: %d, "
"client ip: %s, files total buffer size is not same.temp size is: %d "
"buffer size is:%d.", __LINE__, pClientInfo->ip_addr,
tempFilesTotalBufferSize,filesTotalBufferSize);
resp.status = EINVAL;
break;
}
if(*filesCount * FDFS_FILE_EXT_NAME_MAX_LEN != filesTotalExtNameBufferSize)
{
logError("file: "__FILE__", line: %d, "
"client ip: %s, files ext name buffer size is not same."
"buffer size is:%d.but real fileFDFS_PROTO_PKG_LEN_SIZE ext name buffer size is %d.",
__LINE__, pClientInfo->ip_addr,
filesTotalExtNameBufferSize,
*filesCount * FDFS_PROTO_PKG_LEN_SIZE);
resp.status = EINVAL;
break;
}

char file_ext_name[*filesCount][FDFS_FILE_EXT_NAME_MAX_LEN+1];
char ext_name_buffer[filesTotalExtNameBufferSize];
logDebug("begin recv ext name buffer.");
if ((resp.status = tcprecvdata_nb(pClientInfo->sock, ext_name_buffer,
filesTotalExtNameBufferSize,g_network_timeout)) != 0)
{
logError("file: "__FILE__", line: %d, "
"client ip:%s, recv extName buffer fail, "
"errno: %d, error info: %s. in the batch upload files.", __LINE__, pClientInfo->ip_addr,
resp.status, strerror(resp.status));
break;
}
logDebug("begin analyze files ext name buffer.");
for(i = 0;i<*filesCount;i++)
{
memcpy(file_ext_name+i,ext_name_buffer+i*FDFS_FILE_EXT_NAME_MAX_LEN,FDFS_FILE_EXT_NAME_MAX_LEN);
file_ext_name[i][FDFS_FILE_EXT_NAME_MAX_LEN] = '\0';
logDebug("%dth file's ext name is %s.",i+1,file_ext_name[i]);
}

logDebug("recv ext name buffer over.");

int status = 0;
logDebug("begin save files.");
char fileName[128];
char out_buffer[FDFS_PROTO_PKG_LEN_SIZE+128];
for(i = 0;i<*filesCount;i++)
{
//everytime ,first step is clean filename.
memset(fileName,0,128);
memset(out_buffer,0,FDFS_PROTO_PKG_LEN_SIZE+128);
status = storage_deal_file_for_betch(pClientInfo,fileBufferSize[i],
pGroupArray,store_path_index,file_ext_name+i,
fileName, filesNameLength+i ,create_flag);

long2buff(filesNameLength[i],out_buffer);
memcpy(out_buffer+FDFS_PROTO_PKG_LEN_SIZE,fileName,128);

logDebug("!!!!! %dth file name is %s.",i+1,fileName);
if (status != 0 || (*create_flag & STORAGE_CREATE_FLAG_LINK)) {
logDebug(" %dth file save fail.in up",i+1);
break;
}
if ((result = tcpsenddata_nb(pClientInfo->sock, out_buffer,
FDFS_PROTO_PKG_LEN_SIZE+128, 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));
logError("upload file index is %d.",i+1);
break;
}
}
logDebug("go out deal upload.");

if(0 != status || 0 != result) break;
logDebug("save file over.");
logDebug("upload and save files over.");
}while(0);

logDebug("begin send data to client.");

resp.cmd = STORAGE_PROTO_CMD_RESP;

char out_buff[sizeof(resp)];
memset(out_buff,0,sizeof(resp));
if (resp.status == 0)
{
//body is files name size and files name buffer
out_len = FDFS_PROTO_PKG_LEN_SIZE * *filesCount + *filesCount * 128;
long2buff(out_len, resp.pkg_len);
memcpy(out_buff, &resp, sizeof(resp));
}
else
{
logDebug("error. header state is 0.");

out_len = 0;
long2buff(out_len, resp.pkg_len);
memcpy(out_buff, &resp, sizeof(resp));
}

logDebug("begin send to client buffer.");

if ((result = tcpsenddata_nb(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;
}

logDebug("send buffer is over.result number is %d.",result);
return resp.status;
}

case STORAGE_PROTO_CMD_BATCH_UPLOAD_FILE:
{
int64_t filesCount;
if((result = storage_batch_upload_file(&client_info, &group_array,
nInPackLen, &create_flag,&filesCount)) != 0)
{
pthread_mutex_lock(&stat_count_thread_lock);
if (create_flag & STORAGE_CREATE_FLAG_FILE)
{
g_storage_stat.total_upload_count += filesCount;
}
pthread_mutex_unlock(&stat_count_thread_lock);
break;
}

if (create_flag & STORAGE_CREATE_FLAG_FILE) {
CHECK_AND_WRITE_TO_STAT_FILE3(
g_storage_stat.total_upload_count,
g_storage_stat.success_upload_count,
g_storage_stat.last_source_update)
}
break;
}

对于代码先说明一下，上面一段有2个方法，这两个方法是提供了服务器文件的批量上传功能。下面一个case的代码段是需要在void* storage_thread_entrance(void* arg)方法中加入进去的，位置在switch (header.cmd) 段的case中，除了这些我们还要自定义一个服务器和客户端的批量上传协议命令码，也就是header.cmd的值，我个人是遵循dfs的原始代码原则，在tracker_proto.h文件中加入了一段命令码定义，如下：

/**
* 2010-01-17
* modify by Seapeak.Xu
* Add batch upload files to stroages
*/
#define STORAGE_PROTO_CMD_BATCH_UPLOAD_FILE 126

这样，就可以完成文件的批量上传了，当然了，客户端还是要自己写的。

对于为什么要这么写dfs的批量上传服务器端代码，我会看情况在日后的blog文章中详细说明。