memcached 线程模型如下图所示
memcached 线程可分为两种，一是负责基本事件处理(如启动时间更新，连接请求等)和内存管理等的线程， 二是负责网络读写请求处理的线程即 worker threads 。这里只对 worker threads 进行讨论。
在分析之前，先看一下相关的数据结构和变量定义:
1. 连接队列(thread.c)
/* An item in the connection queue. */ typedef struct conn_queue_item CQ_ITEM; struct conn_queue_item { int sfd; // 一个连接对应的 socket fd enum conn_states init_state; int event_flags; // 网络事件对应的 event flags, libevent概念 int read_buffer_size; enum network_transport transport; // 传输协议: local_transport(Unix sockets)、 tcp_transport 或 udp_transport CQ_ITEM *next; // 链表 下一项指针, 用于链接在连接队列中的下一项或空闲项列表中的下一项 }; /* A connection queue. */ typedef struct conn_queue CQ; // 连接队列， 实际上是一个CQ_ITEM的单链表 struct conn_queue { CQ_ITEM *head; CQ_ITEM *tail; pthread_mutex_t lock; // 访问锁 };
2. 线程信息(memcached.h)
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 */ // 用于监听管道事件 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; // 用于描述 worker thread typedef struct { pthread_t thread_id; /* unique ID of this thread */ struct event_base *base; /* libevent handle this thread uses */ } LIBEVENT_DISPATCHER_THREAD; // 用于描述 diapatcher thread
3. 变量定义(thread.c)
/* Locks for cache LRU operations */ pthread_mutex_t lru_locks[POWER_LARGEST]; // 用于 lru cache 的锁 /* Connection lock around accepting new connections */ pthread_mutex_t conn_lock = PTHREAD_MUTEX_INITIALIZER; // 用于接受新连接的锁 /* Lock for global stats */ static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER; // 用于收集统计信息的锁 /* Lock to cause worker threads to hang up after being woken */ static pthread_mutex_t worker_hang_lock; // 用于线程同步的锁， 具体在下面会有交代 /* Free list of CQ_ITEM structs */ static CQ_ITEM *cqi_freelist; // CQ_ITEM空闲项列表 static pthread_mutex_t cqi_freelist_lock; // 相应的访问锁 static pthread_mutex_t *item_locks; // 数组，存放用于访问 hash 表的锁 /* size of the item lock hash table */ static uint32_t item_lock_count; // 上述数组大小 unsigned int item_lock_hashpower; // 上述数组大小以2为底的对数， 即 item_lock_count = 2**item_lock_hashpower #define hashsize(n) ((unsigned long int)1<<(n)) // 2 ** n #define hashmask(n) (hashsize(n)-1) // 2 ** n - 1 static LIBEVENT_DISPATCHER_THREAD dispatcher_thread; // dispatcher 描述 /* * Each libevent instance has a wakeup pipe, which other threads * can use to signal that they've put a new connection on its queue. */ static LIBEVENT_THREAD *threads; // worker threads 描述 /* * Number of worker threads that have finished setting themselves up. */ static int init_count = 0; // 已启动或重启的线程的数目 static pthread_mutex_t init_lock; // 用于启动或重启线程的锁和条件变量 static pthread_cond_t init_cond;
我们从 main 函数开始分析：
int main(int argc, char* argv[]) { // in memcached.c
// ...
/* start up worker threads if MT mode */
memcached_thread_init(settings.num_threads, main_base);
// ...
}
其中settings.num_threads 是 worker threads 的数目， 默认值为 4， 可通过 -t 选项指定， 通常这个值不应超过机器处理器核的数目。
main_base 是一个指向 event_base 结构的指针， 用于网络事件处理(libevent)。
下面贴出 memcached_thread_init 的定义:
/* * Initializes the thread subsystem, creating various worker threads. * * nthreads Number of worker event handler threads to spawn * main_base Event base for main thread */ void memcached_thread_init(int nthreads, struct event_base *main_base) { int i; // 用于循环 int power; // 指数表示的 item lock table 大小， 实际大小为 2**power。 // item lock table 是一个数组， 保存的是用来访问 hash table 的锁。 // 初始化锁以及条件变量 for (i = 0; i < POWER_LARGEST; i++) { pthread_mutex_init(&lru_locks[i], NULL); } pthread_mutex_init(&worker_hang_lock, NULL); pthread_mutex_init(&init_lock, NULL); pthread_cond_init(&init_cond, NULL); pthread_mutex_init(&cqi_freelist_lock, NULL); cqi_freelist = NULL; // 初始化 cqi_freelist /* Want a wide lock table, but don't waste memory */ if (nthreads < 3) { // 0 1 2 power = 10; } else if (nthreads < 4) { // 3 power = 11; } else if (nthreads < 5) { // 4 power = 12; } else { /* 8192 buckets, and central locks don't scale much past 5 threads */ power = 13; } // 为了保证线程安全性， 在访问 hash 表时， 需要对当前访问的链（chain）加锁 // 为了提升访问效率的同时又不浪费过多的存储空间， 要求锁的数目尽可能多但不允许其超过 hash 表的 slots 数 if (power >= hashpower) { fprintf(stderr, "Hash table power size (%d) cannot be equal to or less than item lock table (%d) ", hashpower, power); fprintf(stderr, "Item lock table grows with `-t N` (worker threadcount) "); fprintf(stderr, "Hash table grows with `-o hashpower=N` "); exit(1); } item_lock_count = hashsize(power); // 2 ** power item_lock_hashpower = power; // 初始化 item_locks item_locks = calloc(item_lock_count, sizeof(pthread_mutex_t)); if (! item_locks) { perror("Can't allocate item locks"); exit(1); } for (i = 0; i < item_lock_count; i++) { pthread_mutex_init(&item_locks[i], NULL); } // 为 threads 数组分配空间 threads = calloc(nthreads, sizeof(LIBEVENT_THREAD)); if (! threads) { perror("Can't allocate thread descriptors"); exit(1); } // 初始化 dispatcher_thread dispatcher_thread.base = main_base; // dispatcher thread 即主线程 dispatcher_thread.thread_id = pthread_self(); // 初始化用于线程通信的管道 pipe 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]); // 初始化 thread[i] 的其他字段 /* Reserve three fds for the libevent base, and two for the pipe */ stats.reserved_fds += 5; } /* Create threads after we've done all the libevent setup. */ for (i = 0; i < nthreads; i++) { // 创建线程 pthread_create create_worker(worker_libevent, &threads[i]); } /* Wait for all the threads to set themselves up before returning. */ pthread_mutex_lock(&init_lock); wait_for_thread_registration(nthreads); // 等待所有线程创建完成，利用的是条件变量 pthread_mutex_unlock(&init_lock); }
setup_thread:
/* * Set up a thread's information. */ static void setup_thread(LIBEVENT_THREAD *me) { me->base = event_init(); // 初始化 libevent handle （或 event base） if (! me->base) { fprintf(stderr, "Can't allocate event base "); 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 "); 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); if (pthread_mutex_init(&me->stats.mutex, NULL) != 0) { perror("Failed to initialize mutex"); exit(EXIT_FAILURE); } me->suffix_cache = cache_create("suffix", SUFFIX_SIZE, sizeof(char*), NULL, NULL); if (me->suffix_cache == NULL) { fprintf(stderr, "Failed to create suffix cache "); exit(EXIT_FAILURE); } }
create_worker:
static void create_worker(void *(*func)(void *), void *arg) { // 线程创建 pthread_t thread; pthread_attr_t attr; int ret; pthread_attr_init(&attr); if ((ret = pthread_create(&thread, &attr, func, arg)) != 0) { fprintf(stderr, "Can't create thread: %s ", strerror(ret)); exit(1); } }
wait_for_thread_registration:
static void wait_for_thread_registration(int nthreads) { while (init_count < nthreads) { pthread_cond_wait(&init_cond, &init_lock); } }
在执行 wait_for_thread_registration 时， 我们已经持有锁 init_lock。 每初始化完成一个thread，init_count 都会自增1(这个操作需要保证其原子性， 利用init_cond加锁) 并且调用 pthread_cond_signal(&init_cond)，这些都是在 register_thread_initialized (in thread.c) 中完成的。
调用栈: worker_libevent()->register_thread_initialized()->pthread_cond_signal()， 其中 worker_libevent 是 worker 线程的主方法， 其除了调用 register_thread_initialized() 外， 还启动了当前线程的事件处理循环。源代码如下：
/* * Worker thread: main event loop */ static void *worker_libevent(void *arg) { LIBEVENT_THREAD *me = arg; /* Any per-thread setup can happen here; memcached_thread_init() will block until * all threads have finished initializing. */ register_thread_initialized(); event_base_loop(me->base, 0); return NULL; }
看一下 thread_libevent_process (管道读事件处理器 notify event handler)：
/* * Processes an incoming "handle a new connection" item. This is called when * input arrives on the libevent wakeup pipe. */ 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) // 读取消息 c 或 p if (settings.verbose > 0) fprintf(stderr, "Can't read from libevent pipe "); 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); if (c == NULL) { if (IS_UDP(item->transport)) { fprintf(stderr, "Can't listen for events on UDP socket "); exit(1); } else { if (settings.verbose > 0) { fprintf(stderr, "Can't listen for events on fd %d ", item->sfd); } close(item->sfd); } } else { c->thread = me; } cqi_free(item); } break; /* we were told to pause and report in */ case 'p': // 暂停当前线程 register_thread_initialized(); // 这时 worker_hang_lock 已被其他线程持有， 调用后因在该锁上执行lock操作暂停 break; } }
conn_new (in memcached.c) 初始化连接状态信息。主线程接收到连接请求后就会调用 dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST, DATA_BUFFER_SIZE, tcp_transport)(这里使用的传输层协议是TCP), dispatch_conn_new 选择一个线程处理连接的网络事件(其实只有读事件)。
/* Which thread we assigned a connection to most recently. */ static int last_thread = -1; // 记录最近处理新连接的线程的索引 (index) /* * 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 *item = cqi_new(); // 分配一个新的 CQ_ITEM 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 "); return ; } int tid = (last_thread + 1) % settings.num_threads; // 计算处理这个新连接的线程的 index LIBEVENT_THREAD *thread = threads + tid; // 获取相应的线程描述信息 last_thread = tid; // 更新 last_thread // 初始化 item 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); // 将 item 添加到线程的连接队列 MEMCACHED_CONN_DISPATCH(sfd, thread->thread_id); buf[0] = 'c'; if (write(thread->notify_send_fd, buf, 1) != 1) { // 通知相应的线程处理连接 perror("Writing to thread notify pipe"); } }
write(thread->notify_send_fd, buf, 1) 成功执行后， 线程会监听到管道读事件, 从管道读取消息 "c", 对连接队列中的新连接进行处理， 源代码见 thread_libevent_process() (管道读事件处理器, 或通知处理器)。
这篇文章到此为止。 上述代码涉及的其他细节如 CQ_ITEM 的分配和释放等，这里不再作详细说明。至于 hash 表 和 网络通信 会分别在内存管理(TODO : link)和网络通信(TODO : link)这两节给出比较详尽的描述。