Memcached源码分析三 网络连接建立 接着上一篇继续分析，上一篇请参考 《Memcached源码阅读之网络监听的建立》，这篇主要分析TCP的连接建立（从前面的代码分析可以看出，这个过程是由主线程驱动的），UDP没有连接建立的过程，所以之间进行连接分发，我们后续分析，现在直接上代码进行讲解。 conn *conn_new(const int sfd, enum conn_states init_state, const int event_flags, const int read_buffer_size, enum network_transport transport, struct event_base *base) { conn *c = conn_from_freelist(); //获取一个空闲连接，conn是Memcached内部对网络连接的一个封装 //如果没有空闲的连接 if (NULL == c) { if (!(c = (conn *)calloc(1, sizeof(conn))))//申请空间 { fprintf(stderr, "calloc()\n"); return NULL; }MEMCACHED_CONN_CREATE(c); //进行一些初始化 c->rbuf = c->wbuf = 0; c->ilist = 0; c->suffixlist = 0; c->iov = 0; c->msglist = 0; c->hdrbuf = 0; c->rsize = read_buffer_size; c->wsize = DATA_BUFFER_SIZE; c->isize = ITEM_LIST_INITIAL; c->suffixsize = SUFFIX_LIST_INITIAL; c->iovsize = IOV_LIST_INITIAL; c->msgsize = MSG_LIST_INITIAL; c->hdrsize = 0; //每个conn都自带读入和输出缓冲区，在进行网络收发数据时，特别方便 c->rbuf = (char *)malloc((size_t)c->rsize); c->wbuf = (char *)malloc((size_t)c->wsize); c->ilist = (item **)malloc(sizeof(item *) * c->isize); c->suffixlist = (char **)malloc(sizeof(char *) * c->suffixsize); c->iov = (struct iovec *) malloc(sizeof(struct iovec) * c->iovsize); c->msglist = (struct msghdr *) malloc( sizeof(struct msghdr) * c->msgsize); if (c->rbuf == 0 || c->wbuf == 0 || c->ilist == 0 || c->iov == 0 || c->msglist == 0 || c->suffixlist == 0) { conn_free(c); fprintf(stderr, "malloc()\n"); return NULL; } STATS_LOCK(); //统计变量更新 stats.conn_structs++; STATS_UNLOCK(); } c->transport = transport; c->protocol = settings.binding_protocol; if (!settings.socketpath) { c->request_addr_size = sizeof(c->request_addr); } else { c->request_addr_size = 0; } //输出一些日志信息 if (settings.verbose > 1) { if (init_state == conn_listening) { fprintf(stderr, "<%d server listening (%s)\n", sfd, prot_text(c->protocol)); } else if (IS_UDP(transport)) { fprintf(stderr, "<%d server listening (udp)\n", sfd); } else if (c->protocol == negotiating_prot) { fprintf(stderr, "<%d new auto-negotiating client connection\n", sfd); } else if (c->protocol == ascii_prot) { fprintf(stderr, "<%d new ascii client connection.\n", sfd); } else if (c->protocol == binary_prot) { fprintf(stderr, "<%d new binary client connection.\n", sfd); } else { fprintf(stderr, "<%d new unknown (%d) client connection\n", sfd, c->protocol); assert(false); } } c->sfd = sfd; c->state = init_state; c->rlbytes = 0; c->cmd = -1; c->rbytes = c->wbytes = 0; c->wcurr = c->wbuf; c->rcurr = c->rbuf; c->ritem = 0; c->icurr = c->ilist; c->suffixcurr = c->suffixlist; c->ileft = 0; c->suffixleft = 0; c->iovused = 0; c->msgcurr = 0; c->msgused = 0; c->write_and_go = init_state; c->write_and_free = 0; c->item = 0; c->noreply = false; //建立sfd描述符上面的event事件，事件回调函数为event_handler 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) { //如果建立libevent事件失败，将创建的conn添加到空闲列表中 if (conn_add_to_freelist(c)) { conn_free(c); } perror("event_add"); return NULL; } STATS_LOCK(); //统计信息更新 stats.curr_conns++; stats.total_conns++; STATS_UNLOCK(); MEMCACHED_CONN_ALLOCATE(c->sfd); return c; } //获得conn conn *conn_from_freelist() { conn *c; pthread_mutex_lock(&conn_lock);//操作链表，加锁，保持同步 //freecurr为静态全局变量 if (freecurr > 0) { //freeconns是在Memcached启动时初始化的 c = freeconns[--freecurr]; } else//没有conn { c = NULL; } pthread_mutex_unlock(&conn_lock); return c; } //添加conn到空闲链表中 bool conn_add_to_freelist(conn *c) { bool ret = true; pthread_mutex_lock(&conn_lock); //freeconns还有空间 if (freecurr < freetotal) { freeconns[freecurr++] = c;//直接添加 ret = false; } else { //没有多余空间，进行扩容，按目前容量的2倍进行扩容 size_t newsize = freetotal * 2; conn **new_freeconns = realloc(freeconns, sizeof(conn *) * newsize); if (new_freeconns) { freetotal = newsize; freeconns = new_freeconns; freeconns[freecurr++] = c; ret = false; } } pthread_mutex_unlock(&conn_lock); return ret; } //libevent事件回调函数的处理，回调函数被调用时，表明Memcached监听的端口号有网络事件到了 void event_handler(const int fd, const short which, void *arg) { conn *c; c = (conn *)arg; assert(c != NULL); c->which = which; //这种情况应该很少出现 if (fd != c->sfd) { if (settings.verbose > 0) fprintf(stderr, "Catastrophic: event fd doesn't match conn fd!\n"); conn_close(c); return; } //进入业务处理状态机 drive_machine(c); return; }

Memcached源码阅读八 内存池 Memcached内部维护了一个内存池来减少频繁的malloc和free，在该内存池的基础上面实现了slab内存管理，下面简单介绍下内存池的实现，大家在实现类似结构时，可以做个参考。 static void *mem_base = NULL;//mem_base指向新申请的内存空间，指向整个内存空间的头部 static void *mem_current = NULL;//指向已经分配过的空间，且指向已经分配了空间的尾部 static size_t mem_avail = 0;//剩余空间大小 //部分初始化操作 mem_limit = limit;//初始容量 mem_base = malloc(mem_limit);//申请内存空间 if (mem_base != NULL) //如果不为空 { mem_current = mem_base; //当前还没分配，所以其指向为整个空间 mem_avail = mem_limit; //可用空间为满 } else { fprintf(stderr, "Warning: Failed to allocate requested memory in" " one large chunk.\nWill allocate in smaller chunks\n"); } //分配空间的过程，分配size大小的空间 static void *memory_allocate(size_t size) { void *ret; //如果未初始化 if (mem_base == NULL) { ret = malloc(size); } else { ret = mem_current; if (size > mem_avail) { return NULL; } //执行对齐操作 if (size % CHUNK_ALIGN_BYTES) { size += CHUNK_ALIGN_BYTES - (size % CHUNK_ALIGN_BYTES); } mem_current = ((char*)mem_current) + size; if (size < mem_avail) { mem_avail -= size; } else { mem_avail = 0; } } return ret; }

Memcached源码阅读二 网络监听的建立 Memcahced是一个服务器程序，所以需要建立网络监听来接受其他客户端机器的连接，下面分析下其过程，这次分析是基于Memcached 1.4.15版本分析的。 // 如果socketpath为空，则表示使用的TCP / UDP, 不是使用unix socket //如果socketpath为空，则表示使用的TCP/UDP,不是使用unix socket if (settings.socketpath == NULL) { //可以从环境变量读取端口文件所在的文件路径 const char *portnumber_filename = getenv("MEMCACHED_PORT_FILENAME"); char temp_portnumber_filename[PATH_MAX]; FILE *portnumber_file = NULL; //如果端口文件不为空，则打开 if (portnumber_filename != NULL) { snprintf(temp_portnumber_filename, sizeof(temp_portnumber_filename), "%s.lck", portnumber_filename); portnumber_file = fopen(temp_portnumber_filename, "a"); if (portnumber_file == NULL) { fprintf(stderr, "Failed to open \"%s\": %s\n", temp_portnumber_filename, strerror(errno)); } } //settings.port表示Memcached采用的是TCP协议，创建TCP Socket，监听并且绑定 errno = 0; if (settings.port && server_sockets(settings.port, tcp_transport, portnumber_file)) { vperror("failed to listen on TCP port %d", settings.port); exit(EX_OSERR); } //settings.udpport表示Memcached采用的是UDP协议，创建UDP Socket，监听并且绑定 errno = 0; if (settings.udpport && server_sockets(settings.udpport, udp_transport, portnumber_file)) { vperror("failed to listen on UDP port %d", settings.udpport); exit(EX_OSERR); } //端口文件不为空 if (portnumber_file) { fclose(portnumber_file);//关闭文件 rename(temp_portnumber_filename, portnumber_filename);//重命名端口文件 } } //TCP和UDP使用的是同一个接口来创建监听和绑定 static int server_sockets(int port, enum network_transport transport, FILE *portnumber_file) { //settings.inter指定的是要绑定的ip地址信息，如果为空，则表示是绑定本机一个ip if (settings.inter == NULL) { //执行监听和绑定操作 return server_socket(settings.inter, port, transport, portnumber_file); } //如果服务器有多个ip信息，可以在每个(ip,port)上面绑定一个Memcached实例，下面是一些输入参数的解析，解析完毕之后，执行绑定 else { // tokenize them and bind to each one of them.. char *b; int ret = 0; char *list = strdup(settings.inter); if (list == NULL) { fprintf(stderr, "Failed to allocate memory for parsing server interface string\n"); return 1; } for (char *p = strtok_r(list, ";,", &b); p != NULL; p = strtok_r(NULL, ";,", &b)) { int the_port = port; char *s = strchr(p, ':'); if (s != NULL) { *s = '\0'; ++s; if (!safe_strtol(s, &the_port)) { fprintf(stderr, "Invalid port number: \"%s\"", s); return 1; } } if (strcmp(p, "*") == 0) { p = NULL; } //绑定多次,循环调用单个的绑定函数 ret |= server_socket(p, the_port, transport, portnumber_file); } free(list); return ret; } } //执行真正的绑定 static int server_socket(const char *interface, int port, enum network_transport transport, FILE *portnumber_file) { int sfd; struct linger ling = { 0, 0 }; struct addrinfo *ai; struct addrinfo *next; //设定协议无关，用于监听的标志位 struct addrinfo hints = { .ai_flags = AI_PASSIVE,.ai_family = AF_UNSPEC }; char port_buf[NI_MAXSERV]; int error; int success = 0; int flags = 1; //指定socket的类型，如果是udp，则用数据报协议，如果是tcp,则用数据流协议 hints.ai_socktype = IS_UDP(transport) ? SOCK_DGRAM : SOCK_STREAM; if (port == -1) { port = 0; } snprintf(port_buf, sizeof(port_buf), "%d", port); //调用getaddrinfo,将主机地址和端口号映射成为socket地址信息，地址信息由ai带回 error = getaddrinfo(interface, port_buf, &hints, &ai); if (error != 0) { if (error != EAI_SYSTEM) fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error)); else perror("getaddrinfo()"); return 1; } /*getaddrinfo返回多个addrinfo的情形有如下两种： 1.如果与interface参数关联的地址有多个，那么适用于所请求地址簇的每个地址都返回一个对应的结构。 2.如果port_buf参数指定的服务支持多个套接口类型，那么每个套接口类型都可能返回一个对应的结构。 */ for (next = ai; next; next = next->ai_next) { conn *listen_conn_add; //为每个地址信息建立socket if ((sfd = new_socket(next)) == -1) { //建立socket过程中可能发生的，比如打开文件描述符过多等 if (errno == EMFILE) { perror("server_socket"); exit(EX_OSERR); } continue; } #ifdef IPV6_V6ONLY if (next->ai_family == AF_INET6) { //设定IPV6的选项值，设置了IPV6_V6ONLY，表示只收发IPV6的数据包，此时IPV4和IPV6可以绑定到同一个端口而不影响数据的收发 error = setsockopt(sfd, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&flags, sizeof(flags)); if (error != 0) { perror("setsockopt"); close(sfd); continue; } } #endif //设定socket选项，SO_REUSEADDR表示重用地址信息，具体重用哪些东西自行学习，必须在bind操作之前设置 setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags)); if (IS_UDP(transport)){ maximize_sndbuf(sfd);//扩大发送缓冲区 } else { //设定socket选项，SO_KEEPALIVE表示保活 error = setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags)); if (error != 0) perror("setsockopt"); //设定socket选项，SO_LINGER表示执行close操作时，如果缓冲区还有数据，可以继续发送 error = setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling)); if (error != 0) perror("setsockopt"); //设定IP选项，TCP_NODELAY表示禁用Nagle算法 error = setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags)); if (error != 0) perror("setsockopt"); } if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) { if (errno != EADDRINUSE) { perror("bind()"); close(sfd); freeaddrinfo(ai); return 1; } close(sfd); continue; } else { success++; //如果不是UDP协议，则执行监听操作，监听队列为初始启动的值 if (!IS_UDP(transport) && listen(sfd, settings.backlog) == -1) { perror("listen()"); close(sfd); freeaddrinfo(ai); return 1; } if (portnumber_file != NULL && (next->ai_addr->sa_family == AF_INET || next->ai_addr->sa_family == AF_INET6)) { union { struct sockaddr_in in; struct sockaddr_in6 in6; } my_sockaddr; socklen_t len = sizeof(my_sockaddr); //这时还没连接建立，调用getsockname不知道有什么用？ if (getsockname(sfd, (struct sockaddr*) &my_sockaddr, &len) == 0) { if (next->ai_addr->sa_family == AF_INET) { fprintf(portnumber_file, "%s INET: %u\n", IS_UDP(transport) ? "UDP" : "TCP", ntohs(my_sockaddr.in.sin_port)); } else { fprintf(portnumber_file, "%s INET6: %u\n", IS_UDP(transport) ? "UDP" : "TCP", ntohs(my_sockaddr.in6.sin6_port)); } } } } if (IS_UDP(transport)) { int c; for (c = 0; c < settings.num_threads_per_udp; c++) { //分发连接，因为UDP没有连接建立的过程，直接进行连接的分发 dispatch_conn_new(sfd, conn_read, EV_READ | EV_PERSIST, UDP_READ_BUFFER_SIZE, transport); } } else { //TCP,建立连接 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; //释放资源 freeaddrinfo(ai); return success == 0; } //建立socket static int new_socket(struct addrinfo *ai) { int sfd; int flags; //调用系统函数建立socket if ((sfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) == -1) { return -1; } //设定socket为非阻塞的 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { perror("setting O_NONBLOCK"); close(sfd); return -1; } return sfd; } //如果是UDP协议，调整发送缓存到最大值 static void maximize_sndbuf(const int sfd) { socklen_t intsize = sizeof(int); int last_good = 0; int min, max, avg; int old_size; //读取socket的选项，SO_SNDBF表示发送缓存 if (getsockopt(sfd, SOL_SOCKET, SO_SNDBUF, &old_size, &intsize) != 0) { if (settings.verbose > 0) perror("getsockopt(SO_SNDBUF)"); return; } //二分搜索来设定，很巧的设计 min = old_size; max = MAX_SENDBUF_SIZE; while (min <= max) { avg = ((unsigned int)(min + max)) / 2; if (setsockopt(sfd, SOL_SOCKET, SO_SNDBUF, (void *)&avg, intsize) == 0) { last_good = avg; min = avg + 1; } else { max = avg - 1; } } if (settings.verbose > 1) fprintf(stderr, "<%d send buffer was %d, now %d\n", sfd, old_size, last_good); } 至此，网络相关的部分已经完成，后向连接的建立（conn_new）和连接分发（dispatch_conn_new），我们放到其他文章中进行分析。

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); } 空闲链表类似于一种连接池的实现，服务器开发中经常需要各种池操作，大家在实现类似池时，可以做参考。

