分布式缓存系统 Memcached 工作线程初始化

Memcached采用典型的Master-Worker模式，其核心思想是：有Master和Worker两类进程（线程）协同工作，Master进程负责接收和分配任务，Worker进程负责处理子任务。当各Worker进程将各个子任务处理完成后，将结果返回给Master进程，由Master进程做归纳和汇总。

工作示意图如下所示：



其中每个工作线程维护一个连接队列，以接收由主线程分配的客户连接；同时每个工作线程维护一个Libevent实例，以处理和主线程间的管道通信以及和客户连接间的socket通信事件。

工作线程的创建与初始化等工作由函数void thread_init(int nthreads, struct event_base *main_base)

调用相应函数来完成。

//Memcached内部工作线程的封装
typedef struct {
pthread_t thread_id; //线程ID
struct event_base *base; //libevent的不是线程安全的，每个工作线程持有一个libevent实例，用于pipe管道通信和socket通信
struct event notify_event; //用于监听pipe管道的libevent事件
int notify_receive_fd; //接收pipe管道消息描述符
int notify_send_fd; //发送pipe管道消息描述符
struct thread_stats stats; //每个线程对应的统计信息
struct conn_queue *new_conn_queue; //每个线程都有一个工作队列，主线程接受的连接，挂载到该消息队列中
cache_t *suffix_cache; //后缀cache
uint8_t item_lock_type; //线程操作hash表持有的锁类型，有局部锁和全局锁
} LIBEVENT_THREAD;

//分发线程的封装
typedef struct {
pthread_t thread_id; //线程id
struct event_base *base; //libevent实例
} LIBEVENT_DISPATCHER_THREAD;

//工作线程绑定到libevent实例
static void setup_thread(LIBEVENT_THREAD *me) {
me->base = event_init();//创建libevent实例
if (! me->base) {
fprintf(stderr, "Can't allocate event base\n");
exit(1);
}

//工作线程的初始化:对应liibevent实例的创建与设置，绑定到该libevent实例、注册事件、创建活动连接队列、创建设置线程，分别执行libevent事件循环：Memcached采用了典型的Master-Worker的线程模式，
//Master就是由main线程来充当，而Worker线程则是通过Pthread创建的。
void thread_init(int nthreads, struct event_base *main_base) {//该函数调用其他函数来完成线程创建、初始化等工作。
int i;
int power;

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;

/* Want a wide lock table, but don't waste memory */
if (nthreads < 3) {
power = 10;
} else if (nthreads < 4) {
power = 11;
} else if (nthreads < 5) {
power = 12;
} else {
/* 8192 buckets, and central locks don't scale much past 5 threads */
power = 13;
}

item_lock_count = hashsize(power);
item_lock_hashpower = power;

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);
}
pthread_key_create(&item_lock_type_key, NULL);
pthread_mutex_init(&item_global_lock, 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();
//工作线程的初始化,工作线程和主线程(main线程)是通过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]);//添加工作线程到lievent中
/* 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++) {
create_worker(worker_libevent, &threads[i]);//创建线程，执行libevnt事件循环
}

/* 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);
}

/*
* Set up a thread's information.
*/
//工作线程绑定到libevent实例，注册该线程的通知事件，创建并初始化其消息队列（即就绪事件队列）
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对象：设置事件ev绑定文件描述符或信号值fd（定时事件设为-1）；设置事件类型；设置回调函数及参数
event_set(&me->notify_event, me->notify_receive_fd,
EV_READ | EV_PERSIST, thread_libevent_process, me);
//设置事件ev从属的event_base，以及ev的优先级
event_base_set(me->base, &me->notify_event);

//往事件队列中注册事件处理器，并将对应事件添加到事件多路分发器上
//内部调用event_add_internal函数
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);//初始化连接队列（单链表）
//创建线程的后缀cache？？
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\n");
exit(EXIT_FAILURE);
}
}

//创建工作线程
static void create_worker(void *(*func)(void *), void *arg) {
pthread_t thread;
pthread_attr_t attr;
int ret;

pthread_attr_init(&attr);//Posix线程部分，线程属性初始化

//创建线程：外部传入的线程函数worker_libevent
if ((ret = pthread_create(&thread, &attr, func, arg)) != 0) {
fprintf(stderr, "Can't create thread: %s\n",
strerror(ret));
exit(1);
}
}

//线程处理函数 ：执行libevent事件循环（处理与主线程的通信事件和所分配到的客户连接事件的处理）

static void *worker_libevent(void *arg) {
LIBEVENT_THREAD *me = arg;

/* Any per-thread setup can happen here; thread_init() will block until
* all threads have finished initializing.
*/

/* set an indexable thread-specific memory item for the lock type.
* this could be unnecessary if we pass the conn *c struct through
* all item_lock calls...
*/
//默认的hash表的锁为局部锁
me->item_lock_type = ITEM_LOCK_GRANULAR;
pthread_setspecific(item_lock_type_key, &me->item_lock_type);//设定线程的属性
//用于控制工作线程初始化，通过条件变量来控制
register_thread_initialized();
//工作线程的libevent实例启动
event_base_loop(me->base, 0);
return NULL;
}

其中Libevent的事件循环函数event_baseloop:(代码来自Libevent框架源码)，其处理流程如图：



具体代码如下：

