渐进式解析 memcached 源码 - 线程机制

2018-04-12 阅读量

memcached 多线程网络编程的巅峰，其线程机制的设计十分之精妙。线程池部分主要包含主线程、工作线程以及辅助线程，另外再辅以状态机、通信管道，就大致组成了整个线程机制

memcached 网络线程模型事件完全基于 libevent 处理机制，利用基础的多路复用IO模型进行事件监控，以下我们就 网络模型 以及 源码解析 两部分进行解读

网络模型

memcached 服务器采用 master-woker 模式进行工作，后再辅以辅助线程。服务端采用 socket 与客户端通讯，主线程、工作线程采用 pipe管道进行通讯。主线程采用 libevent 监听 listen、accept 的读事件，事件响应后将连接信息的数据结构封装起来根据算法选择合适的工作线程，将连接任务携带连接信息分发出去，相应的线程利用连接描述符建立与客户端的socket连接并进行后续的存取数据操作。主线程和工作线程处理事件流都采用状态机进行事件转移。

线程模型

线程模型主要为一个主线程和若干个工作线程，另外包含一些 slab、assoc、logger 等辅助线程，后续会一一介绍。

主线程 main 函数在初始化以后会创建 main_base，并将服务端端口的 listen、accept 加入至 libevent 的 event_base 读事件 监控中
主线程 main 函数在初始化以后会创建 若干个 woker thread 工作线程，每个工作线程都会有自己的 LIBEVENT_THREAD数据结构来存储线程信息（线程基本信息、线程队列、pipi信息）。同时每个工作线程都会有属于自己的 pipi 读写端，并加入至 event_base 进行读事件监听，主线程如有需要，会往 pipe写端写入 “c”，工作线程捕获读事件后进行处理，即完成 master-woker 线程通讯
主线程监听到来自 client 的 socket connect 后，创建一个 conn_item（主要包含建立连接所需要的描述符等信息）。主线程根据线程位数通过简单的 求余操作 选定合适的工作线程，并把 conn_item 写入该工作线程的 conn_queue处理队列 中，随后主线程往该工作线程的 pipi 中写入信息通知该线程处理任务
工作线程接到主线程通知后，获取处理队列中的连接信息，与客户端进行通讯，创建针对客户端的 socket 读事件，获取客户端传递的读写请求。client 一旦有消息传递过来，工作线程捕获读事件后通过 状态机 进行事件流的处理
另外 memcached 还有功能丰富的辅助线程，可以根据配置进行按需启动
- slab thread slab 线程主要负责实现一个 slabautomove 的功能，memcahced内存区域分为不同的 slabclass，每个 slabclass 里面包含相同大小的内存块，slab automove 通过在后台运行一个线程，这个线程会在某些 slabclass 内存不足时，在 slabclass 之间移动内存来保证其正常工作。slab automove 主要涉及到两个线程 slab_maintenance_thread 和 slab_rebalance_thread，slab_maintenance_thread 决定是否需要进行一定， slab_rebalance_thread 进行实际的移动工作，两个线程通过一个 do_run_slab_rebalance_thread 来交互。具体 slab 机制后续存储部分加以详细介绍，这里先挖个坑…
- logger thread logger 线程主要负责日志管理
- assoc thread assoc线程主要负责 hash表的管理，负责 hash 的扩展和维护
- conn timeout thread conn timeout 线程主要控制允许空闲客户端的连接情况。memcached 对于连接上来的客户端，都会尽量保持，不会主动断开，但有些程序员写的程序都会开启新的连接，却不会关闭老的连接。导致无用的空闲连接仍然占用资源。因此，从1.4.27版本起 memcached 新增了 idle_timeout 参数，可以设置 idle 时间，默认为0，不会主动断开
- exstore thread exstore 线程主要作用于需要将数据存储至内存之外其他缓存介质的处理，此功能需要编译选项 --enable-extstore，并且在启动的时候配置 -o ext_path=/path/to/a/datafile,ext_page_count=100
- lru thread lru 线程，主要用来处理 LRU爬虫淘汰机制，这个后续会作为重点详细介绍，受控于 settings.lru_crawler 变量，可以通过参数 -o lru_crawler 设置为 true 进行启动