Memcached源码阅读六 get过程 我们在前面分析过，Memcached从网络读取完数据，解析数据，如果是get操作，则执行get操作，下面我们分析下get操作的流程。 //根据key信息和key的长度信息读取数据 item *item_get(const char *key, const size_t nkey) { item *it; uint32_t hv; hv = hash(key, nkey, 0);//获得分段锁信息，如果未进行扩容，则item的hash表是多个hash桶共用同一个锁，即是分段的锁 item_lock(hv);//执行分段加锁 it = do_item_get(key, nkey, hv);//执行get操作 item_unlock(hv);//释放锁 return it; } //执行分段加锁 void item_lock(uint32_t hv) { uint8_t *lock_type = pthread_getspecific(item_lock_type_key); if (likely(*lock_type == ITEM_LOCK_GRANULAR)) { mutex_lock(&item_locks[(hv & hashmask(hashpower)) % item_lock_count]);//执行分段加锁 } else {//如果在扩容过程中 mutex_lock(&item_global_lock); } } //执行分段解锁 void item_unlock(uint32_t hv) { uint8_t *lock_type = pthread_getspecific(item_lock_type_key); if (likely(*lock_type == ITEM_LOCK_GRANULAR)) { mutex_unlock(&item_locks[(hv & hashmask(hashpower)) % item_lock_count]);//释放分段锁 } else {//如果在扩容过程中 mutex_unlock(&item_global_lock); } } //执行读取操作 item *do_item_get(const char *key, const size_t nkey, const uint32_t hv) { item *it = assoc_find(key, nkey, hv);//从Hash表中获取相应的结构 if (it != NULL) { refcount_incr(&it->refcount);//item的引用次数+1 if (slab_rebalance_signal && //如果正在进行slab调整，且该item是调整的对象 ((void *)it >= slab_rebal.slab_start && (void *)it < slab_rebal.slab_end)) { do_item_unlink_nolock(it, hv);//将item从hashtable和LRU链中移除 do_item_remove(it);//删除item it = NULL;//置为空 } } int was_found = 0; //打印调试信息 if (settings.verbose > 2) { if (it == NULL) { fprintf(stderr, "> NOT FOUND %s", key); } else { fprintf(stderr, "> FOUND KEY %s", ITEM_key(it)); was_found++; } } if (it != NULL) { //判断Memcached初始化是否开启过期删除机制，如果开启，则执行删除相关操作 if (settings.oldest_live != 0 && settings.oldest_live <= current_time && it->time <= settings.oldest_live) { do_item_unlink(it, hv);//将item从hashtable和LRU链中移除 do_item_remove(it);//删除item it = NULL; if (was_found) { fprintf(stderr, " -nuked by flush"); } //判断item是否过期 } else if (it->exptime != 0 && it->exptime <= current_time) { do_item_unlink(it, hv);//将item从hashtable和LRU链中移除 do_item_remove(it);//删除item it = NULL; if (was_found) { fprintf(stderr, " -nuked by expire"); } } else { it->it_flags |= ITEM_FETCHED;//item的标识修改为已经读取 DEBUG_REFCNT(it, '+'); } } if (settings.verbose > 2) fprintf(stderr, "\n"); return it; } //移除item void do_item_remove(item *it) { MEMCACHED_ITEM_REMOVE(ITEM_key(it), it->nkey, it->nbytes); assert((it->it_flags & ITEM_SLABBED) == 0);//判断item的状态是否正确 if (refcount_decr(&it->refcount) == 0) {//修改item的引用次数 item_free(it);//释放item } } //释放item void item_free(item *it) { size_t ntotal = ITEM_ntotal(it);//获得item的大小 unsigned int clsid; assert((it->it_flags & ITEM_LINKED) == 0);//判断item的状态是否正确 assert(it != heads[it->slabs_clsid]);//item不能为LRU的头指针 assert(it != tails[it->slabs_clsid]);//item不能为LRU的尾指针 assert(it->refcount == 0);//释放时，需保证引用次数为0 /* so slab size changer can tell later if item is already free or not */ clsid = it->slabs_clsid; it->slabs_clsid = 0;//断开slabclass的链接 DEBUG_REFCNT(it, 'F'); slabs_free(it, ntotal, clsid);//slabclass结构执行释放 } //slabclass结构释放 void slabs_free(void *ptr, size_t size, unsigned int id) { pthread_mutex_lock(&slabs_lock);//保持同步 do_slabs_free(ptr, size, id);//执行释放 pthread_mutex_unlock(&slabs_lock); } //slabclass结构释放 static void do_slabs_free(void *ptr, const size_t size, unsigned int id) { slabclass_t *p; item *it; assert(((item *)ptr)->slabs_clsid == 0);//判断数据是否正确 assert(id >= POWER_SMALLEST && id <= power_largest);//判断id合法性 if (id < POWER_SMALLEST || id > power_largest)//判断id合法性 return; MEMCACHED_SLABS_FREE(size, id, ptr); p = &slabclass[id]; it = (item *)ptr; it->it_flags |= ITEM_SLABBED;//修改item的状态标识，修改为空闲 it->prev = 0;//断开数据链表 it->next = p->slots; if (it->next) it->next->prev = it; p->slots = it; p->sl_curr++;//空闲item个数+1 p->requested -= size;//空间增加size return; } //将item从hashtable和LRU链中移除。是do_item_link的逆操作 void do_item_unlink(item *it, const uint32_t hv) { MEMCACHED_ITEM_UNLINK(ITEM_key(it), it->nkey, it->nbytes); mutex_lock(&cache_lock);//执行同步 if ((it->it_flags & ITEM_LINKED) != 0) {//判断状态值，保证item还在LRU队列中 it->it_flags &= ~ITEM_LINKED;//修改状态值 STATS_LOCK();//更新统计信息 stats.curr_bytes -= ITEM_ntotal(it); stats.curr_items -= 1; STATS_UNLOCK(); assoc_delete(ITEM_key(it), it->nkey, hv);//从Hash表中删除 item_unlink_q(it);//将item从slabclass对应的LRU队列摘除 do_item_remove(it);//移除item } mutex_unlock(&cache_lock); } Memcached的get操作在读取数据时，会判断数据的有效性，使得不用额外去处理过期数据，get操作牵涉到Slab结构，Hash表，LRU队列的更新，我们后面专门分析这些的变更，这里暂不分析。

