Memcached网络模型

之前用libevent开发了一个流媒体服务器。用线程池实现的。之后又看了memcached的网络相关实现，今天来整理一下memcached的实现流程。

memcached不同于Redis的单进程单线程，是采用多线程的工作方式。有一个主线程，同时维护了一个线程池（工作线程）。worker thread工作线程和main thread主线程之间主要通过pipe来进行通信。因为用了libevent，所以感觉比Redis稍微庞大点，没有在生产环境对比过Redis和memcached，所以也不好说什么性能比对。

主线程和工作线程都有一个event base，大体框架如下图：

整体框架图：

线程模型：

每个线程包括主线程都各自有独立的Libevent实例，Memcached的listen fd注册到主线程的Libevent实例上，由主线程来accept新的连接，接受新的连接后根据Round-robin算法选择工作线程，将新连接的socket fd封装为CQ_ITEM后push到所选工作线程的CQ队列中，然后主线程（notify_send_fd）通过管道发送字符“c”到工作线程（notify_receive_fd），而notify_receive_fd已经注册到工作线程的Libevent实例上了，这样工作线程就能收到通知“c”，然后从该工作线程的CQ队列中pop出CQ_ITEM进而取出新连接并将fd注册到工作线程的Libevent实例上，从而由工作线程来处理该连接的所有后续事件

代码层实现图：

源码解读：

1. 关键数据结构

/* An item in the connection queue. */主要用于存储用户socket连接的基本信息

主线程会将用户的socket连接信息封装成CQ_ITEM，并放入工作线程的处理队列中。工作线程得到主线程的pipe通知后，就会将队列中的ITEM取出来，创建libevent的socket读事件。
typedef struct conn_queue_item CQ_ITEM;
struct conn_queue_item {
    int               sfd;//socket fd
    enum conn_states  init_state;//事件类型
    int               event_flags;//libevent的flag
    int               read_buffer_size;//读取buffer的大小
    enum network_transport     transport;
    CQ_ITEM          *next;//下一个CQ_ITEM
};

/**
* The structure representing a connection into memcached.
*/
typedef struct conn conn;
struct conn {
    int    sfd;
    struct event event;
    short  ev_flags;
    short  which;   /** which events were just triggered */

    LIBEVENT_THREAD *thread; /* Pointer to the thread object serving this connection */
};

/* A connection queue. */
typedef struct conn_queue CQ;
struct conn_queue {
    CQ_ITEM *head;//指向队列的第一个节点
    CQ_ITEM *tail;//指向队列的最后一个节点
    pthread_mutex_t lock;//一个队列就对应一个锁
};

 

typedef struct {
    pthread_t thread_id;        /* unique ID of this thread */线程ID
    struct event_base *base;    /* libevent handle this thread uses */线程所使用的event_base
    struct event notify_event;  /* listen event for notify pipe */用于监听管道读事件的event
    int notify_receive_fd;      /* receiving end of notify pipe */管道的读端fd
    int notify_send_fd;         /* sending end of notify pipe */管道的写端fd
    struct thread_stats stats;  /* Stats generated by this thread */
    struct conn_queue *new_conn_queue; /* queue of new connections to handle */
    cache_t *suffix_cache;      /* suffix cache */
} LIBEVENT_THREAD;

2. CQ队列示意图

CQ_ITEM的创建实现了一个内存池，具体可以参考函数static CQ_ITEM *cqi_new(void)和static void cqi_free(CQ_ITEM *item)，解决频繁分配释放内存而产生的内存碎片问题。

3.文字代码流程

（1）在main函数中调用main_base = event_init()来初始化主线程Libevent实例。
（2）在main函数中调用thread_init来初始化工作线程，并将主线程Libevent实例作为参数传入。
（3）在thread_init函数中为指定数量的工作线程分配内存，为每个线程创建管道，并分别绑定到通知收和发的socket描述符上，调用函数setup_thread初始化线程信息，调用函数create_worker为每个线程注册回调函数。每个线程有一个CQ队列。关键代码：
        for (i = 0; i < nthreads; i++) {
                int fds[2];
                if (pipe(fds)) {
                        ...
                }
                threads[i].notify_receive_fd = fds[0];
                threads[i].notify_send_fd = fds[1];
               setup_thread(&threads[i]);
               ...
        }
        /* Create threads after we‘ve done all the libevent setup. */
        for (i = 0; i < nthreads; i++) {
            create_worker(worker_libevent, &threads[i]);
        }

（4）在setup_thread函数中，为工作线程初始化Libevent实例，为主线程通知读（notify_receive_fd）注册回调函数thread_libevent_process，初始化cq队列，关键代码如下：
        static void setup_thread(LIBEVENT_THREAD *me) {
                me->base = event_init();
                ...
                /* Listen for notifications from other threads */
                event_set(&me->notify_event, me->notify_receive_fd,
                          EV_READ | EV_PERSIST, thread_libevent_process, me);
                event_base_set(me->base, &me->notify_event);
                if (event_add(&me->notify_event, 0) == -1) {
                        ...
                }
                me->new_conn_queue = malloc(sizeof(struct conn_queue));
                ...
                cq_init(me->new_conn_queue);
                ...
        }