memcached-threads1

状态机

状态机在 memcached 线程模型中至关重要，主导整个事件流的处理逻辑，状态机的处理函数为 drive_machine，下面为简要的状态转移逻辑，具体命令解析、数据输出等我们后续会详细讲解（再次挖坑…^-^）

初始状态为 conn_lisenting，意为主线程监听来自客户端的请求
分发任务至工作线程后开始进入 conn_new_cmd 状态，自此以后即为工作线程的处理范围。conn_new_cmd 状态下工作线程重置命令句柄，根据 conn_queue队列 pop 出任务，建立 client 端的读监控事件
当 client 产生 send 动作，状态为 conn_read，通过 try_read_network 方法读取客户端报文，如果读取失败则进入 conn_closing，关闭客户端连接；如果没有读取到任何数据，则进入 conn_waiting，继续通过事件监听报文的到来；如果读取成功，状态进入 conn_parse_cmd 进行命令解析，命令内容存入连接数据结构中的 c->rbuf 容器中
conn_parse_cmd 状态的任务就是对命令进行解析，通过 try_read_command 方法对 c->rbuf 中存储的命令按照 “\n” 进行基础分割，如果 c->rbuf 中找不到 “\n” 则状态进入 conn_waiting 继续等待后续报文的到来；如果报文完整，将切割出来的命令交给 process_command 进行处理
process_command 命令通过 “\0” 来对单条命令进行详细拆分；而后的 tokenize_command 这个方法就非常重要了，将命令拆解成多个元素（KEY的最大长度250），比如 get 命令将会跳转到 process_get_command 这个方法，而 add、set、append 等 命令将会跳转到 process_update_command 方法；process_*_command 这代表一系列的命令处理逻辑对应的方法
当命令按照对应处理方法运算完毕，会根据具体情况判断是否需要继续读取客户端报文，比如 set 命令就需要再次读取第二行数据，这样进入 conn_nread 状态（继续类似 conn_read 状态的循环）；如果不需要再次读取报文，则进入 conn_write 状态，准备数据写入连接描述符了（这中间还有一个逻辑，如果是简单输出，类似 “END”、”ERROR”、”STORED” 等将直接进入 conn_write，而类似 get 数据这种复杂输出则进入 conn_mwrite 状态）；随后利用 transmit 方法进行数据输出，同样道理，如果发生异常或者失败进入 conn_closing
数据写入客户端成功后，状态机再次进入 conn_new_cmd，进行监听，循环状态

不过还有两个状态 conn_watch、conn_swallow 不是那么常用，我自个也没真正明白其中的做法，有待后续补充~~~

memcached-threads2

如果仅仅是为了了解 memcached 的线程机制，这里就可以结束了，当然需要知道详细实现细节的请往下看...

源码解析

由于有些方法里的代码量比较大，我们这里按照典型的代码片段进行解析，同志们可以根据文章提示的代码位置和代码里面的关键词在源码中搜素，可能数据结构一些元素看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

常用数据结构

CQ_ITEM 主要用于存储用户 socket 连接的基本信息；主线程将用户的 socket 连接信息封装成 CQ_ITEM，并放入工作线程的处理队列中，工作线程得到主线程的 pipe 通知后，就会将队列中的 ITEM 取出来，创建 libevent 的 socket 读事件