Memcached源码阅读七 cas属性 cas即compare and set或者compare and swap,是实现乐观锁的一种技术，乐观锁是相对悲观锁来说的，所谓悲观锁是在数据处理过程中，完全锁定，这种能完全保证数据的一致性，但在多线程情况下，并发性能差，通常是使用各种锁技术实现；而乐观锁是通过版本号机制来实现数据一致性，过程中会使用CPU提供的原子操作指令，乐观锁能提高系统的并发性能，Memcached使用cas是保证数据的一致性，不是严格为了实现锁。 Memcached是多客户端应用，在多个客户端修改同一个数据时，会出现相互覆盖的情况，在这种情况下，使用cas版本号验证，可以有效的保证数据的一致性，Memcached默认是打开cas属性的，每次存储数据时，都会生成其cas值并和item一起存储，后续的get操作会返回系统生成的cas值，在执行set等操作时，需要将cas值传入，下面我们看看Memcached内部是如何实现cas的，关于如何使用Mecached的CAS协议，请参考文章：Memcached的CAS协议（链接：http://langyu.iteye.com/blog/680052）。 //为新的item生成cas值 uint64_t get_cas_id(void) { static uint64_t cas_id = 0; return ++cas_id; } //这段代码是store_item的代码片段，这里是执行cas存储时执行的判断逻辑， else if (ITEM_get_cas(it) == ITEM_get_cas(old_it))//cas值一致 { pthread_mutex_lock(&c->thread->stats.mutex); c->thread->stats.slab_stats[old_it->slabs_clsid].cas_hits++; pthread_mutex_unlock(&c->thread->stats.mutex); item_replace(old_it, it, hv);//执行存储逻辑 stored = STORED; } //cas值不一致，不进行实际的存储 else { pthread_mutex_lock(&c->thread->stats.mutex); c->thread->stats.slab_stats[old_it->slabs_clsid].cas_badval++; //更新统计信息 pthread_mutex_unlock(&c->thread->stats.mutex); if (settings.verbose > 1) { //打印错误日志 fprintf(stderr, "CAS: failure: expected %llu, got %llu\n", (unsigned long long) ITEM_get_cas(old_it), (unsigned long long) ITEM_get_cas(it)); } stored = EXISTS; }

