之前用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
感谢以上作者的无私分享。