/* thread.c: line 27 */
typedef struct conn_queue_item CQ_ITEM;
struct conn_queue_item {
    // socket 的描述符 fd
    int               sfd;
    // 状态机状态
    enum conn_states  init_state;
    // libevent 的监控类型
    int               event_flags;
    // 读取 buf 的长度
    int               read_buffer_size;
    // 传输协议，tcp or udp
    enum network_transport     transport;
    // 消息队列类型，queue_new_conn普通新连接 和 queue_redispatch从边缘线程分派的
    enum conn_queue_item_modes mode;
    // 连接信息指针
    conn *c;
    // 下一个 CQ_ITEM 指针
    CQ_ITEM          *next;
};

CQ 连接处理队列

/* thread.c: line 40 */
/* A connection queue. */
typedef struct conn_queue CQ;
struct conn_queue {
    // 当前队列第一个元素指针
    CQ_ITEM *head;
    // 当前队列最后一个元素指针
    CQ_ITEM *tail;
    // 队列锁
    pthread_mutex_t lock;
};

LIBEVENT_THREAD 每个工作线程的数据结构，每一个工作线程都有有这么一个自己的数据结构，主要存储线程信息、处理队列、pipe信息等

/* memcached.h: line 520 */
typedef struct {
    // 线程ID
    pthread_t thread_id;        /* unique ID of this thread */
    // 线程的 event_base，切记，每个线程都需要有自己的 event_base
    struct event_base *base;    /* libevent handle this thread uses */
    // 异步 event 事件
    struct event notify_event;  /* listen event for notify pipe */
    // pipe管道接收描述符
    int notify_receive_fd;      /* receiving end of notify pipe */
    // pipe管道发送描述符
    int notify_send_fd;         /* sending end of notify pipe */
    // 线程状态
    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 */
#ifdef EXTSTORE
    cache_t *io_cache;          /* IO objects */
    void *storage;              /* data object for storage system */
#endif
    logger *l;                  /* logger buffer */
    void *lru_bump_buf;         /* async LRU bump buffer */
} LIBEVENT_THREAD;

conn 连接信息数据结构，存储某个 socket 连接的所有连接信息数据

/* memcached.h: line 554*/
struct conn {
    // 该conn对应的 socket fd
    int    sfd;
    sasl_conn_t *sasl_conn;
    bool authenticated;
    // 当前状态机状态
    enum conn_states  state;
    enum bin_substates substate;
    // 最后一次处理客户端命令key value的时间
    rel_time_t last_cmd_time;
    // 该conn对应的event
    struct event event;
    // event当前监听的事件类型
    short  ev_flags;
    // 字面意思：触发event回调函数的原因，没看到具体哪里使用，不太明白...
    short  which;   /** which events were just triggered */
    // 读缓冲区，用于存储客户端数据报文中的命令
    char   *rbuf;   /** buffer to read commands into */
    // 有效数据的开始位置，从rbuf到rcurr之间的数据是已经处理的了，变成无效数据了
    char   *rcurr;  /** but if we parsed some already, this is where we stopped */
    // rbuf的大小
    int    rsize;   /** total allocated size of rbuf */
    // 有效数据的长度
    int    rbytes;  /** how much data, starting from rcur, do we have unparsed */
    char   *wbuf;
    char   *wcurr;
    int    wsize;
    int    wbytes;
    // 写完后的下一个状态
    /** which state to go into after finishing current write */
    enum conn_states  write_and_go;
    void   *write_and_free; /** free this memory after finishing writing */
    // 数据直通车 数据部分赋值过程
    char   *ritem;  /** when we read in an item's value, it goes here */
    // 最开始rlbytes为数据部分总长度，当把数据部分填充好后，其值就刚好为0
    int    rlbytes;
    /* data for the nread state */
    /**
     * item is used to hold an item structure created after reading the command
     * line of set/add/replace commands, but before we finished reading the actual
     * data. The data is read into ITEM_data(item) to avoid extra copying.
     */
    // 该链接当前所处理命令的item结构信息
    void   *item;     /* for commands set/add/replace  */
    