（5）在thread_libevent_process函数中，读取主线程发送的通知接收消息，将主线程accept来的fd注册到工作线程的Libevent实例中，主线程accept来的fd从conn_queue队列获取，关键代码如下：
        static void thread_libevent_process(int fd, short which, void *arg) {
                LIBEVENT_THREAD *me = arg;
                CQ_ITEM *item;
                char buf[1];
                if (read(fd, buf, 1) != 1)
                        ...
                switch (buf[0]) {
                case ‘c‘:
                item = cq_pop(me->new_conn_queue);
                if (NULL != item) {
                        conn *c = conn_new(item->sfd, item->init_state, item->event_flags,
                                                        item->read_buffer_size, item->transport, me->base);
               ...
                }
        }
（6）在函数conn_new中，创建conn句柄，为句柄注册回调函数event_handler处理事件，将该句柄作为参数传入回调函数并设置到Libevent中，该函数的关键代码如下：
        conn *conn_new(const int sfd, enum conn_states init_state,
                        const int event_flags,
                        const int read_buffer_size, enum network_transport transport,
                        struct event_base *base) {
            conn *c = conn_from_freelist();
            if (NULL == c) {
                if (!(c = (conn *)calloc(1, sizeof(conn)))) {
                        ...
            }
           ...
           event_set(&c->event, sfd, event_flags, event_handler, (void *)c);
           event_base_set(base, &c->event);
           c->ev_flags = event_flags;
           if (event_add(&c->event, 0) == -1) {
                  ...
           }
           ...
        }

（7）在create_worker函数中，创建工作线程并注册回调函数，在工作线程的回调函数work_libevent中，开始Libevent主循环。

（8）在main函数中，调用函数server_sockets，再调用函数server_socket，进而调用函数new_socket，在调用函数conn_new，创建并注册listen fd到主线程Libevent实例上，最后开始Libevent主循环即event_base_loop。在conn_new函数关键代码见步骤（6）
（9）在event_handler函数中，调用函数drive_machine，在该函数中处理所有事件，其关键代码如下：
        static void drive_machine(conn *c) {
                ...
                while (!stop) {
                    switch(c->state) {
                        case conn_listening:
                                addrlen = sizeof(addr);
                                if ((sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen)) == -1) {
                                       ...                                     
                                }
                               ...
                               if (settings.maxconns_fast &&
                                    stats.curr_conns + stats.reserved_fds >= settings.maxconns - 1) {
                                        ...
                               } else {
                                    dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST,
                                             DATA_BUFFER_SIZE, tcp_transport);
                              }
                            stop = true;
                            break;
                            ...
                        }
                    }
                return;
        }
        在处理事件时，如果是listening事件，则调用函数dispatch_conn_new（Memcached.c/3785行）将accept fd分配给工作线程。

（10）在dispatch_conn_new函数中，根据round-robin算法将新连接push到所分配线程的CQ队列中，并通过管道发送通知消息“c，关键代码如下：
        void dispatch_conn_new(int sfd, enum conn_states init_state, int event_flags,
                               int read_buffer_size, enum network_transport transport) {
                    CQ_ITEM *item = cqi_new();
                    char buf[1];
                    int tid = (last_thread + 1) % settings.num_threads;
                    LIBEVENT_THREAD *thread = threads + tid;
                    last_thread = tid;
                    ...
                    cq_push(thread->new_conn_queue, item);
                    ...
                    buf[0] = ‘c‘;
                    if (write(thread->notify_send_fd, buf, 1) != 1) {
                        perror("Writing to thread notify pipe");
                    }
        }

dispatch_conn_new函数只在主线程中调用，last_thread为静态变量，每次将该变量值+1，再模线程数来选择工作线程。

4. conn状态机

1.listening:这个状态是主线程的connection默认状态，它只有这一个状态，它做的工作就是把接到连接分发到worker子线程。

2.conn_new_cmd:每个新连接的初始状态，这个状态会清空读写buf。

3.conn_waiting:这个状态就是在event_base中设置读事件，然后状态机暂停，挂起当前connection（函数退出，回调函数的attachment会记录这个connection），等待有新的信息过来，然后通过回调函数的attachment重新找到这个connection，然后启动状态机。

4.conn_read:该状态从sfd中读取客户端的指令信息。例如有用户提交数据过来的时候，工作线程监听到事件后，最终会走到这里。

5.conn_parse_cmd:判断具体的指令，如果是update的指令，那么需要跳转到conn_nread中，因为需要在从网络中读取固定byte的数据，如果是查询之类的指令，就直接查询完成后，跳转到conn_mwrite中，返回数据

6.conn_nread:从网络中读取指定大小的数据，这个数据就是更新到item的数据，然后将数据更新到hash和lru中去，然后跳转到conn_write

7.conn_write:这个状态主要是调用out_string函数会跳转到这个状态，一般都是提示信息和返回的状态信息，然后输出这些数据，然后根据write_to_go的状态，继续跳转

8.conn_mwrite:这个写是把connection中的msglist返回到客户端，这个msglist存的是item的数据，用于那种get等获得item的操作的返回数据。

9.conn_swallow:对于那种update操作，如果分配item失败，显然后面的nread，是无效的，客户端是不知道的，这样客户端继续发送特定的数量的数据，就需要把读到的这些数据忽略掉，然后如果把后面指定的数据都忽略掉了（set的两部提交，数据部分忽略掉），那么connection跳转到conn_new_cmd，如果读nread的那些特定数量的数据没有读到，直接跳转到conn_closing。

10.conn_closing:服务器端主动关闭连接，调用close函数关闭文件描述符，同时把conn结构体放到空闲队列中，供新的连接重用这写conn结构体。

Memcached的数据结构那块源码还没有看过，下次也梳理下。看过Redis的，数据结构设计的还是十分精妙的。

参考资料：

http://www.lvtao.net/c/623.html?utm_source=tuicool

http://www.linuxidc.com/Linux/2012-01/52515p3.htm

http://blog.csdn.net/funkri/article/details/17022309

http://blog.chinaunix.net/uid-16723279-id-3568441.html

http://blog.csdn.net/luotuo44/article/details/42705475

http://blog.chinaunix.net/uid-27767798-id-3415510.html

感谢以上作者的无私分享。

原文：http://www.cnblogs.com/chenxianpao/p/5572895.html