memcache的线程模型

MC采用一master多worker的工作模型,由master负责accept客户端请求,然后以RR分发给worker;
-t 线程数,用于处理请求,默认为4 
-b backlog队列长度,默认1024 

线程结构体
  1. typedef struct {  
  2.     pthread_t thread_id;        /* unique ID of this thread */  
  3.     struct event_base *base;    /* libevent handle this thread uses */  
  4.     struct event notify_event;  /* listen event for notify pipe */  
  5.     int notify_receive_fd;      /* receiving end of notify pipe */  
  6.     int notify_send_fd;         /* sending end of notify pipe */  
  7.     CQ  new_conn_queue;         /* queue of new connections to handle */  
  8. } LIBEVENT_THREAD;  
每个线程都包含一个CQ队列,一条通知管道pipe 和一个libevent的实例event_base ;
当master接受新连接后,通过pipe通知worker;
无论是主线程还是workers线程全部通过libevent管理网络事件,实际上每个线程都是一个单独的libevent实例 ;

启动流程
 



代码实现
master调用accept等待客户端连接(accept为非阻塞调用),返回的sfd也同样被设置为非阻塞,之后分发给worker;
下面看看dispatch_conn_new内部是如何进行链接分发的。
  1. void dispatch_conn_new(int sfd, enum conn_states init_state, int event_flags,  
  2.                        int read_buffer_size, enum network_transport transport) {  
  3.     CQ_ITEM *item = cqi_new();//创建一个conn_item 
  4.     char buf[1];  
  5.     int tid = (last_thread + 1) % settings.num_threads;//通过round-robin算法选择一个线程 

  6.     LIBEVENT_THREAD *thread = threads + tid;//thread数组存储了所有的工作线程  

  7.     cq_push(thread->new_conn_queue, item);//投递item信息到Worker线程的工作队列中  

  8.     buf[0] = ‘c‘;  
  9.     //在Worker线程的notify_send_fd写入字符c,表示有连接     
  10.     if (write(thread->notify_send_fd, buf, 1) != 1) {  
  11.         perror("Writing to thread notify pipe");  
  12.     } 

worker thread在管道描述符notify_send_fd的回调函数为thread_libevent_process,故一旦有数据到达立刻执行
//子线程会在PIPE管道读上面建立libevent事件,事件回调函数是thread_libevent_process 
  1. static void thread_libevent_process(int fd, short which, void *arg) {  
  2.     LIBEVENT_THREAD *me = arg;  
  3.     CQ_ITEM *item;  
  4.     char buf[1];  
  5.   
  6.     if (read(fd, buf, 1) != 1)//PIPE管道读取一个字节的数据  
  7.         if (settings.verbose > 0)  
  8.             fprintf(stderr, "Can‘t read from libevent pipe\n");  
  9.   
  10.     switch (buf[0]) {  
  11.     case ‘c‘://如果是c,则处理网络连接  
  12.     item = cq_pop(me->new_conn_queue);//从连接队列读出Master线程投递的消息 \

  1.     if (NULL != item) {  
  2.         conn *c = conn_new(item->sfd, item->init_state, item->event_flags,  
  3.                            item->read_buffer_size, item->transport, me->base);//创建连接 

conn_new里面会建立sfd的网络监听libevent事件,事件回调函数为event_handler,而event_handler的执行流程最终会进入到业务处理的状态机中。
  1. conn *conn_new(const int sfd, enum conn_states init_state, const int event_flags,  
  2.         const int read_buffer_size, enum network_transport transport,  
  3.         struct event_base *base)  
  4. {  
  5.     conn *c = conn_from_freelist();//获取一个空闲连接,conn是Memcached内部对网络连接的一个封装  
  6.   
  7.     if (NULL == c)//如果没有空闲的连接  
  8.     {  
  9.         if (!(c = (conn *) calloc(1, sizeof(conn))))//申请空间  
  10.         {  
  11.             fprintf(stderr, "calloc()\n");  
  12.             return NULL;  
  13.         }MEMCACHED_CONN_CREATE(c);

  1. //每个conn都自带读入和输出缓冲区,在进行网络收发数据时,特别方便  
  2.         c->rbuf = (char *) malloc((size_t) c->rsize);  
  3.         c->wbuf = (char *) malloc((size_t) c->wsize);  
  4.         c->ilist = (item **) malloc(sizeof(item *) * c->isize);  

  1.  //建立sfd描述符上面的event事件,事件回调函数为event_handler  
  2.     event_set(&c->event, sfd, event_flags, event_handler, (void *) c);  
  3.     event_base_set(base, &c->event);  
  4.     c->ev_flags = event_flags;  
  5.     if (event_add(&c->event, 0) == -1)  
  6.     {         
  7.        //如果建立libevent事件失败,将创建的conn添加到空闲列表中  
  8.        if (conn_add_to_freelist(c))  
  9.         {  
  10.             conn_free(c);  
  11.         }  
  12.         perror("event_add");  
  13.         return NULL;  
  14.     }  

  1. //libevent事件回调函数的处理,回调函数被调用时,表明Memcached监听的端口号有网络事件到了  
  2. void event_handler(const int fd, const short which, void *arg)  
  3. {  
  4. conn *c;  
  1. //进入业务处理状态机  
  2. drive_machine(c);  
  3.   


线程通信
master和worker通过连接队列进行单向通信,即Master接受到新的客户端连接时,将sfd封装到conn_queue_item并投送给woker,Worker从其连接队列中读取conn_queue_item同客户端连接进行交互,详情参见dispatch_conn_new;
  1. struct conn_queue_item {  
  2.     int               sfd;//accept之后的描述符  
  3.     enum conn_states  init_state;//连接的初始状态  
  4.     int               event_flags;//libevent标志  
  5.     int               read_buffer_size;//读取数据缓冲区大小  
  6.     enum network_transport     transport;//内部通信所用的协议  
  7.     CQ_ITEM          *next;//用于实现链表的指针  
  8. };  

  9. struct conn_queue {  
  10.     CQ_ITEM *head;//头指针,注意这里是单链表,不是双向链表  
  11.     CQ_ITEM *tail;//尾部指针,  
  12.     pthread_mutex_t lock;//锁  
  13.     pthread_cond_t  cond;//条件变量  
  14. };  

  15. //获取一个连接  
  16. static CQ_ITEM *cq_pop(CQ *cq) {  
  17.     CQ_ITEM *item;  
  18.   
  19.     pthread_mutex_lock(&cq->lock);//执行加锁操作  
  20.     item = cq->head;//获得头部指针指向的数据  
  21.     if (NULL != item) {  
  22.         cq->head = item->next;//更新头指针信息  
  23.         if (NULL == cq->head)//这里为空的话,则尾指针也为空,链表此时为空  
  24.             cq->tail = NULL;  
  25.     }
  26.     pthread_mutex_unlock(&cq->lock);//释放锁操作  
  27.   
  28.     return item;  
  29. }  

参考链接 http://blog.csdn.net/column/details/lc-memcached.html

郑重声明:本站内容如果来自互联网及其他传播媒体,其版权均属原媒体及文章作者所有。转载目的在于传递更多信息及用于网络分享,并不代表本站赞同其观点和对其真实性负责,也不构成任何其他建议。