Memcached源码阅读十 Hash表操作 Memcached的Hash表用来提高数据访问性能，通过链接法来解决Hash冲突，当Hash表中数据多余Hash表容量的1.5倍时，Hash表就会扩容，Memcached的Hash表操作没什么特别的，我们这里简单介绍下Memcached里面的Hash表操作。 //hash表插入元素 int assoc_insert(item *it, const uint32_t hv) { unsigned int oldbucket; //如果已经开始扩容，且扩容的桶编号大于目前的item所在桶的编号 if (expanding && (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket) { //这里是类似单链表的，按单链表的操作进行插入 it->h_next = old_hashtable[oldbucket]; old_hashtable[oldbucket] = it; } else { //已经扩容，则按新的Hash规则进行路由 it->h_next = primary_hashtable[hv & hashmask(hashpower)]; //这里在新的Hash表中执行单链表插入 primary_hashtable[hv & hashmask(hashpower)] = it; } hash_items++;//元素个数+1 if (! expanding && hash_items > (hashsize(hashpower) * 3) / 2) { //开始扩容 assoc_start_expand();//唤醒扩容条件变量 } MEMCACHED_ASSOC_INSERT(ITEM_key(it), it->nkey, hash_items); return 1; } //hash表删除元素 void assoc_delete(const char *key, const size_t nkey, const uint32_t hv) { item **before = _hashitem_before(key, nkey, hv); //获得item对应的桶的前一个元素 if (*before) { item *nxt; hash_items--;//元素个数-1 MEMCACHED_ASSOC_DELETE(key, nkey, hash_items); nxt = (*before)->h_next;//执行单链表的删除操作 (*before)->h_next = 0; *before = nxt; return; } assert(*before != 0); } 像Hash表的扩容，初始化等已经在其他博客中介绍过了，这里就不在阐述。

