Chinaunix首页 | 论坛 | 博客

doit_zzg的ChinaUnix博客

首页　| 　博文目录　| 　关于我

doit_zzg

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

全部博文（7）

文章存档

2012年（7）

我的朋友
最近访客
推荐博文
相关博文
FastDFS分布式文件系统点滴记录3 -- 网络模型、libevent框架使用

分类： 云计算

2012-03-23 23:17:37


FastDFS v3.0 版本以后，使用到了libevent库来处理网络连接请求。 其实，FastDFS 中 tracker、storage 的网络架构基本一样。下面以tracker 为例子介绍。

fdfs_trackerd.c 是 tracker 中main 函数的所在文件。
fdfs_trackerd.c  137行：

  1.     sock = socketServer(bind_addr, g_server_port, &result);
  2.     if (sock < 0)
  3.     {
  4.         log_destroy();
  5.         return result;
  6.     }

  8.     if ((result=tcpsetserveropt(sock, g_fdfs_network_timeout)) != 0)
  9.     {
  10.         log_destroy();
  11.         return result;
  12.     }
代码分析：
1. 调用sockopt.c 中socketServer函数，打开监听端口，返回socket句柄；
2. 通过 tcpsetserveropt 函数，内部是在调用setsockopt 实现 SO_SNDTIMEO、SO_RCVTIMEO、TCP_NODELAY、tcpsetkeepalive等参数的设置。
接着往下走，是tracker 的一些初始化代码tracker_service_init，我们暂时先跳过，只分析网络部分。详细的入口分析，放在下一篇文章完成。
fdfs_trackerd.c  309行：

  1.     bTerminateFlag = false;
  2.     bAcceptEndFlag = false;

  4.     tracker_accept_loop(sock);
  5.     bAcceptEndFlag = true;
  6.     if (g_schedule_flag)
  7.     {
  8.         pthread_kill(schedule_tid, SIGINT);
  9.     }
  10.     tracker_terminate_threads();
这里，通过调用 tracker_accept_loop 函数 进入 socket 的 accept 循环。

我们进入这个函数，
tracker_service.c  202行：

  1. void tracker_accept_loop(int server_sock)
  2. {
  3.     int incomesock;
  4.     struct sockaddr_in inaddr;
  5.     socklen_t sockaddr_len;
  6.     struct tracker_thread_data *pThreadData;

  8.     while (g_continue_flag)
  9.     {
  10.         sockaddr_len = sizeof(inaddr);
  11.         incomesock = accept(server_sock, (struct sockaddr*)&inaddr, &sockaddr_len);
  12.         if (incomesock < 0) //error
  13.         {
  14.             if (!(errno == EINTR || errno == EAGAIN))
  15.             {
  16.                 logError("file: "__FILE__", line: %d, " \
  17.                     "accept failed, " \
  18.                     "errno: %d, error info: %s", \
  19.                     __LINE__, errno, STRERROR(errno));
  20.             }

  22.             continue;
  23.         }

  25.         pThreadData = g_thread_data + incomesock % g_work_threads;
  26.         if (write(pThreadData->pipe_fds[1], &incomesock, \
  27.             sizeof(incomesock)) != sizeof(incomesock))
  28.         {
  29.             close(incomesock);
  30.             logError("file: "__FILE__", line: %d, " \
  31.                 "call write failed, " \
  32.                 "errno: %d, error info: %s", \
  33.                 __LINE__, errno, STRERROR(errno));
  34.         }
  35.     }
  36. }
注意收到网络connect请求后，依据接收来的socket句柄 incomesock，找到执行线程对象；

  1. pThreadData = g_thread_data + incomesock % g_work_threads;
接着，使用管道通信的方式通知网络连接incomesock的到来。
全局变量g_thread_data的类型 是 struct tracker_thread_data *

  1. struct tracker_thread_data *g_thread_data = NULL;
而线程结构的声明在tracker_service.h 25行：

  1. struct tracker_thread_data
  2. {
  3.         struct event_base *ev_base;
  4.         int pipe_fds[2];
  5. };
libevent实现了网络IO,timer,signal的事件触发机制. 可以很方便的应用于event-driven服务器中,作为其底层事件处理模块. 比较成功的案例有 memcache(分布式缓存).  event_base是整个libevent的核心部分. 
而对g_thread_data的初始化动作，是在tracker_service_init 初始化函数完成的。

