Memcached源码阅读九 连接队列

Memcached中Master线程和Worker线程之间通信连接信息时，是通过连接队列来通信的，即Master线程投递一个消息到Worker线程的连接队列中，Worker线程从连接队列中读取链接信息来执行连接操作，下面我们简单分析下Memcached的连接队列结构。

typedef struct conn_queue_item CQ_ITEM;//每个连接信息的封装
struct conn_queue_item {
    int               sfd;//accept之后的描述符
    enum conn_states  init_state;//连接的初始状态
    int               event_flags;//libevent标志
    int               read_buffer_size;//读取数据缓冲区大小
    enum network_transport     transport;//内部通信所用的协议
    CQ_ITEM          *next;//用于实现链表的指针
};

typedef struct conn_queue CQ;//连接队列的封装
struct conn_queue {
    CQ_ITEM *head;//头指针，注意这里是单链表，不是双向链表
    CQ_ITEM *tail;//尾部指针，
    pthread_mutex_t lock;//锁
    pthread_cond_t  cond;//条件变量
};

//连接队列初始化
static void cq_init(CQ *cq) {
    pthread_mutex_init(&cq->lock, NULL);//初始化锁
    pthread_cond_init(&cq->cond, NULL);//初始化条件变量
    cq->head = NULL;
    cq->tail = NULL;
}

//获取一个连接
static CQ_ITEM *cq_pop(CQ *cq) {
    CQ_ITEM *item;

    pthread_mutex_lock(&cq->lock);//执行加锁操作
    item = cq->head;
    //获得头部指针指向的数据
    if (NULL != item) {
        //更新头指针信息
        cq->head = item->next;
        //这里为空的话，则尾指针也为空，链表此时为空
        if (NULL == cq->head)
            cq->tail = NULL;
    }
    //释放锁操作
    pthread_mutex_unlock(&cq->lock);

    return item;
}

//添加一个连接信息
static void cq_push(CQ *cq, CQ_ITEM *item) {
    item->next = NULL;

    pthread_mutex_lock(&cq->lock);//执行加锁操作
    //如果链表目前是空的
    if (NULL == cq->tail)
        //则头指针指向该结点
        cq->head = item;
    else
        cq->tail->next = item;//添加到尾部

    cq->tail = item;
    //尾部指针后移
    pthread_cond_signal(&cq->cond);
    //唤醒条件变量，如果有阻塞在该条件变量的线程，则会唤醒该线程
    pthread_mutex_unlock(&cq->lock);
}

//创建连接队列
static CQ_ITEM *cqi_new(void) {
    CQ_ITEM *item = NULL;
    pthread_mutex_lock(&cqi_freelist_lock);
    //加锁，保持数据同步
    if (cqi_freelist) {
        //更新空闲链表信息
        item = cqi_freelist;
        cqi_freelist = item->next;
    }
    pthread_mutex_unlock(&cqi_freelist_lock);

    //如果空闲链表没有多余的链接
    if (NULL == item) {
        int i;

        //初始化64个空闲连接信息
        item = malloc(sizeof(CQ_ITEM) * ITEMS_PER_ALLOC);
        if (NULL == item)
            return NULL;

        //将空闲的连接信息进行链接
        for (i = 2; i < ITEMS_PER_ALLOC; i++)
            item[i - 1].next = &item[i];

        pthread_mutex_lock(&cqi_freelist_lock);
        item[ITEMS_PER_ALLOC - 1].next = cqi_freelist;//加入到空闲链表中
        cqi_freelist = &item[1];
        pthread_mutex_unlock(&cqi_freelist_lock);
    }

    return item;
}

//释放item，也就是将item添加到空闲链表中
static void cqi_free(CQ_ITEM *item) {
    pthread_mutex_lock(&cqi_freelist_lock);
    item->next = cqi_freelist;
    cqi_freelist = item;
    pthread_mutex_unlock(&cqi_freelist_lock);
}

空闲链表类似于一种连接池的实现，服务器开发中经常需要各种池操作，大家在实现类似池时，可以做参考。