Memcached源码阅读十二 set操作 之前分析了Memcached的get操作，下面分析set操作的流程。 //存储item enum store_item_type store_item(item *item, int comm, conn* c) { enum store_item_type ret; uint32_t hv; hv = hash(ITEM_key(item), item->nkey, 0);//获取Hash表的分段锁 item_lock(hv);//执行数据同步 ret = do_store_item(item, comm, c, hv);//存储item item_unlock(hv); return ret; } //存储item enum store_item_type do_store_item(item *it, int comm, conn *c,const uint32_t hv) { char *key = ITEM_key(it);//读取item对应的key item *old_it = do_item_get(key, it->nkey, hv); //读取相应的item,如果没有相关的数据,old_it为NULL enum store_item_type stored = NOT_STORED;//item状态标记 item *new_it = NULL; int flags; //如果old_it不为NULL,且操作为add操作 if (old_it != NULL && comm == NREAD_ADD) { do_item_update(old_it);//更新数据 } //old_it为空，且操作为REPLACE，则什么都不做 else if (!old_it && (comm == NREAD_REPLACE || comm == NREAD_APPEND || comm == NREAD_PREPEND)) { //memcached的Replace操作是替换已有的数据，如果没有相关数据，则不做任何操作 } //以cas方式读取 else if (comm == NREAD_CAS) { if (old_it == NULL) //为空 { // LRU expired stored = NOT_FOUND;//修改状态 pthread_mutex_lock(&c->thread->stats.mutex);//更新Worker线程统计数据 c->thread->stats.cas_misses++; pthread_mutex_unlock(&c->thread->stats.mutex); } //old_it不为NULL，且cas属性一致 else if (ITEM_get_cas(it) == ITEM_get_cas(old_it)) { pthread_mutex_lock(&c->thread->stats.mutex); c->thread->stats.slab_stats[old_it->slabs_clsid].cas_hits++; //更新Worker线程统计信息 pthread_mutex_unlock(&c->thread->stats.mutex); item_replace(old_it, it, hv); //执行item的替换操作，用新的item替换老的item stored = STORED;//修改状态值 } else //old_it不为NULL,且cas属性不一致 { pthread_mutex_lock(&c->thread->stats.mutex); c->thread->stats.slab_stats[old_it->slabs_clsid].cas_badval++; //更新Worker线程统计信息 pthread_mutex_unlock(&c->thread->stats.mutex); if (settings.verbose > 1) { fprintf(stderr, "CAS: failure: expected %llu, got %llu\n", (unsigned long long) ITEM_get_cas(old_it), (unsigned long long) ITEM_get_cas(it)); } stored = EXISTS; //修改状态值，修改状态值为已经存在，且不存储最新的数据 } } else //执行其他操作的写 { //以追加的方式执行写 if (comm == NREAD_APPEND || comm == NREAD_PREPEND) { //验证cas有效性 if (ITEM_get_cas(it) != 0) { //cas验证不通过 if (ITEM_get_cas(it) != ITEM_get_cas(old_it)) { stored = EXISTS;//修改状态值为已存在 } } //状态值为没有存储,也就是cas验证通过，则执行写操作 if (stored == NOT_STORED) { flags = (int) strtol(ITEM_suffix(old_it), (char **) NULL, 10); //申请新的空间 new_it = do_item_alloc(key, it->nkey, flags, old_it->exptime,it->nbytes + old_it->nbytes - 2 , hv); if (new_it == NULL) { //空间不足 if (old_it != NULL) do_item_remove(old_it);//删除老的item return NOT_STORED; } if (comm == NREAD_APPEND)//追加方式 { memcpy(ITEM_data(new_it), ITEM_data(old_it),old_it->nbytes);//老数据拷贝到新数据中 memcpy(ITEM_data(new_it) + old_it->nbytes - 2,ITEM_data(it), it->nbytes);//同时拷贝最近缓冲区已有的数据 } else { //这里和具体协议相关 memcpy(ITEM_data(new_it), ITEM_data(it), it->nbytes);//拷贝it的数据到new_it中 memcpy(ITEM_data(new_it) + it->nbytes - 2 ,ITEM_data(old_it), old_it->nbytes);//同时拷贝最近缓冲区已有的数据 } it = new_it; } } if (stored == NOT_STORED) { if (old_it != NULL)//如果old_it不为空 item_replace(old_it, it, hv);//替换老的值 else do_item_link(it, hv);//重新存储数据 c->cas = ITEM_get_cas(it);//获取cas值 stored = STORED; } } if (old_it != NULL) do_item_remove(old_it);//释放空间 if (new_it != NULL) do_item_remove(new_it);//释放空间 if (stored == STORED)//如果已经存储了 { c->cas = ITEM_get_cas(it);//获取cas属性 } return stored; } //更新item，这个只更新时间 void do_item_update(item *it) { MEMCACHED_ITEM_UPDATE(ITEM_key(it), it->nkey, it->nbytes); if (it->time < current_time - ITEM_UPDATE_INTERVAL) { //更新有时间限制 assert((it->it_flags & ITEM_SLABBED) == 0); mutex_lock(&cache_lock);//保持同步 //更新LRU队列的Item if ((it->it_flags & ITEM_LINKED) != 0) { item_unlink_q(it);//断开连接 it->time = current_time;//更新item的时间 item_link_q(it);//重新添加 } mutex_unlock(&cache_lock); } } //用新的item替换老的item int do_item_replace(item *it, item *new_it, const uint32_t hv) { MEMCACHED_ITEM_REPLACE(ITEM_key(it), it->nkey, it->nbytes, ITEM_key(new_it), new_it->nkey, new_it->nbytes); //判断it是已经分配过的，如果未分配，则断言失败 assert((it->it_flags & ITEM_SLABBED) == 0); do_item_unlink(it, hv);//断开连接 return do_item_link(new_it, hv);//重新添加 } 有些item的操作已经在get操作中有分析，我们此处不做分析，我们下一篇分析下Memcached内部如何选择合适的空间来存放item。