tracker_service_init 中 tracker_service.c 83行：
 
  1.     g_thread_data = (struct tracker_thread_data *)malloc(sizeof( \
  2.                 struct tracker_thread_data) * g_work_threads);
  3.     if (g_thread_data == NULL)
  4.     {
  5.         logError("file: "__FILE__", line: %d, " \
  6.             "malloc %d bytes fail, errno: %d, error info: %s", \
  7.             __LINE__, (int)sizeof(struct tracker_thread_data) * \
  8.             g_work_threads, errno, STRERROR(errno));
  9.         return errno != 0 ? errno : ENOMEM;
  10.     }

  12.     g_tracker_thread_count = 0;
  13.     pDataEnd = g_thread_data + g_work_threads;
  14.     for (pThreadData=g_thread_data; pThreadData<pDataEnd; pThreadData++)
  15.     {
  16.         pThreadData->ev_base = event_base_new();
  17.         if (pThreadData->ev_base == NULL)
  18.         {
  19.             result = errno != 0 ? errno : ENOMEM;
  20.             logError("file: "__FILE__", line: %d, " \
  21.                 "event_base_new fail.", __LINE__);
  22.             return result;
  23.         }

  25.         if (pipe(pThreadData->pipe_fds) != 0)
  26.         {
  27.             result = errno != 0 ? errno : EPERM;
  28.             logError("file: "__FILE__", line: %d, " \
  29.                 "call pipe fail, " \
  30.                 "errno: %d, error info: %s", \
  31.                 __LINE__, result, STRERROR(result));
  32.             break;
  33.         }

  35.         if ((result=set_nonblock(pThreadData->pipe_fds[0])) != 0)
  36.         {
  37.             break;
  38.         }

  40.         if ((result=pthread_create(&tid, &thread_attr, \
  41.             work_thread_entrance, pThreadData)) != 0)
  42.         {
  43.             logError("file: "__FILE__", line: %d, " \
  44.                 "create thread failed, startup threads: %d, " \
  45.                 "errno: %d, error info: %s", \
  46.                 __LINE__, g_tracker_thread_count, \
  47.                 result, STRERROR(result));
  48.             break;
  49.         }
  50.         else
  51.         {
  52.             if ((result=pthread_mutex_lock(&tracker_thread_lock)) != 0)
  53.             {
  54.                 logError("file: "__FILE__", line: %d, " \
  55.                     "call pthread_mutex_lock fail, " \
  56.                     "errno: %d, error info: %s", \
  57.                     __LINE__, result, STRERROR(result));
  58.             }
  59.             g_tracker_thread_count++;
  60.             if ((result=pthread_mutex_unlock(&tracker_thread_lock)) != 0)
  61.             {
  62.                 logError("file: "__FILE__", line: %d, " \
  63.                     "call pthread_mutex_lock fail, " \
  64.                     "errno: %d, error info: %s", \
  65.                     __LINE__, result, STRERROR(result));
  66.             }
  67.         }
  68.     }
代码分析：
1. 首先根据配置g_work_threads，创建工作者线程对象g_thread_data；
2. 循环初始化libevent  event_base结构体 pThreadData->ev_base = event_base_new();
3. 创建管道 if (pipe(pThreadData->pipe_fds) != 0)
4. 创建工作者线程，线程执行函数是 work_thread_entrance


上面分析到，当accept 收到socket连接后，会通过pipe 通知给工作者线程。下面分析线程执行函数work_thread_entrance。

