memcachedd事件模型

/* start up worker threads if MT mode */
thread_init(settings.num_threads, main_base);

typedef struct {
    pthread_t thread_id;        /* unique ID of this thread 线程ID*/
    struct event_base *base;    /* libevent handle this thread uses libevent事件*/
    struct event notify_event;  /* listen event for notify pipe 注册事件*/
    int notify_receive_fd;      /* receiving end of notify 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 */
    uint8_t item_lock_type;     /* use fine-grained or global item lock */
} LIBEVENT_THREAD;

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]);
        /* Reserve three fds for the libevent base, and two for the pipe */
        stats.reserved_fds += 5;
    }

/*
 * Set up a thread's information.
 */
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);
    }
    ...
}

/*
 * 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) {
    ...

    /* 从管道读取消息 */
    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); //读取连接

    ...
}    

/* Create threads after we've done all the libevent setup. */
for (i = 0; i < nthreads; i++) {
     create_worker(worker_libevent, &threads[i]);
}

/*
 * Creates a worker thread.
 */
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\n",
                strerror(ret));
        exit(1);
    }
}

/*
 * Worker thread: main event loop
 */
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...
     */
    me->item_lock_type = ITEM_LOCK_GRANULAR;
    pthread_setspecific(item_lock_type_key, &me->item_lock_type);

    register_thread_initialized();

    event_base_loop(me->base, 0);
    return NULL;
}

static struct event_base* mian_base;

/* initialize main thread libevent instance */
main_base = event_init();

if (settings.port && server_sockets(settings.port, tcp_transport,  portnumber_file)) {
            vperror("failed to listzhefen on TCP port %d", settings.port);
            exit(EX_OSERR);
}

static int server_sockets(int port, enum network_transport transport,
                          FILE *portnumber_file) {
    if (settings.inter == NULL) {
        return server_socket(settings.inter, port, transport, portnumber_file);
    }
    ...
}

/* 设置线程事件 */
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) {
    perror("event_add");
    return NULL;
}

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];
    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 ;
    }

    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); //连接加入懂啊队列

    memcachedD_CONN_DISPATCH(sfd, thread->thread_id);
    buf[0] = 'c';
    if (write(thread->notify_send_fd, buf, 1) != 1) {//向管道写入消息
        perror("Writing to thread notify pipe");
    }
}