Memcached源码阅读十六 线程交互 Memcached按之前的分析可以知道，其是典型的Master-Worker线程模型，这种模型很典型，其工作模型是Master绑定端口，监听网络连接，接受网络连接之后，通过线程间通信来唤醒Worker线程，Worker线程已经连接的描述符执行读写操作，这种模型简化了整个通信模型，下面分析下这个过程。 case conn_listening: addrlen = sizeof(addr); //Master线程(main)进入状态机之后执行accept操作，这个操作也是非阻塞的。 if ((sfd = accept(c->sfd, (struct sockaddr *) &addr, &addrlen)) == -1) { //非阻塞模型，这个错误码继续等待 if (errno == EAGAIN || errno == EWOULDBLOCK) { stop = true; } //连接超载 else if (errno == EMFILE) { if (settings.verbose > 0) fprintf(stderr, "Too many open connections\n"); accept_new_conns(false); stop = true; } else { perror("accept()"); stop = true; } break; } //已经accept成功，将accept之后的描述符设置为非阻塞的 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { perror("setting O_NONBLOCK"); close(sfd); break; } //判断是否超过最大连接数 if (settings.maxconns_fast && stats.curr_conns + stats.reserved_fd >= settings.maxconns - 1) { str = "ERROR Too many open connections\r\n"; res = write(sfd, str, strlen(str)); close(sfd); STATS_LOCK(); stats.rejected_conns++; STATS_UNLOCK(); } else { //直线连接分发 dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST, DATA_BUFFER_SIZE, tcp_transport); } stop = true; break; 这个是TCP的连接建立过程，由于UDP不需要建立连接，所以直接分发给Worker线程，让Worker线程进行读写操作，而TCP在建立连接之后，也执行连接分发（和UDP的一样），下面看看dispatch_conn_new内部是如何进行链接分发的。 ...