我们进入这个函数，
tracker_service.c  239行：

  1. static void *work_thread_entrance(void* arg)
  2. {
  3.     int result;
  4.     struct tracker_thread_data *pThreadData;
  5.     struct event ev_notify;

  7.     pThreadData = (struct tracker_thread_data *)arg;
  8.     do
  9.     {
  10.         event_set(&ev_notify, pThreadData->pipe_fds[0], \
  11.             EV_READ | EV_PERSIST, recv_notify_read, NULL);
  12.         if ((result=event_base_set(pThreadData->ev_base, &ev_notify)) != 0)
  13.         {
  14.             logCrit("file: "__FILE__", line: %d, " \
  15.                 "event_base_set fail.", __LINE__);
  16.             break;
  17.         }
  18.         if ((result=event_add(&ev_notify, NULL)) != 0)
  19.         {
  20.             logCrit("file: "__FILE__", line: %d, " \
  21.                 "event_add fail.", __LINE__);
  22.             break;
  23.         }

  25.         while (g_continue_flag)
  26.         {
  27.             event_base_loop(pThreadData->ev_base, 0);
  28.         }
  29.     } while (0);

  31.     event_base_free(pThreadData->ev_base);

  33.     if ((result=pthread_mutex_lock(&tracker_thread_lock)) != 0)
  34.     {
  35.         logError("file: "__FILE__", line: %d, " \
  36.             "call pthread_mutex_lock fail, " \
  37.             "errno: %d, error info: %s", \
  38.             __LINE__, result, STRERROR(result));
  39.     }
  40.     g_tracker_thread_count--;
  41.     if ((result=pthread_mutex_unlock(&tracker_thread_lock)) != 0)
  42.     {
  43.         logError("file: "__FILE__", line: %d, " \
  44.             "call pthread_mutex_lock fail, " \
  45.             "errno: %d, error info: %s", \
  46.             __LINE__, result, STRERROR(result));
  47.     }

  49.     return NULL;
  50. }

代码分析：
1. event_set 设置事件回调函数 recv_notify_read
2. while 循环，event_base_loop
3. event_base_loop 等待事件被触发，然后调用事件的回调函数。

通俗点讲，就是，pipe 上面有事件到来，那么就去执行 recv_notify_read 。好，我们接着分析 recv_notify_read 函数。

tracker_nio.c 63行：

  1. void recv_notify_read(int sock, short event, void *arg)
  2. {
  3.     int bytes;
  4.     int incomesock;
  5.     int result;
  6.     struct tracker_thread_data *pThreadData;
  7.     struct fast_task_info *pTask;
  8.     char szClientIp[IP_ADDRESS_SIZE];
  9.     in_addr_t client_addr;

  11.     while (1)
  12.     {
  13.         if ((bytes=read(sock, &incomesock, sizeof(incomesock))) < 0)
  14.         {
  15.             if (!(errno == EAGAIN || errno == EWOULDBLOCK))
  16.             {
  17.                 logError("file: "__FILE__", line: %d, " \
  18.                     "call read failed, " \
  19.                     "errno: %d, error info: %s", \
  20.                     __LINE__, errno, STRERROR(errno));
  21.             }

  23.             break;
  24.         }
  25.         else if (bytes == 0)
  26.         {
  27.             break;
  28.         }

  30.         if (incomesock < 0)
  31.         {
  32.             struct timeval tv;
  33.                         tv.tv_sec = 1;
  34.                         tv.tv_usec = 0;
  35.             pThreadData = g_thread_data + (-1 * incomesock - 1) % \
  36.                     g_work_threads;
  37.             event_base_loopexit(pThreadData->ev_base, &tv);
  38.             return;
  39.         }

  41.         client_addr = getPeerIpaddr(incomesock, \
  42.                 szClientIp, IP_ADDRESS_SIZE);
  43.         if (g_allow_ip_count >= 0)
  44.         {
  45.             if (bsearch(&client_addr, g_allow_ip_addrs, \
  46.                     g_allow_ip_count, sizeof(in_addr_t), \
  47.                     cmp_by_ip_addr_t) == NULL)
  48.             {
  49.                 logError("file: "__FILE__", line: %d, " \
  50.                     "ip addr %s is not allowed to access", \
  51.                     __LINE__, szClientIp);

  53.                 close(incomesock);
  54.                 continue;
  55.             }
  56.         }

  58.         if (tcpsetnonblockopt(incomesock) != 0)
  59.         {
  60.             close(incomesock);
  61.             continue;
  62.         }

  64.         pTask = free_queue_pop();
  65.         if (pTask == NULL)
  66.         {
  67.             logError("file: "__FILE__", line: %d, " \
  68.                 "malloc task buff failed, you should " \
  69.                 "increase the parameter: max_connections", \
  70.                 __LINE__);
  71.             close(incomesock);
  72.             continue;
  73.         }

  75.         pThreadData = g_thread_data + incomesock % g_work_threads;

  77.         strcpy(pTask->client_ip, szClientIp);
  78.     
  79.         event_set(&pTask->ev_read, incomesock, EV_READ, \
  80.                 client_sock_read, pTask);
  81.         if (event_base_set(pThreadData->ev_base, &pTask->ev_read) != 0)
  82.         {
  83.             task_finish_clean_up(pTask);
  84.             close(incomesock);

  86.             logError("file: "__FILE__", line: %d, " \
  87.                 "event_base_set fail.", __LINE__);
  88.             continue;
  89.         }

  91.         event_set(&pTask->ev_write, incomesock, EV_WRITE, \
  92.                 client_sock_write, pTask);
  93.         if ((result=event_base_set(pThreadData->ev_base, \
  94.                 &pTask->ev_write)) != 0)
  95.         {
  96.             task_finish_clean_up(pTask);
  97.             close(incomesock);

  99.             logError("file: "__FILE__", line: %d, " \
  100.                     "event_base_set fail.", __LINE__);
  101.             continue;
  102.         }

  104.         if (event_add(&pTask->ev_read, &g_network_tv) != 0)
  105.         {
  106.             task_finish_clean_up(pTask);
  107.             close(incomesock);

  109.             logError("file: "__FILE__", line: %d, " \
  110.                 "event_add fail.", __LINE__);
  111.             continue;
  112.         }
  113.     }
  114. }
