Memcached的线程模型及状态机

-
 
 

Han 通过 Google 阅读器发送给您的内容：

 
 

于 11-9-4 通过 basic coder 作者：levin

Memcached是一种应用较广泛的分布式内存对象缓存系统，应用之余总想了解它的实现机理，这也就是开源的好处，以至于每接触一款优秀的开源软件都有去阅读它源代码的冲动，Memcached-1.4.7的代码量还是可以接受的，只有10K行左右，我比较关心的两个方面还是它的进程（线程）管理机制和内存管理机制，这里先简单写一下我对Memcached进程管理方面的理解。

Memcached使用libevent实现事件循环，libevent在Linux环境下默认采用epoll作为IO多路复用方法，这个不重要，接下来要讨论的是Memcached的进程管理模型。

Memcached采用了很典型的Master-Worker模型，采用的是多线程而不是多进程，而线程之间没有冗余的共享数据，这样便降低了多线程进行线程同步的开销，核心的共享数据是消息队列，主线程会把收到的事件请求放入队列，随后调度程序会选择一个空闲的Worker线程来从队列中取出事件请求进行处理。

在main()函数里面，Memcached为主线程调用event_init()创建了一个libevent base对象，随后调用thread_init()来初始化线程，我们来看下这个函数的实现。

void thread_init(int nthreads, struct event_base *main_base) {     int         i;       pthread_mutex_init(&cache_lock, NULL);     pthread_mutex_init(&stats_lock, NULL);       pthread_mutex_init(&init_lock, NULL);     pthread_cond_init(&init_cond, NULL);       pthread_mutex_init(&cqi_freelist_lock, NULL);     cqi_freelist = NULL;       threads = calloc(nthreads, sizeof(LIBEVENT_THREAD));     if (! threads) {         perror("Can't allocate thread descriptors");         exit(1);     }       dispatcher_thread.base = main_base;     dispatcher_thread.thread_id = pthread_self();       for (i = 0; i < nthreads; i++) {         int fds[2];         if (pipe(fds)) {             perror("Can't create notify pipe");             exit(1);         }           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]);     }       /* Wait for all the threads to set themselves up before returning. */     pthread_mutex_lock(&init_lock);     while (init_count < nthreads) {         pthread_cond_wait(&init_cond, &init_lock);     }     pthread_mutex_unlock(&init_lock); }

代码很简单，初始化几个全局锁和一个条件变量，init_lock和init_cond这一对锁/条件变量的作用很明显，它们的作用是用来等待所有的worker线程启动完毕后，thread_init()才可以继续执行，见该函数最后的几句，非常通用的用法，等待init_count数达到预定的线程数后主线程方可继续执行，否则就一直在wait，每创建一个worker线程就会让init_count的值加1，创建worker线程的工作是在setup_thread()函数中进行的，这个后面讨论。

thread_init()在初始化完全局锁并为线程池分配分配空间之后便开始对线程池中的每个线程进行初始化，哦，在这之前把全局的调度线程设置成为主线程，OK，接下来就开始遍历线程池，每个线程都有一对notify_fd，它们通过管道连接，这个管道便是主线程对线程池中的工作线程进行调度的接口，看一下setup_thread()这个函数的实现（节省版面，只保留了关键代码）。