    /* data for the swallow state */
    int    sbytes;    /* how many bytes to swallow */
    /* data for the mwrite state */
    // iovec数组指针
    struct iovec *iov;
    // 数组大小
    int    iovsize;   /* number of elements allocated in iov[] */
    // 已经使用的iov数组元素个数
    int    iovused;   /* number of elements used in iov[] */
    // msghdr结构体里面的iovec结构体数组长度是有限制的，为了能传输更多的数据，只能增加msghdr结构体的个数add_msghdr函数负责增加
    struct msghdr *msglist;
    // 数组大小  默认MSG_LIST_INITIAL
    int    msgsize;   /* number of elements allocated in msglist[] */
    // 已经使用了的msghdr元素个数
    int    msgused;   /* number of elements used in msglist[] */
    // 正在用sendmsg函数传输msghdr数组中的哪一个元素
    int    msgcurr;   /* element in msglist[] being transmitted now */
    // msgcurr指向的msghdr总共多少个字节
    int    msgbytes;  /* number of bytes in current msg */
    // worker线程需要占有这个item，直至把item的数据都写回给客户端了故需要一个item指针数组记录本conn占有的item
    item   **ilist;   /* list of items to write out */
    // 数组的大小
    int    isize;
    // 当前使用到的item(在释放占用item时会用到)
    item   **icurr;
    // ilist数组中有多少个item需要释放
    int    ileft;
    char   **suffixlist;
    int    suffixsize;
    char   **suffixcurr;
    int    suffixleft;
#ifdef EXTSTORE
    int io_wrapleft;
    unsigned int recache_counter;
    io_wrap *io_wraplist; /* linked list of io_wraps */
    bool io_queued; /* FIXME: debugging flag */
#endif
    // 根据接收的第一个字符来确定是那种协议，文本ascii_prot 和 二进制binary_prot
    enum protocol protocol;   /* which protocol this connection speaks */
    enum network_transport transport; /* what transport is used by this connection */
    /* data for UDP clients */
    int    request_id; /* Incoming UDP request ID, if this is a UDP "connection" */
    // 客户端IP地址
    struct sockaddr_in6 request_addr; /* udp: Who sent the most recent request */
    socklen_t request_addr_size;
    unsigned char *hdrbuf; /* udp packet headers */
    int    hdrsize;   /* number of headers' worth of space is allocated */
    // 是否不用回复客户端信息
    bool   noreply;   /* True if the reply should not be sent. */
    /* current stats command */
    struct {
        char *buffer;
        size_t size;
        size_t offset;
    } stats;
    /* Binary protocol stuff */
    /* This is where the binary header goes */
    protocol_binary_request_header binary_header;
    uint64_t cas; /* the cas to return */
    short cmd; /* current command being processed */
    int opaque;
    int keylen;
    // 下一个conn的指针
    conn   *next;     /* Used for generating a list of conn structures */
    // 所属worker线程指针
    LIBEVENT_THREAD *thread; /* Pointer to the thread object serving this connection */
};

线程池源码

libevent 基本事件 API

// 创建event_base
main_base = event_init();
// 设置event 监听类型(EV_READ | EV_PERSIST)、回调方法(cb)、回调参数(*arg)
event_set(&ev, listen_fd, EV_READ | EV_PERSIST, cb, *arg);
// 将event与相应的event_base关联
event_base_set(main_base, &cev);
// 将event添加的event_base
event_add(&ev, NULL);
// 启动事件循环，监听事件响应
event_base_loop(main_base, 0);

main 方法