代码分析：
1.首先从管道里面读取 socket 句柄 incomesock；
2.  126行，pTask = free_queue_pop(); 取出 task 结构；
3.  接下来设置libevent中对 incomesock 的读、写事件的回调函数：client_sock_read、client_sock_write
4. 通过event_set ，这样incomesock 也进入libevent 事件模型里面，等待着网络事件的发生。

当socket 网络可读事件到来的时候，会执行client_sock_read，当读取足够的数据后，
tracker_nio.c 325 行：
 
  1. pTask->offset += bytes;
  2.         if (pTask->offset >= pTask->length) //recv done
  3.         {
  4.             pTask->req_count++;
  5.             tracker_deal_task(pTask);
  6.             return;
  7.         }
tracker_deal_task 函数功能 是 tracker 根据协议处理不同的任务：
tracker_service.c 3137行：

  1. int tracker_deal_task(struct fast_task_info *pTask)
  2. {
  3.     TrackerHeader *pHeader;
  4.     int result;

  6.     pHeader = (TrackerHeader *)pTask->data;
  7.     switch(pHeader->cmd)
  8.     {
  9.         case TRACKER_PROTO_CMD_STORAGE_BEAT:
  10.             TRACKER_CHECK_LOGINED(pTask)
  11.             result = tracker_deal_storage_beat(pTask);
  12.             break;

  14.         ...
  15. ...
  16. ...

  18.         case TRACKER_PROTO_CMD_TRACKER_COMMIT_NEXT_LEADER:
  19.             result = tracker_deal_commit_next_leader(pTask);
  20.             break;
  21.         default:
  22.             logError("file: "__FILE__", line: %d, " \
  23.                 "client ip: %s, unkown cmd: %d", \
  24.                 __LINE__, pTask->client_ip, \
  25.                 pHeader->cmd);
  26.             result = EINVAL;
  27.             break;
  28.     }

  30.     pHeader = (TrackerHeader *)pTask->data;
  31.     pHeader->status = result;
  32.     pHeader->cmd = TRACKER_PROTO_CMD_RESP;
  33.     long2buff(pTask->length - sizeof(TrackerHeader), pHeader->pkg_len);

  35.     send_add_event(pTask);

  37.     return 0;
  38. }
代码分析：
1. 依据协议pHeader->cmd，处理具体业务；
2. 业务处理完成后，调用send_add_event(pTask); 

接着跟踪send_add_event，会进入client_sock_write，将数据通过socket发送给客户端。
tracker_nio.c 353行：

  1. bytes = send(sock, pTask->data + pTask->offset, \
  2.                 pTask->length - pTask->offset, 0);
至此，tracker 的网络部分，分析完成了。storage 和 tracker 分析过程一样，不再重复。

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

阅读(5771) | 评论(0) | 转发(1) |
0

上一篇：FastDFS分布式文件系统点滴记录2 -- 架构概述

下一篇：FastDFS分布式文件系统点滴记录4 -- tracker 、storage 入口分析

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6