static void setup_thread(LIBEVENT_THREAD *me) {     me->base = event_init();     if (! me->base) {         fprintf(stderr, "Can't allocate event base\n");         exit(1);     }       /* 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) {         fprintf(stderr, "Can't monitor libevent notify pipe\n");         exit(1);     }       me->new_conn_queue = malloc(sizeof(struct conn_queue));     if (me->new_conn_queue == NULL) {         perror("Failed to allocate memory for connection queue");         exit(EXIT_FAILURE);     }     cq_init(me->new_conn_queue); }

首先调用event_init()为该线程初始化一个事件池，我们知道libevent的事件池（event base）不是线程安全的，所以每个线程需要有自己的事件池，所有的IO事件都由这个事件池来处理，在这个函数里面，首先将notify_receive_fd添加到事件监听循环中，设置回调函数thread_libevent_process()，当主线程向该worker线程通过pipe IPC发送消息时，便会触发该worker线程执行thread_libevent_process()函数。随后setup_thread()函数创建connection队列并将其初始化。

接着回过头看thread_init()函数，初始化线程池结束以后，所需要做的工作便是为线程池创建对应的线程，也就是调用了create_worker()函数，这个函数只不过把pthread_create()做了一个简单的封装，至少到目前为止里面没有什么特别的代码，线程的执行函数是worker_libevent()，该函数将全局线程数加一，随后便调用event_base_loop()，完全将事件循环交给了libevent，当然到目前为止libevent的事件池中只有用来进行IPC的管道文件描述符（说到这里，当时我还在奇怪呢，代码执行这里还并没有添加socket描述符到事件池里面，那event_loop岂不一执行就结束了，后来才反应过来里面有IPC管道文件描述符了），那么这些描述符是在什么时候添加的呢？刚才已经讨论过了，是在setup_thread()函数里面，它的事件回调函数是thread_libevent_process()，OK，接下来我们看看这个函数吧。

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)         if (settings.verbose > 0)             fprintf(stderr, "Can't read from libevent pipe\n");       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);         if (c == NULL) {             close(item->sfd);         } else {             c->thread = me;         }         cqi_free(item);     } }

函数的一开始便从fd中读取一个字节，不在乎这一个字节是什么，只是主线程发过来的一个单字节（实际上是个\0），用来触发fd的READ事件，主线程通过往worker线程的notify_write_fd中写入一个单字节来实现对worker线程的调度，因此这个函数里面首先要先把这一个字节读取出来，否则会影响以后的事件循环。紧接着从connection队列中弹出一个item，然后根据这个item的信息再创建一个connection，这里的conn还是不要理解为连接的好，理解成任务更好一些，conn对象中有读写缓冲区，状态（用来实现Memcached的有限状态机）等，这conn_new()这个函数里面会创建一个conn对象并对它进行初始化，但最重要的操作是设置对socketfd的事件监听函数，event_handler()，这个函数又调用drive_machine()，其中就实现了Memcached的有限状态机，通过对state的不同值来进行不同的操作，具体的状态如下：

const char* const statenames[] = { "conn_listening", 						      "conn_new_cmd", 						        "conn_waiting", 							"conn_read", 							"conn_parse_cmd", 							"conn_write", 							"conn_nread", 							"conn_swallow", 							"conn_closing", 							"conn_mwrite" };

关于状态机是怎么运行的就不讨论了，代码篇幅过长，涉及到的细节太多，不展开说了，要说的一点是状态机在哪里进入的conn_listening状态，这是某个socketfd的起始状态，socket需要先监听然后accept，再之后才可以进行read,write等操作，listen必然会在程序初始化的过程中调用，我们可以看到在main()函数里面线程初始化结束以后，便开始对socket描述符进行初始化，获取系统存在几个可用的地址信息，然后创建socket描述符，再然后bind()，再之后调用listen()，这些操作都是在server_socket()函数中执行的，该函数初始化监听socket，并针对服务器做一些socket options的设置，最后最关键的一步是针对该socket调用了conn_new()，但事件池是main_base，也就是说accept()事件是由主线程接收到的，accept()是发生在主线程内，这样也就避免了多线程accept()的惊群，accept()函数的执行也是在状态机循环中执行的，drive_machine()函数中，accept()完成后，主线程便调用dispatch_conn_new()对该描述符进行调度，状态机的代码太长，不列出来了，看一看dispatch_conn_machine()的代码吧。

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();     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;       cq_push(thread->new_conn_queue, item);       MEMCACHED_CONN_DISPATCH(sfd, thread->thread_id);     if (write(thread->notify_send_fd, "", 1) != 1) {         perror("Writing to thread notify pipe");     } }

创建一个item并压入队列，然后通过一个简单的哈希从线程池中找出一个线程，写一个空字符来激活这个线程（前面讨论过了），接着被激活的线程就会从conn_queue中弹出一个item来进行处理。

OK，关于Memcached的线程模型要说的就这些吧，我自己读源码时对它的浅显的理解，欢迎大家批评讨论。


 
 

可从此处完成的操作：

 
 

评论

发表评论

此博客中的热门博文

浅析Linux Kernel 哈希路由表实现(一)

-
    Han 通过 Google 阅读器发送给您的内容：     浅析Linux Kernel 哈希路由表实现(一) 于 11-3-13 通过 basic coder 作者：levin 1. 路由表 目前Linux内核中支持两种路由表，一种是Hash路由表，另一种是Trie路由表，Trie算法查找效率很高，但它也因极其消毫内存资源而闻名，因此一般用在大型机上，否则在路由项过多时很可能造成内存的耗尽。在一般的机器上最好还是使用Hash路由表，之前有争论说Linux使用Hash路由表相比于二叉树效率太低，不适合商用，其实只要设计良好的哈希算法，尽量减少哈希碰撞，Hash路由表的查找效率也是很高的，在最好的情况下算法复杂算可以达到O(1)，当然，最差也不过是O(n)，我们有理由相信Linux中存在各种优秀的哈希算法，这些都是值得我们学习的。 Linux内核中IPv4路由表采用了分层的结构，第一层是是fib_table，表示路由表，最多可以支持256张路由表，用户可以根据策略路由的需求来创建自己的路由表，系统默认的两张路由表为RT_TABLE_LOCAL和RT_TABLE_MAIN。下面是系统保留的路由表标识符： enum rt_class_t { RT_TABLE_UNSPEC = 0 , /* User defined values */ RT_TABLE_COMPAT = 252 , RT_TABLE_DEFAULT = 253 , RT_TABLE_MAIN = 254 , RT_TABLE_LOCAL = 255 , RT_TABLE_MAX = 0xFFFFFFFF } ; 下面是路由表的定义： struct fib_table { struct hlist_node tb_hlist ; /* 路由表的标识，即为上面提到的类型 */ u32 tb_id ; /* 该路由中默认路由条目在哈希路由表中的索...
阅读全文

通过blktrace, debugfs分析磁盘IO

-
    Han 通过 Google 阅读器发送给您的内容：     通过blktrace, debugfs分析磁盘IO 于 12-11-14 通过 Linux Kernel and Tao 作者：admin 前几天微博上有同学问我磁盘util达到了100%时程序性能下降的问题，由于信息实在有限，我也没有办法帮太大的忙，这篇blog只是想给他列一下在磁盘util很高的时候如何通过blktrace+debugfs找到发生IO的文件，然后再结合自己的应用程序，分析出这些IO到底是谁产生的，最终目的当然是尽量减少不必要的IO干扰，提高程序的性能。 blktrace是Jens Axobe写的一个跟踪IO请求的工具，Linux系统发起的IO请求都可以通过blktrace捕获并分析，关于这个工具的介绍请自行google之，这里推荐我们部门的褚霸同学的 blog ，里面有好几篇文章分别介绍了blktrace, blkparse以及blkiomon等工具的使用。 debugfs是ext2, ext3, ext4文件系统提供的文件系统访问工具，通过它我们可以不通过mount文件系统而直接访问文件系统的内容，它是e2fsprogs的一部分，默认应该都是安装的，详细的说明可以通过man debugfs得到。 下面我来演示一下如何通过这两个工具的配合来找到磁盘IO的源头。 先看一个简单的例子： 在一个终端会输入如下命令： while [ 1 ];do dd if=/dev/zero of=test_file bs=4k count=20 seek=$RANDOM oflag=sync;done 随机的在test_file里面写数据造成较大的IO压力，现在看看如何通过blktrace和debugfs抓到它。 1. 通过iostat观察到有很大的磁盘压力 Device: rrqm/s wrqm/s r/s w/s rsec/s wsec/s avgrq-sz avgqu-sz await svctm %util sdb ...
阅读全文

ext4 bigalloc 答疑

-
    Han 通过 Google 阅读器发送给您的内容：     ext4 bigalloc 答疑 于 13-2-20 通过 斯巴达第二季 问题都来自 pamirs 的信，感谢他对bigalloc特性的关注。 问：文件分配的时候，ext4文件系统请求的时候是按照blocks来请求,但是ext4文件系统自己会按照cluster来分配（这个的理解是否正确呢？） 答：应该说，不太准确，事实上, ext4文件系统请求分配磁盘空间的时候也是按照cluster来请求的了，ext4_ext_map_blocks() 函数是用来检查文件的逻辑位置对应的物理块的，里面申请新磁盘空间的调用是：         ar.len = EXT4_NUM_B2C(sbi, offset+allocated);         ......         newblock = ext4_mb_new_blocks(handle, &ar, &err); ar.len的单位是cluster，所以实际上申请空间就是以cluster为单位 问：如果cluster为32K(8个blocks), 那么对于一个3k的文件来说,一次就分配了一个cluster,这样就有7个blocks不能在被其他文件使用了。group desc中的block_bitmap中将cluster对应的bit位置1，那么后续这个文件在增加的时候,是否按照增加的blocks数量,以及当前该文件占用的cluster,计算出已经占用的cluster中有多少blocks是空闲的,还可以继续分配呢? 答：是的，需要一些琐碎的计算，这些计算的代码都放在了函数get_implied_cluster_alloc()里 问：ext4_group_desc中的bg_free_blocks_count_lo字段表示当前group中有多少空闲的blocks,使用了bigalloc之后,该字段的意义并没有发生变化吧?好像没有看到有新增一个字段来表示free_clusters数量,这个是否不需要保存这样的变量?通过bitmap查询得知呢? 答：bg_free_blocks_coun...
阅读全文