memcached客户端连接建立过程笔记

/* 设置conn的初始状态为conn_listening，表示这是一个监听socket，在之后的事件状态机中专门用于接收用户连接 */
if (!(listen_conn_add = conn_new(sfd, conn_listening,	
EV_READ | EV_PERSIST, 1,
transport, main_base))) {
fprintf(stderr, "failed to create listening connection\n");
exit(EXIT_FAILURE);
}
listen_conn_add->next = listen_conn;
listen_conn = listen_conn_add; /* listen_conn是一个全局变量，用来保存所有的监听连接 */

struct conn {
int sfd; /** 连接对应的fd，即监听fd或者用户连接fd*/

char *rbuf; /** 存储读取到的命令 */
char *rcurr; /** 已经解析到的rbuf的位置 */
int rsize; /** rbuf的全部长度 */
int rbytes; /** 从rcurr开始还有多少未解析的数据 */

char *wbuf;
char *wcurr;
int wsize;
int wbytes;
/** 标识当前状态结束之后的下一个状态，在状态机中使用 */
enum conn_states write_and_go;
void *write_and_free; /** free this memory after finishing writing */

char *ritem; /** 用来存储key-value中的value,在状态机中做了接收到的数据复制到存储位置的操作 */
int rlbytes;

/* 用于读取value的数据结构*/
/* data for the nread state */

/**
* item is used to hold an item structure created after reading the command
* line of set/add/replace commands, but before we finished reading the actual
* . The data is read into ITEM_data(item) to avoid extra copying.
*/
/**
* 在未读取到key-value中的value数据时，item用来存储处理set/add/replace这三个命令锁生成的item1数据结构，
* 之后读取到value会被直接读进item1的data区域，防止多一次的数据复制，这里需要多体会一下
*/
void *item; /* 在这几个命令set/add/replace会用到 */

/* data for the swallow state */
int sbytes; /* how many bytes to swallow */

enum protocol protocol; /* 标识传送数据类型 char型或者是二进制 */
enum network_transport transport; /* 标识数据传输的方式tcp或者udp或unix socket */

int hdrsize; /* number of headers‘ worth of space is allocated */

bool noreply; /* 标识该命令是否需要回复 */

conn *next; /* 指向下一个连接形成单链表结构 */
LIBEVENT_THREAD *thread; /* 指向所属的线程，每一个用户连接都会被固定分配给一个worker线程 */
};　

c->state = init_state; /** 设置连接的初始状态 */
...
event_set(&c->event, sfd, event_flags, event_handler, (void *)c); /** 设置事件监听以及event_handler回调函数*/
event_base_set(base, &c->event); /**注册到libevent*/

static void drive_machine(conn *c) {
bool stop = false;
int sfd;
socklen_t addrlen;
struct sockaddr_storage addr;
int nreqs = settings.reqs_per_event;
int res;
const char *str;
assert(c != NULL);
while (!stop) {
switch(c->state) {
case conn_listening: /** 状态为conn_listening的监听连接负责接收客户端连接*/
/** 关键部分，接收连接并设置为非阻塞*/
accept(c->sfd, (struct sockaddr *)&addr, &addrlen);
fcntl(sfd, F_SETFL, fcntl(sfd, F_GETFL) | O_NONBLOCK) < 0)
/** 先判断是否达到设置的最大连接数量*/
if (settings.maxconns_fast &&
stats.curr_conns + stats.reserved_fds >= settings.maxconns - 1) {
...
} else { /** 未达到最大连接数则分配该连接并创建conn表示客户单连接上下文*/
/** 新conn的初始状态为conn_new_cmd,监听读事件，传输协议为tcp*/
/** -------------------跳到下面dispach_conn_new的大致逻辑----------*/
dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST,
DATA_BUFFER_SIZE, tcp_transport);
}
stop = true; /** 跳出while循环*/
break;

/** 下面的状态基本由客户端建立连接之后的事件驱动*/
case ...:
...
return;
}

/**
* 这个函数只能由主线程来调用，用于将新接收的连接分配给worker线程
*/
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是主线程与worker线程数据交互的数据结构，对一个连接的包装*/
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; /** robin-round循环获取一个目标worker线程*/

LIBEVENT_THREAD *thread = threads + tid;
...
cq_push(thread->new_conn_queue, item); /** 将新的连接包装实体压入到woker线程待处理连接队列*/

MEMCACHED_CONN_DISPATCH(sfd, thread->thread_id);
buf[0] = ‘c‘;
/** 主线程通过初试化线程时创建的pipe通道给worker发送消息‘c‘表示分配给一个新的连接*/
if (write(thread->notify_send_fd, buf, 1) != 1) {
...
}
}
/**
*worker线程收到消息后会回调线程启动时pipe关注的事件注册的回调函数
*thread_libevent_process,下面继续看看该函数逻辑
*/
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");
/** 判断消息类型*/
switch (buf[0]) {
case ‘c‘: /** 新连接*/
item = cq_pop(me->new_conn_queue); /** pop出主线程压入的新连接结构*/

if (NULL != item) {
/** 创建新连接 me->base 表示将该连接关注的事件注册到的libevent 事件监听结构*/
conn *c = conn_new(item->sfd, item->init_state, item->event_flags,
item->read_buffer_size, item->transport, me->base);
...
cqi_free(item);
}
break;
/** 下面两个消息标识主要跟线程启动注册有关*/
/* we were told to flip the lock type and report in */
case ‘l‘:
me->item_lock_type = ITEM_LOCK_GRANULAR;
register_thread_initialized();
break;
case ‘g‘:
me->item_lock_type = ITEM_LOCK_GLOBAL;
register_thread_initialized();
break;
}
}