/* memcached.c: line 6504*/
int main (int argc, char **argv) {
    // ....省略了一部分代码
    
    // 初始化一个 event_base
    /* initialize main thread libevent instance */
    main_base = event_init();
    // 日志模块、统计模块、assoc模块、连接模块、slabs存储模块 初始化
    /* initialize other stuff */
    logger_init();
    stats_init();
    assoc_init(settings.hashpower_init);
    conn_init();
    slabs_init(settings.maxbytes, settings.factor, preallocate, use_slab_sizes ? slab_sizes : NULL);
    // ....省略了一部分代码
    
    // 工作线程创建，默认创建 4个工作线程
    memcached_thread_init(settings.num_threads, NULL);
    // ....省略了一部分代码
    // 这边的server_sockets方法主要是socket的bind、listen、accept等操作，主线程主要用于接收客户端的socket连接，然后将连接交给工作线程接管
    if (settings.port && server_sockets(settings.port, tcp_transport, portnumber_file)) {
        vperror("failed to listen on TCP port %d", settings.port);
        exit(EX_OSERR);
    }

主线程的 socket server 主要通过 server_sockets 这个方法创建，而 server_sockets 中主要调用了 server_socket 这个方法

/* memcached.c: line 5796*/
/**
 * Create a socket and bind it to a specific port number
 * @param interface the interface to bind to
 * @param port the port number to bind to
 * @param transport the transport protocol (TCP / UDP)
 * @param portnumber_file A filepointer to write the port numbers to
 *        when they are successfully added to the list of ports we
 *        listen on.
 */
static int server_socket(const char *interface,
                         int port,
                         enum network_transport transport,
                         FILE *portnumber_file) {
                         
    // ....省略了一部分代码
    // 创建一个新的事件，工作线程其实也是调用这个方法，但是区别是这个方法指定了 state 的类型为 conn_listening
    if (!(listen_conn_add = conn_new(sfd, conn_listening,
                                     EV_READ | EV_PERSIST, 1,
                                     transport, main_base))) {
        fprintf(stderr, "failed to create listening connection\n");
        exit(EXIT_FAILURE);
    }
    listen_conn_add->next = listen_conn;
    listen_conn = listen_conn_add;

conn_new 方法

/* memcached.c: line 628*/
// 此方法中又在创建 event，实际上是监听 socket 的读写等事件，主线程主要是监听用户的 socket 连接事件，而工作线程主要监听socket的读写事件，当用户 socket 的连接有数据传递过来的时候，就会调用 event_handler 这个回调函数  
event_set(&c->event, sfd, event_flags, event_handler, (void *)c);
event_base_set(base, &c->event);
c->ev_flags = event_flags;
// 将事件添加到libevent的loop循环中
if (event_add(&c->event, 0) == -1) {
    perror("event_add");
    return NULL;
}

然后我们继续看 event_handler 方法，里面主要是 drive_machine 状态机方法，另外下方涉及到的命令解析、响应数据封装等模块我们后续会专门进行详细解读

/* memcached.c: line 5350*/
static void drive_machine(conn *c) {
    bool stop = false;
    int sfd;
     
    // ....省略了一部分代码
    while (!stop) {
        switch(c->state) {
        case conn_listening:
            addrlen = sizeof(addr);
            // accept 接受客户端的 socket 连接
#ifdef HAVE_ACCEPT4
            if (use_accept4) {
                sfd = accept4(c->sfd, (struct sockaddr *)&addr, &addrlen, SOCK_NONBLOCK);
            } else {
                sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen);
            }
#else
            sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen);
#endif
            // ....省略了一部分代码
            if (settings.maxconns_fast &&
                // ....省略了一部分代码
            } else {
                // 如果客户端用 socket 连接上来，则会调用这个分发逻辑的函数，这个函数会将连接信息分发到某一个工作线程中，然后工作线程接管具体的读写操作
                dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST,
                                     DATA_BUFFER_SIZE, c->transport);
            }
            stop = true;
            break;
        // 继续等待客户端的数据报文到来
        case conn_waiting:
            if (!update_event(c, EV_READ | EV_PERSIST)) {
                if (settings.verbose > 0)
                    fprintf(stderr, "Couldn't update event\n");
                conn_set_state(c, conn_closing);
                break;
            }
            // 等待的过程中，将连接状态设置为读取状态，并且stop设置为true，退出while(stop)的循环
            conn_set_state(c, conn_read);
            stop = true;
            break;
        // 读取数据的事件，当客户端有数据报文上传的时候，就会触发libevent的读事件
        case conn_read:
            // try_read_network 主要读取TCP数据，返回 try_read_result 的枚举类型结构，通过这个枚举类型，来判断是否已经读取到数据，是否读取失败等情况
            res = IS_UDP(c->transport) ? try_read_udp(c) : try_read_network(c);
            switch (res) {
            // 没有读取到数据，继续将事件设置为等待，while(stop)会继续循环，去调用 conn_waiting 这个case
            case READ_NO_DATA_RECEIVED:
                conn_set_state(c, conn_waiting);
                break;
            // 如果有数据读取到了，这个时候就需要调用 conn_parse_cmd 逻辑，conn_parse_cmd主要用来解析读取到的命令
            case READ_DATA_RECEIVED:
                conn_set_state(c, conn_parse_cmd);
                break;
            // 读取失败，直接调用 conn_closing 关闭客户端的连接
            case READ_ERROR:
                conn_set_state(c, conn_closing);
                break;
            case READ_MEMORY_ERROR: /* Failed to allocate more memory */
                /* State already set by try_read_network */
                break;
            }
            break;
        // 解析客户端命令，比如 set username foo
        case conn_parse_cmd :
            // try_read_command 方法用来读取命令，如果这个方法返回为0，则表示解析命令失败（因为TCP粘包拆包的原因，可能命令不完整，需要继续等待数据到来）
            if (try_read_command(c) == 0) {
                /* wee need more data! */
                conn_set_state(c, conn_waiting);
            }
            break;
        }
        // ....省略了一部分代码
    }
    return;
}

