Chinaunix首页 | 论坛 | 博客
  • 博客访问: 57920
  • 博文数量: 7
  • 博客积分: 171
  • 博客等级: 入伍新兵
  • 技术积分: 121
  • 用 户 组: 普通用户
  • 注册时间: 2012-03-16 18:48
文章分类

全部博文(7)

文章存档

2012年(7)

我的朋友

分类: 云计算

2012-03-28 08:03:51

关于下载,其实和上传文件很相似。这里我们暂时先不考虑nginx httpd的方式,只通过client api 方式与FastDFS 交互。
首先,我们看fdfs_download_file.c ,这个文件下载文件的客户端例子,直接看71行:

  1.     result = storage_download_file_to_file1( \
  2.             pTrackerServer, NULL, \
  3.             file_id, local_filename, &file_size);
  4.     if (result != 0)
  5.     {
  6.         printf("download file fail, " \
  7.             "error no: %d, error info: %s\n", \
  8.             result, STRERROR(result));
  9.     }
我们到这个函数的内部,storage_client.c 707行:

  1. int storage_download_file_to_file1(TrackerServerInfo *pTrackerServer, \
  2.         TrackerServerInfo *pStorageServer, \
  3.         const char *file_id, \
  4.         const char *local_filename, int64_t *file_size)
  5. {
  6.     FDFS_SPLIT_GROUP_NAME_AND_FILENAME(file_id)

  7.     return storage_download_file_to_file(pTrackerServer, \
  8.             pStorageServer, group_name, filename, \
  9.             local_filename, file_size);
  10. }
代码分析:
1. 首先从file_id ,解析出group name、file name
2. 调用storage_download_file_to_file

接着往下分析,进入storage_download_file_to_file,storage_client.c 719行:

  1. int storage_download_file_to_file(TrackerServerInfo *pTrackerServer, \
  2.             TrackerServerInfo *pStorageServer, \
  3.             const char *group_name, const char *remote_filename, \
  4.             const char *local_filename, int64_t *file_size)
  5. {
  6.     char *pLocalFilename;
  7.     pLocalFilename = (char *)local_filename;
  8.     return storage_do_download_file(pTrackerServer, pStorageServer, \
  9.             FDFS_DOWNLOAD_TO_FILE, group_name, remote_filename, \
  10.             &pLocalFilename, NULL, file_size);
  11. }
storage_download_file_to_file函数内部,会调用storage_do_download_file下载文件,注意命令FDFS_DOWNLOAD_TO_FILE。
我们通过分析storage_do_download_file,内部会和tracker服务器命令交互TRACKER_PROTO_CMD_SERVICE_QUERY_FETCH_ONE,获取合适的storage服务器。然后和storage 通过 STORAGE_PROTO_CMD_DOWNLOAD_FILE 命令下载文件。这部分和上传机制类似。我们不做详细分析。
这里,由于与tracker 交互时候发的命令是TRACKER_PROTO_CMD_SERVICE_QUERY_FETCH_ONE,这点与upload机制类似。读者可以参考此前关于upload的机制分析。
我们着重分析下与storage 交互的过程。

storage_service.c 6456行:

  1. int storage_deal_task(struct fast_task_info *pTask)
storage_deal_task 是storage 处理任务的入口,我们之间观察STORAGE_PROTO_CMD_DOWNLOAD_FILE,storage_service.c 6467行:

  1.     case STORAGE_PROTO_CMD_DOWNLOAD_FILE:
  2.             result = storage_server_download_file(pTask);
  3.             break;
接着分析storage_server_download_file,storage_service.c 5489行:

  1.     return storage_read_from_file(pTask, file_offset, download_bytes, \
  2.             storage_download_file_done_callback, store_path_index);
代码分析:
1. storage_server_download_file 首先解析协议,解析出file相关信息;
2. 然后调用storage_read_from_file,注意回调函数storage_download_file_done_callback

接着往下分析,storage_service.c 5520行:

  1.     static int storage_read_from_file(struct fast_task_info *pTask, \
  2.         const int64_t file_offset, const int64_t download_bytes, \
  3.         FileDealDoneCallback done_callback, \
  4.         const int store_path_index)
  5. {
  6.     StorageClientInfo *pClientInfo;
  7.     StorageFileContext *pFileContext;
  8.     TrackerHeader *pHeader;
  9.     int result;

  10.     pClientInfo = (StorageClientInfo *)pTask->arg;
  11.     pFileContext = &(pClientInfo->file_context);

  12.     pClientInfo->deal_func = dio_read_file;
  13.     pClientInfo->clean_func = dio_read_finish_clean_up;
  14.     pClientInfo->total_length = sizeof(TrackerHeader) + download_bytes;
  15.     pClientInfo->total_offset = 0;

  16.     pFileContext->op = FDFS_STORAGE_FILE_OP_READ;
  17.     pFileContext->open_flags = O_RDONLY | g_extra_open_file_flags;
  18.     pFileContext->offset = file_offset;
  19.     pFileContext->start = file_offset;
  20.     pFileContext->end = file_offset + download_bytes;
  21.     pFileContext->dio_thread_index = storage_dio_get_thread_index( \
  22.         pTask, store_path_index, pFileContext->op);
  23.     pFileContext->done_callback = done_callback;

  24.     pTask->length = sizeof(TrackerHeader);

  25.     pHeader = (TrackerHeader *)pTask->data;
  26.     pHeader->status = 0;
  27.     pHeader->cmd = STORAGE_PROTO_CMD_RESP;
  28.     long2buff(download_bytes, pHeader->pkg_len);

  29.     if ((result=storage_dio_queue_push(pTask)) != 0)
  30.     {
  31.         if (pFileContext->fd >= 0)
  32.         {
  33.             close(pFileContext->fd);
  34.         }
  35.         pClientInfo->total_length = sizeof(TrackerHeader);
  36.         return result;
  37.     }

  38.     return STORAGE_STATUE_DEAL_FILE;
  39. }
