FastDFS分布式文件系统点滴记录5 -- upload上传机制剖析3

有了tracker 分析的基础，我们直接进入storage 的任务处理函数 int storage_deal_task(struct fast_task_info *pTask)；

storage_service.c 6473行：

case STORAGE_PROTO_CMD_UPLOAD_FILE:

result = storage_upload_file(pTask,
false);

break;

如果命令是上传文件，会调用storage_upload_file函数处理，我们进入这个函数。

storage_service.c 3735行，是storage_upload_file 的入口，

storage_service.c 3771行 ：

p = pTask->data
+ sizeof(TrackerHeader);

store_path_index =
*p++;

if (store_path_index
< 0 || store_path_index
>= g_fdfs_path_count)

{

logError("file: "__FILE__", line: %d, "
\

"client ip: %s, store_path_index: %d "
\

"is invalid", __LINE__,
\

pTask->client_ip, store_path_index);

pClientInfo->total_length
= sizeof(TrackerHeader);

return EINVAL;

}

file_bytes = buff2long(p);

p += FDFS_PROTO_PKG_LEN_SIZE;

if (file_bytes
< 0 || file_bytes
!= nInPackLen
- \

(1
+ FDFS_PROTO_PKG_LEN_SIZE +
\

FDFS_FILE_EXT_NAME_MAX_LEN))

{

logError("file: "__FILE__", line: %d, "
\

"client ip: %s, pkg length is not correct, "
\

"invalid file bytes: "INT64_PRINTF_FORMAT
\

", total body length: "INT64_PRINTF_FORMAT,
\

__LINE__, pTask->client_ip, file_bytes, nInPackLen);

pClientInfo->total_length
= sizeof(TrackerHeader);

return EINVAL;

}

memcpy(file_ext_name, p, FDFS_FILE_EXT_NAME_MAX_LEN);

*(file_ext_name
+ FDFS_FILE_EXT_NAME_MAX_LEN)
= '\0';

p += FDFS_FILE_EXT_NAME_MAX_LEN;

pFileContext->calc_crc32
= true;

pFileContext->calc_file_hash
= g_check_file_duplicate;

pFileContext->extra_info.upload.start_time
= time(NULL);

代码分析：
1. 首先解析出store_path_index；
2. 解析出file_bytes；
3. 解析出file_ext_name；
4. 将相关的属性赋值给pFileContext

storage_service.c 3810行 ：

pFileContext->extra_info.upload.file_type
= _FILE_TYPE_REGULAR;

pFileContext->sync_flag
= STORAGE_OP_TYPE_SOURCE_CREATE_FILE;

pFileContext->timestamp2log
= pFileContext->extra_info.upload.start_time;

pFileContext->op
= FDFS_STORAGE_FILE_OP_WRITE;

if (bAppenderFile)

{

pFileContext->extra_info.upload.file_type
|=
\

_FILE_TYPE_APPENDER;

}

else

{

if (g_if_use_trunk_file
&& trunk_check_size(
\

TRUNK_CALC_SIZE(file_bytes)))

{

pFileContext->extra_info.upload.file_type
|=
\

_FILE_TYPE_TRUNK;

}

}

注意，这里把pFileContext->op 置为 FDFS_STORAGE_FILE_OP_WRITE;说明要执行的是写操作。
为了简化分析，我们暂时不考虑trunk 这种方式，以后会单独详细分析。

接着往下执行，storage_service.c 3893行 ：

return storage_write_to_file(pTask, file_offset, file_bytes,
\

p - pTask->data, dio_write_file,
\

storage_upload_file_done_callback,
\

clean_func, store_path_index);

开始真正的写文件操作了。

我们进入storage_write_to_file，storage_service.c 5567行 ：

static int storage_write_to_file(struct fast_task_info
*pTask,
\

const int64_t file_offset,
const int64_t upload_bytes,
\

const
int buff_offset, TaskDealFunc deal_func,
\

FileDealDoneCallback done_callback,
\

DisconnectCleanFunc clean_func,
const int store_path_index)

{

StorageClientInfo *pClientInfo;

StorageFileContext *pFileContext;

int result;

pClientInfo =
(StorageClientInfo *)pTask->arg;

pFileContext =
&(pClientInfo->file_context);

pClientInfo->deal_func
= deal_func;

pClientInfo->clean_func
= clean_func;

pFileContext->fd
= -1;

pFileContext->buff_offset
= buff_offset;

pFileContext->offset
= file_offset;

pFileContext->start
= file_offset;

pFileContext->end
= file_offset + upload_bytes;

pFileContext->dio_thread_index
= storage_dio_get_thread_index(
\

pTask, store_path_index, pFileContext->op);

pFileContext->done_callback
= done_callback;

if (pFileContext->calc_crc32)

{

pFileContext->crc32
= CRC32_XINIT;

}

if (pFileContext->calc_file_hash)

{

INIT_HASH_CODES4(pFileContext->file_hash_codes)

}

if ((result=storage_dio_queue_push(pTask))
!= 0)

{

pClientInfo->total_length
= sizeof(TrackerHeader);

return result;

}

return STORAGE_STATUE_DEAL_FILE;

}

代码分析：
1. 注意参数的回调函数，这里含义是写动作执行完成后，会主动调用；
2.storage_dio_queue_push 函数，把pTask push 到dio 队列中；
3. 前面在分析storage 入口函数的时候，在main函数init 时，已经完成了对dio 线程的初始化动作；
4. 这里是往dio队列中push pTask；
5. storage_dio_queue_push 内部 使用的task_queue_push，同时，pthread_cond_signal 通知队列另端有数据到来。
6.特别需要注意 pClientInfo->deal_func = deal_func; 下面会用到。