dispatch_conn_new 工作线程任务分发方法

/* threads.c: line 467*/
/*
 * Dispatches a new connection to another thread. This is only ever called
 * from the main thread, either during initialization (for UDP) or because
 * of an incoming connection.
 */
void dispatch_conn_new(int sfd, enum conn_states init_state, int event_flags,
                       int read_buffer_size, enum network_transport transport) {
    // 每个连接连上来的时候，都会申请一块CQ_ITEM的内存块，用于存储连接的基本信息
    CQ_ITEM *item = cqi_new();
    char buf[1];
    if (item == NULL) {
        close(sfd);
        /* given that malloc failed this may also fail, but let's try */
        fprintf(stderr, "Failed to allocate memory for connection object\n");
        return ;
    }
    // 这个模块就比较厉害了，主要是通过求余数的方法来得到当前的连接需要哪个线程来接管，而且 last_thread 会记录每次最后一次使用的工作线程，每次记录之后就可以让工作线程进入一个轮询，保证了每个工作线程处理的连接数的平衡 
    int tid = (last_thread + 1) % settings.num_threads;
    // 获取指定线程
    LIBEVENT_THREAD *thread = threads + tid;
    last_thread = tid;
    item->sfd = sfd;
    item->init_state = init_state;
    item->event_flags = event_flags;
    item->read_buffer_size = read_buffer_size;
    item->transport = transport;
    item->mode = queue_new_conn;
    // 向对应工作线程的队列中放入CQ_ITEM
    cq_push(thread->new_conn_queue, item);
    MEMCACHED_CONN_DISPATCH(sfd, thread->thread_id);
    buf[0] = 'c';
    // 向对应工作线程的 pipe 中写入1，工作线程监听到 pipe 中有写入数据，就会向 thread->new_conn_queue 队列中 pop 出一个 item，然后进行连接的接收，同时该工作线程进入 状态机的conn_new_cmd状态
    if (write(thread->notify_send_fd, buf, 1) != 1) {
        perror("Writing to thread notify pipe");
    }
}

本文作者： wettper
本文链接： http://www.web-lovers.com/memcached-source-threads.html
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！