代码分析:
1.这里和上传相似,通过队列,实现nio和dio 的转换;
2. 注意 FDFS_STORAGE_FILE_OP_READ,说明操作是读取文件;
3. pClientInfo->deal_func = dio_read_file;
4. 上面的函数指针是dio 执行函数;
5. pFileContext->done_callback = done_callback;完成时的回调行数,也就是之前提到的storage_download_file_done_callback。

storage_dio.c 中 dio_thread_entrance 函数处理和上传机制一样。我们直接分析deal_func函数,这里是dio_read_file。
storage_dio.c 334行:

  1. int dio_read_file(struct fast_task_info *pTask)
  2. {
  3.     StorageFileContext *pFileContext;
  4.     int result;
  5.     int64_t remain_bytes;
  6.     int capacity_bytes;
  7.     int read_bytes;

  8.     pFileContext = &(((StorageClientInfo *)pTask->arg)->file_context);

  9.     do
  10.     {
  11.     if ((result=dio_open_file(pFileContext)) != 0)
  12.     {
  13.         break;
  14.     }

  15.     remain_bytes = pFileContext->end - pFileContext->offset;
  16.     capacity_bytes = pTask->size - pTask->length;
  17.     read_bytes = (capacity_bytes < remain_bytes) ? \
  18.                 capacity_bytes : remain_bytes;

  19.     /*
  20.     logInfo("###before dio read bytes: %d, pTask->length=%d, file offset=%ld", \
  21.         read_bytes, pTask->length, pFileContext->offset);
  22.     */

  23.     if (read(pFileContext->fd, pTask->data + pTask->length, \
  24.         read_bytes) != read_bytes)
  25.     {
  26.         result = errno != 0 ? errno : EIO;
  27.         logError("file: "__FILE__", line: %d, " \
  28.             "read from file: %s fail, " \
  29.             "errno: %d, error info: %s", \
  30.             __LINE__, pFileContext->filename, \
  31.             result, STRERROR(result));
  32.     }

  33.     pthread_mutex_lock(&g_dio_thread_lock);
  34.     g_storage_stat.total_file_read_count++;
  35.     if (result == 0)
  36.     {
  37.         g_storage_stat.success_file_read_count++;
  38.     }
  39.     pthread_mutex_unlock(&g_dio_thread_lock);

  40.     if (result != 0)
  41.     {
  42.         break;
  43.     }

  44.     pTask->length += read_bytes;
  45.     pFileContext->offset += read_bytes;

  46.     /*
  47.     logInfo("###after dio read bytes: %d, pTask->length=%d, file offset=%ld", \
  48.         read_bytes, pTask->length, pFileContext->offset);
  49.     */

  50.     if (pFileContext->offset < pFileContext->end)
  51.     {
  52.         storage_nio_notify(pTask); //notify nio to deal
  53.     }
  54.     else
  55.     {
  56.         /* file read done, close it */
  57.         close(pFileContext->fd);
  58.         pFileContext->fd = -1;

  59.         pFileContext->done_callback(pTask, result);
  60.     }

  61.     return 0;

  62.     } while (0);

  63.     /* file read error, close it */
  64.     if (pFileContext->fd > 0)
  65.     {
  66.         close(pFileContext->fd);
  67.         pFileContext->fd = -1;
  68.     }

  69.     pFileContext->done_callback(pTask, result);
  70.     return result;
  71. }
代码分析:
1. 文件操作,打开文件,读取文件内容;
2.通过storage_nio_notify函数,通知nio 处理;
3.通过nio、dio直接的交互,直到正确读取、发送完文件数据,调用done_callback

我们接着分析 done_callback,也就是storage_download_file_done_callback。storage_service.c 596行:

  1. static void storage_download_file_done_callback(struct fast_task_info *pTask, \
  2.             const int err_no)
  3. {
  4.     StorageFileContext *pFileContext;
  5.     TrackerHeader *pHeader;

  6.     pFileContext = &(((StorageClientInfo *)pTask->arg)->file_context);
  7.     if (err_no != 0)
  8.     {
  9.         pthread_mutex_lock(&stat_count_thread_lock);
  10.         g_storage_stat.total_download_count++;
  11.         g_storage_stat.total_download_bytes += \
  12.                 pFileContext->offset - pFileContext->start;
  13.         pthread_mutex_unlock(&stat_count_thread_lock);

  14.         if (pTask->length == sizeof(TrackerHeader)) //never response
  15.         {
  16.             pHeader = (TrackerHeader *)pTask->data;
  17.             pHeader->status = err_no;
  18.             storage_nio_notify(pTask);
  19.         }
  20.         else
  21.         {
  22.             task_finish_clean_up(pTask);
  23.         }
  24.     }
  25.     else
  26.     {
  27.         CHECK_AND_WRITE_TO_STAT_FILE2_WITH_BYTES( \
  28.             g_storage_stat.total_download_count, \
  29.             g_storage_stat.success_download_count, \
  30.             g_storage_stat.total_download_bytes, \
  31.             g_storage_stat.success_download_bytes, \
  32.             pFileContext->end - pFileContext->start)

  33.         storage_nio_notify(pTask);
  34.     }
  35. }
分析发现,storage_download_file_done_callback 也即是做一些文件下载结尾的动作。

OK,这里下载就完成了。和上传很是相似。通过http方式下载,我们放到后面的章节分析。

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









阅读(6447) | 评论(0) | 转发(2) |
给主人留下些什么吧!~~