既然已经放入队列，那么，另端就应该从队列中取出任务执行。这里，我们看一下dio线程执行函数。

storage_dio.c 646行：

static void *dio_thread_entrance(void* arg)

{

int result;

struct storage_dio_context *pContext;

struct fast_task_info *pTask;

pContext =
(struct storage_dio_context *)arg;

pthread_mutex_lock(&(pContext->lock));

while (g_continue_flag)

{

if ((result=pthread_cond_wait(&(pContext->cond),
\

&(pContext->lock)))
!= 0)

{

logError("file: "__FILE__", line: %d, "
\

"call pthread_cond_wait fail, "
\

"errno: %d, error info: %s",
\

__LINE__, result, STRERROR(result));

}

while
((pTask=task_queue_pop(&(pContext->queue)))
!=
NULL)

{

((StorageClientInfo
*)pTask->arg)->deal_func(pTask);

}

}

pthread_mutex_unlock(&(pContext->lock));

if ((result=pthread_mutex_lock(&g_dio_thread_lock))
!= 0)

{

logError("file: "__FILE__", line: %d, "
\

"call pthread_mutex_lock fail, "
\

"errno: %d, error info: %s",
\

__LINE__, result, STRERROR(result));

}

g_dio_thread_count--;

if ((result=pthread_mutex_unlock(&g_dio_thread_lock))
!= 0)

{

logError("file: "__FILE__", line: %d, "
\

"call pthread_mutex_lock fail, "
\

"errno: %d, error info: %s",
\

__LINE__, result, STRERROR(result));

}

logDebug("file: "__FILE__", line: %d, "
\

"dio thread exited, thread count: %d",
\

__LINE__, g_dio_thread_count);

return NULL;

}

代码分析：
1. 当队列有数据，取出数据，执行之；
2. ((StorageClientInfo *)pTask->arg)->deal_func(pTask); 这里通过回调函数完成任务的处理。

结合前面的分析，deal_func 函数指针，其实是函数 dio_write_file；
我们接着分析 dio_write_file，这个函数在 storage_dio.c 421行。

storage_dio.c 454行：

if (write(pFileContext->fd, pDataBuff,
write_bytes) != write_bytes)

{

result = errno
!= 0 ? errno
: EIO;

logError("file: "__FILE__", line: %d, "
\

"write to file: %s fail, fd=%d, write_bytes=%d, "
\

"errno: %d, error info: %s",
\

__LINE__, pFileContext->filename,
\

pFileContext->fd, write_bytes,
\

result, STRERROR(result));

}

调用write 写数据了。

接着分析， storage_dio.c 478行：

if (pFileContext->calc_crc32)

{

pFileContext->crc32
= CRC32_ex(pDataBuff, write_bytes,
\

pFileContext->crc32);

}

storage 通过crc32 保证数据的完整性。

接着分析， storage_dio.c 495行：

pFileContext->offset
+= write_bytes;

if (pFileContext->offset
< pFileContext->end)

{

pFileContext->buff_offset
= 0;

storage_nio_notify(pTask);
//notify nio
to deal

}

else

{

if (pFileContext->calc_crc32)

{

pFileContext->crc32
= CRC32_FINAL(
\

pFileContext->crc32);

}

if (pFileContext->calc_file_hash)

{

FINISH_HASH_CODES4(pFileContext->file_hash_codes)

}

if (pFileContext->extra_info.upload.before_close_callback
!=
NULL)

{

result = pFileContext->extra_info.upload.
\

before_close_callback(pTask);

}

/* file write done, close it
*/

close(pFileContext->fd);

pFileContext->fd
= -1;

if (pFileContext->done_callback
!=
NULL)

{

pFileContext->done_callback(pTask, result);

}

}

代码分析：
1. 如果文件数据长度没有接收完，会storage_nio_notify 继续从nio(网络io) 读取文件内容；
2. 写文件完成后，得到文件 crc32 的值；
3. 执行回调函数，pFileContext->done_callback(pTask, result);
4. 而这个done_callback 实际上是 storage_upload_file_done_callback。

接着分析storage_upload_file_done_callback， storage_dio.c 1004行：

static void storage_upload_file_done_callback(struct fast_task_info
*pTask,
\

const
int err_no)

接着分析， 先跳过trunk文件的处理，trunk 后面分析，storage_dio.c 1029行：

if (result
== 0)

{

result = storage_service_upload_file_done(pTask);

if (result
== 0)

{

if (pFileContext->create_flag
& STORAGE_CREATE_FLAG_FILE)

{

result = storage_binlog_write(\

pFileContext->timestamp2log,
\

STORAGE_OP_TYPE_SOURCE_CREATE_FILE,
\

pFileContext->fname2log);

}

}

}

代码分析：

1. 调用storage_service_upload_file_done，完成接收文件后的操作。目前暂时先不展开分析，因为这里涉及到slave、link、fastdht 等机制，后面专门详细介绍。
2. 文件上传结束后，会调用 storage_binlog_write 写入binlog。

接着往下分析， storage_dio.c 1085行：

pHeader =
(TrackerHeader *)pTask->data;

pHeader->status
= result;

pHeader->cmd
= STORAGE_PROTO_CMD_RESP;

long2buff(pClientInfo->total_length
- sizeof(TrackerHeader),
\

pHeader->pkg_len);

storage_nio_notify(pTask);

构造应答的数据包，storage_nio_notify发送至客户端。

至此，我们简要分析了storage 接收文件的处理流程。大概的调用脉络上还是清晰的。像slave、trunk、link、fastdht 等机制，比较复杂，后面慢慢分析。本章节主要解释的就是storage对文件上传的大概处理流程。

欢迎感兴趣的朋友一起交流研究，提出意见。
FastDFS技术交流群：164684842