//事件处理主循环：调用I/O事件多路分发器监听函数，以等待事件发生，
//当有事件发生时将其插入到活动事件队列数组对应优先级的队列中，然后调用其事件的回调函数依次处理事件。
int
event_base_loop(struct event_base *base, int flags)
{
const struct eventop *evsel = base->evsel;
struct timeval tv;
struct timeval *tv_p;
int res, done, retval = 0;

/* Grab the lock. We will release it inside evsel.dispatch, and again
* as we invoke user callbacks. */
EVBASE_ACQUIRE_LOCK(base, th_base_lock);//获取对event_base的独占锁

//如果事件循环已经启动，则不能再启动
if (base->running_loop) {
event_warnx("%s: reentrant invocation. Only one event_base_loop"
" can run on each event_base at once.", __func__);
EVBASE_RELEASE_LOCK(base, th_base_lock);//释放对event_base的独占锁
return -1;
}

base->running_loop = 1;//标记循环已经启动

clear_time_cache(base);//清除系统时间缓存

if (base->sig.ev_signal_added && base->sig.ev_n_signals_added)
evsig_set_base(base);//设置信号处理器的event_base实例

done = 0;

#ifndef _EVENT_DISABLE_THREAD_SUPPORT
base->th_owner_id = EVTHREAD_GET_ID(); //不支持多线程时
#endif
//多线程
base->event_gotterm = base->event_break = 0;

while (!done) {
base->event_continue = 0;

/* Terminate the loop if we have been asked to */
if (base->event_gotterm) {
break;
}

if (base->event_break) {
break;
}

// 校正系统时间，如果系统支持monotonic时间，该时间是系统从boot后到现在所经过的时间，因此不需要执行校正。
//如果系统使用的是非MONOTONIC时间，用户可能会向后调整了系统时间
// 在timeout_correct函数里，比较last wait time和当前时间，如果当前时间< last wait time
// 表明时间有问题，这是需要更新timer_heap中所有定时事件的超时时间。
timeout_correct(base, &tv);

tv_p = &tv;
// 根据timer heap中事件的最小超时时间，设置系统I/O demultiplexer的最大等待时间
if (!N_ACTIVE_CALLBACKS(base) && !(flags & EVLOOP_NONBLOCK)) {//没有就绪事件则设置最大等待时间
//如果就绪事件为0，且、、
timeout_next(base, &tv_p);////获取时间堆中超时时间最小的事件处理器的,计算等待时间，返回值存在tv_p中
//超时时间作为I/O分发器的超时返回时间，这样当I/O分发器返回时即可以去处理
//事件堆上就绪的定时事件
} else {//有事件就绪了，就直接返回
/*
* if we have active events, we just poll new events
* without waiting.
*/
//如果有就绪事件尚未处理，则将I/O复用系统调用的超时时间设置为0，
//这样I/O复用系统调用将直接返回，程序即立即处理就绪事件
evutil_timerclear(&tv);//(tvp)->tv_sec = (tvp)->tv_usec = 0
}

/* If we have no events, we just exit */
//如果event_base中没有注册任何事件，则直接退出
if (!event_haveevents(base) && !N_ACTIVE_CALLBACKS(base)) {
event_debug(("%s: no events registered.", __func__));
retval = 1;
goto done;
}

/* update last old time */
gettime(base, &base->event_tv);//更新base->event_tv为缓存时间（没清空时）或系统当前那时间

clear_time_cache(base);//清除系统时间缓存

//调用后端i/o复用机制的事件多路分发器的dispath函数等待事件，将就绪信号事件，I/O事件插入到活动事件队列中
res = evsel->dispatch(base, tv_p);//阻塞调用：如epoll_wait

if (res == -1) {
event_debug(("%s: dispatch returned unsuccessfully.",
__func__));
retval = -1;
goto done;
}

update_time_cache(base);//更新时间缓存为当前系统时间（缓存时间记录了本次事件就绪返回的时间）
// 检查heap中的timer events，将就绪的timer event从heap上删除，并插入到激活链表中
timeout_process(base);
// 调用event_process_active()处理激活链表中的就绪event，调用其回调函数执行事件处理
// 该函数会寻找最高优先级（priority值越小优先级越高）的激活事件链表，
// 然后处理链表中的所有就绪事件；
// 因此低优先级的就绪事件可能得不到及时处理；
if (N_ACTIVE_CALLBACKS(base)) {//有就绪事件就绪了
int n = event_process_active(base);//处理就绪事件
if ((flags & EVLOOP_ONCE)
&& N_ACTIVE_CALLBACKS(base) == 0
&& n != 0)
done = 1;
} else if (flags & EVLOOP_NONBLOCK)
done = 1;
}
event_debug(("%s: asked to terminate loop.", __func__));

done:
// 循环结束，清空时间缓存
clear_time_cache(base);
base->running_loop = 0;//设置停止循环标志

EVBASE_RELEASE_LOCK(base, th_base_lock);//释放锁

return (retval);
}

//阻塞工作线程
static void wait_for_thread_registration(int nthreads) {
while (init_count < nthreads) {
pthread_cond_wait(&init_cond, &init_lock);//在条件变量init_cond上面阻塞，阻塞个数为nthreads-init_count
}
}
//唤醒工作线程
static void register_thread_initialized(void) {
pthread_mutex_lock(&init_lock);
init_count++;
pthread_cond_signal(&init_cond);
pthread_mutex_unlock(&init_lock);
}

//每个线程持有的统计信息
struct thread_stats {
pthread_mutex_t mutex;
uint64_t get_cmds;
uint64_t get_misses;
uint64_t touch_cmds;
uint64_t touch_misses;
uint64_t delete_misses;
uint64_t incr_misses;
uint64_t decr_misses;
uint64_t cas_misses;
uint64_t bytes_read;
uint64_t bytes_written;
uint64_t flush_cmds;
uint64_t conn_yields; /* # of yields for connections (-R option)*/
uint64_t auth_cmds;
uint64_t auth_errors;
struct slab_stats slab_stats[MAX_NUMBER_OF_SLAB_CLASSES];
};

本节主要分析了工作线程的初始化部分，至于具体的主线程任务分发调度，以及工作线程如何具体处理连接和与主线程之间的通信等后续讲解。