Memcached源码阅读十七 状态机 按我们之前的描述，Master线程建立连接之后，分发给Worker线程，而Worker线程处理业务逻辑时，会进入状态机，状态机按不同的状态处理业务逻辑，我们在分析连接分发时，已经看到了Master线程进入状态机时在有新连接建立的时候，后续的状态都是业务逻辑的状态，其处理流程如下图所示： 共有10个状态（代码中的状态不止这些，有些没什么用，此处就没展现），状态listenning状态是Master建立连接的过程，我们已经分析过了，我们接下来分不同的文章分析其余的9中状态。 enum conn_states { conn_listening, //监听状态 conn_new_cmd, //为新连接做一些准备 conn_waiting, //等待读取一个数据包 conn_read, //读取网络数据 conn_parse_cmd, //解析缓冲区的数据 conn_write, //简单的回复数据 conn_nread, //读取固定数据的网络数据 conn_swallow, //处理不需要的写缓冲区的数据 conn_closing, //关闭连接 conn_mwrite, //顺序的写多个item数据 conn_max_state //最大状态，做断言使用 }; 这篇文件先分析conn_new_cmd和conn_wating状态，子线程最初进入的状态就是conn_new_cmd状态，这个状态主要是做一些清理。 case conn_new_cmd: //全局变量，记录每个libevent实例处理的事件，通过初始启动参数配置 --nreqs; //还可以处理请求 if (nreqs >= 0) { //整理缓冲区 reset_cmd_handler(c); } //拒绝请求 else { pthread_mutex_lock(&c->thread->stats.mutex); c->thread->stats.conn_yields++;//更新统计数据 pthread_mutex_unlock(&c->thread->stats.mutex); //如果缓冲区有数据，则需要处理 if (c->rbytes > 0) { //更新libevent状态 if (!update_event(c, EV_WRITE | EV_PERSIST)) { if (settings.verbose > 0) fprintf(stderr, "Couldn't update event\n"); conn_set_state(c, conn_closing);//关闭连接 } } stop = true; } break; //整理缓冲区 static void reset_cmd_handler(conn *c) { c->cmd = -1; c->substate = bin_no_state; //还有item if (c->item != NULL) { //删除item，本篇不分析其实现，后续分析 item_remove(c->item); c->item = NULL; } //整理缓冲区 conn_shrink(c); //缓冲区还有数据 if (c->rbytes > 0) { //更新状态 conn_set_state(c, conn_parse_cmd); } //如果没有数据 else { //进入等待状态，状态机没有数据要处理，就进入这个状态 conn_set_state(c, conn_waiting); } } //缩小缓冲区 static void conn_shrink(conn *c) { assert(c != NULL); //如果是UDP协议，不牵涉缓冲区管理 if (IS_UDP(c->transport)) return; //读缓冲区空间大小>READ_BUFFER_HIGHWAT && 已经读到的数据还没解析的数据小于 DATA_BUFFER_SIZE if (c->rsize > READ_BUFFER_HIGHWAT && c->rbytes < DATA_BUFFER_SIZE) { char *newbuf; if (c->rcurr != c->rbuf) //目前数据是从rcurr开始的，移动数据到rbuf中 memmove(c->rbuf, c->rcurr, (size_t) c->rbytes); //按DATA_BUFFER_SIZE扩大缓冲区 newbuf = (char *) realloc((void *) c->rbuf, DATA_BUFFER_SIZE); if (newbuf) { //更新读缓冲区 c->rbuf = newbuf; //更新读缓冲区大小 c->rsize = DATA_BUFFER_SIZE; } c->rcurr = c->rbuf; } //需要写出的item的个数，也就是要发送给客户端的item的个数 if (c->isize > ITEM_LIST_HIGHWAT) { //增大存放item的空间 item **newbuf = (item**) realloc((void *) c->ilist,ITEM_LIST_INITIAL * sizeof(c->ilist[0])); if (newbuf) { //更新信息 c->ilist = newbuf; //更新信息 c->isize = ITEM_LIST_INITIAL; } } //msghdr的个数，memcached发送消息是通过sendmsg批量发送的 if (c->msgsize > MSG_LIST_HIGHWAT) { struct msghdr *newbuf = (struct msghdr *) realloc((void *) c->msglist,MSG_LIST_INITIAL * sizeof(c->msglist[0]));//增大空间 if (newbuf) { //更新信息 c->msglist = newbuf; //更新信息 c->msgsize = MSG_LIST_INITIAL; } } //msghdr里面iov的数量 if (c->iovsize > IOV_LIST_HIGHWAT) { //增大空间 struct iovec *newbuf = (struct iovec *) realloc((void *) c->iov,IOV_LIST_INITIAL * sizeof(c->iov[0])); if (newbuf) { //更新信息 c->iov = newbuf; //更新信息 c->iovsize = IOV_LIST_INITIAL; } } } 从conn_new_cmd状态会进入conn_parse_cmd状态（如果有数据）或者conn_waiting（如果没有数据）状态，下面看看conn_waiting状态。 ...