转自
Memcached 结构分析
Memcached是一个分布式的内存缓存库，正好自己想写个cache的模块，那么就偷偷师吧。 功能库看的是实现原理和思路，性能库看的是实现细节，memcahed是属于一个看性能的库(实现cache功能的模块很多，但是性能就有高低了) 1、memcached的数据交互协议 memcached是分布式的内存缓存服务器，它是通过socket(tcp/udp/unixsock)与其他程序交换数据的，这样就需要一套协议来保证正常通信。 现在先看看memcached的通信协议，memcached的通信协议是以" "标志符来表示一个完整的解析单元的，这里说的解析单元是指服务器可以"理解"的最小数据分块，一个完整的命令是由一个或多个可解析单元组成的，现在列举一下通信命令：
对于所有的命令都有可能产生错误，这时服务器返回： "ERROR ":服务器收到一个无效的命令 "CLIENT_ERROR error_msg ":命令格式错误 "SERVER_ERROR error_msg ":服务器处理错误，这时连接会被关闭2、memcached的内存管理 内存管理都需要解决：分配、回收、碎片这几个一般性问题，memcached的处理方式是(通过宏USE_SYSTEM_MALLOC 可以控制memcached使用系统的malloc/free管理内存，这里不讨论)； 分配-->预分配 + 动态2倍分配 --> 减少realloc的调用 回收-->从来不释放内存 --> memcached的目的就是通过内存缓存数据，没有必要释放 碎片-->固定大小分配 + 通过额外的动态指针数组保存各个分块的地址 + (增加HEAD/TAIL指针数组)LRU算法回收 --> 加快获得空闲"内存块"的获得 memcached将内存划分为不同大小的集合(通过结构体slabclass_t来维护一个集合的信息)，现在看看slabclass_t结构。这里有几个概念： 内存单元：集合中维护的"逻辑内存块"的大小，它是以sizeof(void*)字节对齐的 内存页： slabclass_t分配内存的时候是以perslab个内存单元分配的，这perslab个连续的内存单元就是内存页
了解了memcached的内存管理的数据结构后，下面看看内存管理相关的几个主要函数： do_slabs_newslab(class_id) 这个函数分配一个新的内存页 |-->检查内存分配是否已经超过了限制 | 检查是否需要进行内存页指针的2倍扩展 | 从操作系统中申请内存malloc(len)-->对于需要在不同的slabclass间转移数据的应用len使用的固定大小1M | 而其他应用这里分配的是size*perslab,这里size是sizeof(void *)对齐的 | 确认malloc成功后初时化它的值为0(memset), （我想这里可以使用calloc代替） 将end_page_ptr指向新的内存页，并把它加入到内存页数组中，同时修改对应的计算变量 (这个函数是memcached中唯一分配"用户可用内存"的地方, "用户可用"是指set/update/replace指令可以控制的内存) do_slabs_alloc(need_size) 这个函数功能是根据需要的大小申请内存块 | 根据需要的大小查找对应的slabclass_t结构 | 检查是否超出了设置的内存限额 检查内存单元指针数组slots是否为空，如果非空-->返回一个空的内存单元 检查是否分配了新的内存页， 如果是-->返回一个"新的"内存单元(没有加入到slots中) 如果新的内存页为空，那么调用do_slabs_newslab从系统分配内存 （当然do_slabs_alloc还会修改对应的计数变量） do_slabs_free(pointer, mem_unit_size) | 根据mem_unit_size查找对应的slabclass_t结构 | 检查是否需要进行内存单元数组slots的2倍扩展，(2倍扩展都是使用realloc完成数据转移的) | 将释放的内存加入到内存单元数组中 (可以看到memcached是不真正释放内存的，而且它的分配与释放操作都是很简单的指针赋值操作， 这就是memcached内存的管理，也是它快的原因之一，另外一个原因在于它的hash算法)do_slabs_reassign(src_id, des_id) 这个函数将个集合中的某一个内存页的内容复制到另一个集合的一个内存页中 | 根据id获得源集合和目标集合的slabcalss_t结构 | 检查源集合和目标集合的状态：只有在源集合没有新分配的内存页，而且存在有效的内存单元; 目标集合没有新分 | 配的内存页，并且内存页指针有空闲的空间(这个可能通过2倍扩展得到);源集合和目标集合的内存页 | 大小都是1M |循环源集合中每个内存单元(事实上它保存的是struct item 结构体)，对于已经被"申请使用"的内存单元进行"清理"：从关联列表中(hashtable)删除该键值assoc_delete(ITEM_key(it), it->nkey)这时还会检查item的引用计数是否为0，如果不是就会设置该忙的状态was_buse = true 从访问历史双向链表中删除对应的item， item_unlink_q(it) 检查item的引用计数，当它是0时，释放该item，item_free(it) |循环检查源集合的内存单元数组，并从中修改指向需要"复制"的内存页中的内存块 | 如果的忙状态为true，那么就会返回-1（从处理流程来看，这时该内存页的内存单元已经清理出了hashtable和访问历史链表。如果返回-1, 客户端就可以在稍后重新提交ressign的请求） | 将内存分页挂到目标集合的新的分页上，并修改对应的计数变量和将item的slabs_clsid设置为0，这个主要是要保此代码对于slabs_clsid检查的一致性3. 名值对hashtable的管理 memcached使用的是hashtable来维护名值对(通过hash值和掩码计算最后的hashtable的下标)，为了降低hash值的碰撞，memcached 使用自动扩展的策略，当hashtable中保存的item数目大于它的大小的1.5倍时，memcached就会实行hashtable的扩展，并把原hanshtable中的元素会重新hash并放到新的hashtable中，而且为了降低查找的延时，这种数据迁移会分散在多次的访问中(后面我们再详细分析)。 现在我们看看相关的几个函数的实现。 assoc_init()分配hashtable的所需的内存 |-->unsigned int hash_size = hashsize(hashpower) * sizeof(void*); primary_hashtable = malloc(hash_size); memset(primary_hashtable, 0, hash_size); assoc_find(key, key_len) 根据键寻找对应的值 | 根据key和key_len计算hash值(hash算法的细节可以参考http://burtleburtle.net/bob/hash/doobs.html) | 根据hash值和掩码计算hashtable的下标 如果当前处于hashtable的扩展过程，并且下标值小于数据迁移的记录值，那么就从新的hashtable中获得该下标对应的item链表,否则就从原来的hashtable中获得item链表 | 循环对比链表中的item的key寻找对应的item assoc_insert(item) 将item加入到hashtable中 | 验证item的key不在hashtable中 | 计算hash值和需要更新的hashtable的下标和assoc_find的算法一样，根据下标和当前是否处于hashtable的扩展过程中来更新旧的hashtable或新的hashtable | 如果当前不是处于扩展状态，那么就检查hashtable中保存的item数是否超过其大小的1.5倍，如果是就进行2倍的容量扩展assoc_expand() assoc_expand() |-->hashtable的2倍容量扩展 | 将hashtable中的第一个下标的item列表重新计算hash值并移到新的hashtable中 (这里只移动了一个下标的item链表do_assoc_move_next_bucket) | 对于其他的元素的迁移会在用户用户请求的时候进行移动，这是把时间消耗分散的延迟处理方式 当元素迁移完成后，就会释放旧的hashtable占用的资源free assoc_delete(key, key_len) 从hashtable中删除对应key的item |-->寻找key对应的元素的指针变量的地址 _hashitem_before | 修改item的h_next指针，从链表中删除该元素4. 数据保存对象struct item结构 memcached在内部使用双向链表维护item的数据，现在我们看看item结构体和几个主要的函数
do_item_alloc(key, key_len, client_flag, expiretime, data_len) 申请足够大的内存单元 | 计算所需的内存单元大小item_make_header(key_len+1(加1表示字符串末尾的"0"), client_flag, data_len, buf, &extral_len)) | 根据所需大小查找内存单元 slabs_clsid(need_size) --> 检查是否存在内存分组 slabs_alloc(need_size) --> slabs_alloc是一个宏，对于多线程模式和单线程模式，它会映射到不同的函数 | 如果slabs_alloc失败，就从保存访问历史的全局变量tails中查找最多50次，得到一个最近最久没有使用的并且引用计数为0的item，将它从hashtable和访问历史链表中"删除"do_item_unlink（do_item_unlink调用本身并不保证item占用的内存返回到可用队列中，只有当item的引用计数变成0时才会进行真正的资源返还，由于在调用do_item_unlink前已经检查了引用计数的值，所以item占用的内存将会变成可用） | 重新调用slabs_alloc请求内存，如果失败就直接返回NULL | 初始化新的item的成员-->需要注意的是next/prev/h_next都会初始化为0，而且引用计算refcount会设置为1,也就是说使用方需要保证最后会将refcount减少1，这里item的状态变量it_flags也会初始化为0 item_free(item*) 删除item的资源，包括hashtable、访问历史链表 heads/tails/sizes, 把item占用的资源返回缓存中 | 将item的状态变量设为ITEM_SLABBED，并将对应的资源返回到缓冲中 slabs_free(item*, item_total_size); item_unlink_q(item *) 从历史链表中删除对应的item item_link_q(item *) 将item加入到历史链表中 这两个函数的细节要结合一起看， item_link_q中是将item依次加入到heads的第一个元素，也就是说它是时间反向的，而tails是时间一致的，最先访问的item在tails的前面，而且tails的指针不是每次都修改的，只有在tails为空的时候才会更新，而链表的构造也是在插入到heads的时候完成了。可以看到item_unlink_q中处理(heads[class_id] == item) 和 (tails[class_id] == item)的情况，它们修改的指针分别是"head = it->next"和"*tail = it->prev;" 这两个函数是不修改item的任何表示状态的变量的 do_item_link(item*) do_item_unlink(item *) 这两个函数分别将item对象加入到历史链表和将item从历史链表中删除，它们首先将item加入hashtabl(或从hashtable删除)，然后分别进行item_link_q/item_unlink_q 需要注意的是，对于do_item_unlink函数，还会检查refcount，当为零时，它会将释放item的资源item_free 这两个函数还会修改item的状态变量"it->it_flags |= ITEM_LINKED/it->it_flags &= ~ITEM_LINKED;" do_item_update(item*)/do_item_replace(item *it, item *new_it) 这两个函数的很相似，do_item_update调用的是item_unlink_q和item_link_q，do_item_replace调用的是do_item_unlink和do_item_link,它们的代码都比较简单。 do_item_get_notedeleted(key, key_len, &delete_lock) 这个函数根据key查找没有被删除的item元素 | 从hashtable中查找对应的item 检查item是否已经超时，这里检查主要是memcached是5秒一次检查超时的，如果在这段时间内获取item就有可能得到实际上已经超时的item, if(!item_delete_lock_over(it)){*delete_lock = false; it = NULL;} | 检查全局设置，如果设置了统一个有效时间，并且当前时间已经超过了这个有效时间，同时item是在有效时间前设置的，那么就删除该item 检查item的超时时间，如果已经超时，就把item删除 do_item_unlink(it); |增加item的引用计数，并返回item指针 do_item_flush_expired() 这个函数是把settting.oldest_live时间后的item全部删除，这个函数的调用主要是处理memcached客户端的"flush_all"指令使用 需要注意的是当client发送"flush_all"时，memcached会修改settting.oldest_live的值，这是会对do_item_get_notedeleted造成影响，因为在下一次调用do_item_get_notedeleted的时候会对settting.oldest_live进行对比，在这个时间之前的item也会被删除，这个也是"flush_all"的分散处理时间的策略，这里也是出于性能因数作的设计。 do_store_item(item* , comm) 根据comm的值(add/replace/set), 处理item的值 这个函数主要是调用其他相关的函数完成功能，需要注意的是它的处理逻辑： "add":如果存在该key对应的非删除的item(do_item_get_notedeleted得到)，那么更新这个item的访问历史，对于新接收到的数据是忽略的，如果由于delete_lock不存在，那么不作任何处理,如果完全没有这个key对应的item， 那么就增加这个key的item(do_item_link(it);) "replace":如果存在该key对应的非删除的item(do_item_get_notedeleted得到),更新这个key对应的值(do_item_replace), 如果由于delete_lock不存在，那么不作任何处理,如果完全没有这个key对应的item， 那么就增加这个key的item(do_item_link(it);) "set":查询key对应的item(do_item_get_nocheck，其实前面已经调用了do_item_get_notedeleted，如果发现是delete_lock,那么调用do_item_get_nocheck)，如果存在那么更新key对应的ite(do_item_replace,对应的item的it_flag会被完全更新), 否则调用 do_item_link将item加入到hashtable和访问历史链表中 item相关的操作函数中还有几个是关于状态查询的，这里就不展开了^_^。5、memcached中表示网络连接的数据抽象 memcached的网络通信使用的是libevent(关于libevent的讨论请参考我的另一篇blog)，同时定义了相关的数据结构，现在我们来看一下：
conn结构体是memcached中成员最多的结构体，下面我们在看看这些成员是怎么使用的。6、memcached的处理流程 现在我们来看看memcached的整个运作流程： 注册SIGINT信号-->简单的退出 signal(SIGINT, sig_handler) | 初始化全局设置settings_init | 取消标准错误输出的缓冲setbuf(stderr, NULL) | 非常经典的解析命令行参数，覆盖默认设置(注意全局变量optarg的使用) while( c = getopt(argc, argv, ...)){ switch(c){....} } | 将系统资源设置到最大的xiandu getrlimit(RLIMIT_CORE, &rlim) <==> setrlimit(RLIMIT_CORE, &rlim_new) | 创建监听socket(这里以TCP socket为例说明) server_socket(settings.port, 0)-->第二个参数0表示这个是tcp socket | |--->创建socket，并将它设置为非阻塞模式 | socket(...) | if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0){...} | 很经典的if语句^_^ | (如果使用ioctrl来设置socket为非阻塞的话{ u_long flag = 1; ioctl(sfd, FIONBIO , &flag);}，即使ioctl返回成 | 功，也不表示该socket已经设置为非阻塞模式了) | | | 将socket设置为地址重用，这个主要是可以在TIME_WAIT状态下马上就能绑定该端口 | setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags)); | 这里总结一下SO_REUSEADDR的作用： | 1、单进程在TIME_WAIT状态下重新绑定一个端口，IP和PORT完全相同 | 2、多网卡，多IP状态下，多进程可以用不同的IP绑定同一个端口 | 3、单进程多IP可以绑定同一个端口 | 4、在UDP多播的应用中，多进程可以用相同的IP和PORT绑定相同的地址 | | | 设置socket的选项： | setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags)); | | | 设置SO_KEEPALIVE选项后，如果2小时(具体时间与TCP协议栈的实现有关)内socket完全空闲，TCP将 | 发送一个主机存活探测包，这是TCP协议中必须响应的包，接受方若正常就以正确ACK包响应，如果接收 | 方崩溃或重启，则以RST包响应，RST接收方设置错误号为ECONNREST，如果探测包接收方无任何响应， | 源自Berkelay的TCP协议栈等待75秒再次发送一个探测包，当一共发送9个探测包仍然没有任何响应时， | 那么发送方放弃并将发送socket的错误号设为ETIMEOUT | | | setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling)); | 设置socket的关闭方式，以struct linger结构调用setsocketopt | 当参数为SO_DONTLINGER时(相当于SO_LINGER,且struct linger 为{0,0}),表示调用closesocket时强制 | 关闭socket，所有悬挂的数据将丢弃，对方recv调用将返回ECONNRESET。如果struct linger结构的 | l_onoff =1 同时l_linger != 0，当阻塞的socket调用closesocket时将一直阻塞直到悬挂的数据发送完或者超 | 时(l_linger表示超时的秒数)。当非阻塞的socket调用closesocket时将返回EWOULDBLOCK/EAGAIN错误 | | | setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags)); | 这是设置数据包的Nagle化，Nagle化是将小的数据组装为一个比较大的帧一次发送的算法，它的处理方式 | 很简单：就是在接收到前一个包的确认到来前一直缓存发送的数据(write调用)，Nagle算法可以缓解网络 | 拥塞。对于一些网络应用，需要把网络数据包组装为一个最大的网络传输单元一次发送可以节省流量，这种 | 情况下可以使用TCP_CORK选项来强制socket使用MSS发送数据，不过这个选项只可以在linux平台上使用 | | | 绑定端口并将socket转换为被动socket | bind(sfd, (struct sockaddr *)&addr, sizeof(addr)) | listen(sfd, 1024) |把权限降低到普通的用户(当然需要启动的用户为root) (getuid/geteuid-->getpwnam(username)-->setgid(pw->pw_gid); setuid(pw->pw_uid)) | 当前进程进入daemon模式 daemon(maxcore, settings.verbose);关于daemon进程相关情况请参考我的另一篇blog，这里不展开了^_^ | 初始化libevent event_init(); | 初始化memcached的运行环境 item_init();-->将历史访问链表初始化为NULL stats_init();-->全局状态初始化，主要是统计信息的初始化，如：命令请求、数据流量等 assoc_init();-->hashtable的初始化，计算hashtable的大小-->分配空间-->初始化空间为NULL conn_init();-->抽象tcp连接的初始化, 分配了一个指向struct conn* 的指针数组， 这个数组有200个元素 slabs_init(mem_limit, unit_factor);-->内存管理初始化 | |-->循环计算每个内存集合的内存单元大小和每个内存页所包含的内存单元数目 | 对于每个内存页的大小是与sizeof(void*)对齐的，而且内存集合的内存单元大小是以factor因子增加的 | 如果当前的内存单元的大小超过0.5M,就好停止扩展，并在最后增加一个内存单元为1M的集合 | (在memcached中内存页大小的上限是1M) | | | 如果编译的时候没有定义DONT_PREALLOC_SLABS而且环境变量中也没有定义T_MEMD_SLABS_ALLOC | 那么memcached就会进行内存的预分配slabs_preallocate(power_largest)-->调用do_slabs_newslab()真正分 | 配内存 | 禁止memcached分配的内存被交换到swap分区上 mlockall(MCL_CURRENT | MCL_FUTURE);-->只对支持的平台调用 | 忽略SIGPIPE信号 sa.sa_handler = SIG_IGN; sa.sa_flags = 0; if (sigemptyset(&sa.sa_mask) == -1 ||sigaction(SIGPIPE, &sa, 0) == -1){...} (对于SIGPIPE信号的默认处理是进程退出，为了memcached的稳定性，这里忽略了这个信号) 说明：对于突然中断的socket，系统会向对方发送RST包，如果这个包在对方进行读操作前到达，那么读取方得到的错误是ECONNRESET，如果RST包在读操作后到达，那么得到的错误是"an unexpected EOF";对于写操作，如果写发生在收到RST前，那么在收到RST时会返回已经发送的字节数 如果再次调用就会得到ECONNRESET错误号, 并触发SIGPIPE信号; 如果再write前就收到了RST包，那么就会直接得到ECONNRESET错误号，而且触发 SIGPIPE信号。 |将监听socket加入到封装到struct conn中，并加入到libevent的监听队列中 conn_new(l_socket, conn_listening, EV_READ | EV_PERSIST, 1, false, main_base) |-->从freeconns中提取一个空闲的conn,conn *c = conn_from_freelist(); | | | 如果没有空闲的conn对象(c == NULL), 构建新的conn | c = (conn *)malloc(sizeof(conn)) | | | 初始化conn 的成员，这里需要分配的内存包括： | rbuf:读缓存，主要用于解析命令，当命令解析后，数据就会放到item的data中 | wbuf:写缓存 | ilist:item列表数组，用于"get"指令中 | iov:iovec结构体数组指针 | msglist:msghdr结构体数组，用于划分过大的数据块 | | | 设置事件的结构 | event_set(&c->event, sfd, event_flags, event_handler, (void *)c); | 设置事件的base成员 | event_base_set(base, &c->event); | 将事件安装到libevent监听的队列中 | event_add(&c->event, 0) | 保存daemon的pid文件 save_pid(getpid(), pid_file) | 初始化多线程的工作环境(建立共享互斥量) thread_init(thread_num, event_base) | |-->初始化mutex和条件变量 | cache的操作互斥量，例如操作item和hashtable等 | pthread_mutex_init(&cache_lock, NULL); | connection list的共享变量互斥量 | pthread_mutex_init(&conn_lock, NULL); | slabs内存单元操作互斥量 | pthread_mutex_init(&slabs_lock, NULL); | 查询memcached状态互斥量 | 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); | | | 分配线程相关的LIBEVENT_THREAD数据结构 | typedef struct { | pthread_t thread_id; 线程id | struct event_base *base; libevent的base结构 | struct event notify_event; 监听通知事件 | int notify_receive_fd; 管道的接收方 | int notify_send_fd; 管道的发送方 | CQ new_conn_queue; 连接队列 | } LIBEVENT_THREAD; | threads = malloc(sizeof(LIBEVENT_THREAD) * nthreads); | | | 循环初始化工作线程相关的LIBEVENT_THREAD结构 | threads[0].base = main_base; | threads[0].thread_id = pthread_self(); | (这里设置0是为了下面代码处理的一致性，它是不另外创建线程运行的) | threads[i].notify_receive_fd = pipe(fds)[0] | threads[i].notify_send_fd = pipe(fds)[1] | setup_thread((LIBEVENT_THREAD * )&threads[i]); | | |-->(i!=0) | | 对于每个线程数据LIBEVENT_THREAD初始化libevent | | pthread_libevnet->base = event_init() | | | | | 设置libevent的事件结构 | | event_set(&pthread_libevnet->notify_event, pthread_libevnet->notify_receive_fd, EV_READ | | | EV_PERSIST, thread_libevent_process, pthread_libevnet); | | event_base_set(pthread_libevnet->base, &pthread_libevnet->notify_event); | | | | | 将事件加入到监听队列中 | | event_add(&pthread_libevnet->notify_event, 0) | | | | | 初始化连接对列 | | cq_init(&pthread_libevnet->new_conn_queue); | | |-->初始化连接对列互斥量 | | pthread_mutex_init(&conn_queue->lock, NULL); | | pthread_cond_init(&conn_queue->cond, NULL); | | | 循环创建工作线程 | create_worker(worker_libevent, &threads[i]), 需要注意的是，这里是从下标1开始循环的，因为0表示主线程 | | |-->初始化线程的属性变量 | | pthread_attr_init(&attr) | | | | | 创建线程pthread_create(&thread, &attr, worker_libevent, &threads[i]) | | | | | 设置条件变量，表示线程已经开始运行 | | pthread_mutex_lock(&init_lock); | | init_count++; | | pthread_cond_signal(&init_cond); | | pthread_mutex_unlock(&init_lock); | | | | | 启动事件的监听event_base_loop(me->base, 0); | | | 等待工作线程初始化结束 | pthread_mutex_lock(&init_lock); | init_count++; // 主线程(接收外部请求的线程)已经初始化完成 | while (init_count < nthreads) { | pthread_cond_wait(&init_cond, &init_lock); | } | pthread_mutex_unlock(&init_lock); | 这里使用了条件变量来表示各个工作线程的初始化, init_count表示已经初始化完成的线程 |注册数据定时器，更新memcached的内部"当前时间" clock_handler(0, 0, 0); |--> 由于定期器是一次性的，所以当定时事件已经在libevent的中时，这里会重新注册 | evtimer_del(&clockevent)-->clockevent是全局变量 | | | 将定时器加入到监听队列中 | evtimer_set(&clockevent, clock_handler, 0)-->可以看到，这里是把定时器的回调函数重新设置为clock_handler | event_base_set(main_base, &clockevent) | evtimer_add(&clock_base, &struct timeval{.tv_set = 1, .tv_usec=0}) | | | 更新memcached的当前时间set_current_time() | 注册删除数据的定时器 |-->delete_handler(0, 0, 0); | | | 这里的流程和clock_handler基本相同,定时时间为5秒 | | | 删除超时的数据 | run_deferred_deletes() | | | (在多线程模式下，这里会先加锁互斥量cache_lock) | do_run_deferred_deletes() | |-->这里会先检查删除对列中item的超时时间，只有大于当前时间的item才会真正删除 | 启动主线程监听外部请求 event_base_loop(main_base, 0); memcached的启动细节我们已经清楚了，下面我们看一下"set-get"的数据处理流程(这里略去了关于udp的设置)。7、memcached的命令处理流程 (这里以"set"-->"get"-->"delete" 命令流程为例) 客户端调用socket的connect连接到memcached服务器 | memcached监测到listen socket可读(libevent的event_base_loop) 调用回调函数event_handler() | 调用状态机处理请求 drive_machine(conn*) |由于监听socket的状态是conn_listening，所以这里调用sfd = accept(c->sfd, &addr, &addrlen) |如果errno == EMFILE(打开了过多的描述符)，那么就关闭监听socket 否则设置sfd为非阻塞状态并创建新的连接(conn)与它对应，新的连接的状态为conn_read 对于单线程模式，dispatch_conn_new是一个指向conn_new函数 对于多线程模式， | dispatch_conn_new的功能是将请求的conn挂到工作线程的conn queue上(这里我们跟踪一下多线程的工作 | 方式) | | | 创建新的connection item | CQ_ITME * item = cqi_new() | | |-->首先检查全局变量cqi_freelist，它指向的是最后一个空闲的CQ_ITEM，如果没有分配或者 | | 没有空闲的，就指向NULL | | | | | 如果cqi_freelist指向NULL，那么一次分配64个CQ_ITEM加入到cqi_freelist中，并将它们 | | "串接"起来。 | | item = malloc(sizeof(CQ_ITEM) * ITEMS_PER_ALLOC); | | pthread_mutex_lock(&cqi_freelist_lock); | | item[ITEMS_PER_ALLOC - 1].next = cqi_freelist; | | cqi_freelist = &item[1]; | | pthread_mutex_unlock(&cqi_freelist_lock); | | | 将请求连接加入到工作线程的处理连接队列中 | cq_push(&threads[thread].new_conn_queue, item); | | |-->借助CQ->lock把item加入到请求对列中 | | pthread_mutex_lock(&cq->lock); | | if (NULL == cq->tail) | | cq->head = item; | | else | | cq->tail->next = item; | | cq->tail = item; | | 这里设计一个条件变量是因为在cq_pop函数中会等待这个条件 | | 但是memcached中并没有调用cq_pop函数，所以我想这可能是历史原因留下来的 | | (也许以前的多线程的工作方式是其他线程一直在等待条件变量来获得新的请求，例如： | | while(conn_item = cq_pop(CQ)){ | | process_the_request; | | } | | 现在的方式是工作线程等待pipe可读，这个表示主线程已经将connection item加入到了 | | conn queue中了，工作线程监测到可读信号后，直接从conn queue取cq_peek就可以了。而 | | 等待时的挂起现在转移到了libevnet中，而不是pthread_cond_wait上了。 | | ) | | pthread_cond_signal(&cq->cond); | | pthread_mutex_unlock(&cq->lock); | | | 通过管道通知工作线程新的请求到来 | write(threads[thread].notify_send_fd, "", 1) | 由于管道是通过系统内部的缓冲来交互数据的，所以写入数据后，读端(libevent)就会收到可读的信号 |主线程返回----------------------->工作线程唤醒调用回调函数 | thread_libevent_process | 从管道中读取一个字符(内容是什么其实不重要，它只是一个通知信号)从而清空可读信号 read(fd, buf, 1) | 提取一个请求连接元素item = cq_peek(&LIBEVENT_THREAD->new_conn_queue)-->通过cq->lock互斥量 得到等待的CQ_ITME，这里是从head指针提取的，和cq_push中对tail指针的操作实对应的，保证了请求 按到来的请求得到处理 | 将连接加入到工作线程的监听队列中 conn_new(item->sfd, item->init_state, item->event_flags, item->read_buffer_size, item->is_udp, me->base); 新的conn的处理函数会被设置为event_handler 需要注意的是:这里的struct event_base是工作线程的base，也就是它是加入到工作线程的监听队列中的 | "释放"CQ_ITEM-->将CQ_ITEM 对象连接到cgi_freelist上(现在客户端socket调用返回成功) | 客户端发送set指令:"set key1 0 10 10 helloworld " | memcached的工作线程发现socket可读，回调event_handler | drive_machine(c); | 当conn状态是conn_read时 尝试解析指令try_read_command()-->tcp面向字节，可能收到不完整的命令 | |-->如果当前接收缓存非空，而且在缓存中找到可识别的命令单元(" "), | | 那么解析命令process_command(c, c->rcurr)，并更新已经处理的缓存内容 | | | | | 增加消息对列add_msghdr(conn * ) | | |-->对比msgsize和msgused的大小，当msgsize==msgused时， | | | 进行2倍扩展c->msglist=realloc(c->msglist, c->msgsize*2*sizeof(struct msghdr)) | | | c->msgsize *= 2 | | | | | | | 初始化struct msghdr结构 | | | memset(msg, 0, sizeof(struct msghdr)); | | | msg->msg_iov = &c->iov[c->iovused]; | | | msg->msg_name = &c->request_addr; | | | msg->msg_namelen = c->request_addr_size; | | | | | 解析指令ntokens = tokenize_command(command, tokens, MAX_TOKENS); | | tokenize_ocmmand只是简单地划分单词 | | | | | 现在的指令是"set key1 0 10 10 helloworld " | | ntokens = 6 | | tokens[COMMAND_TOKEN].value-->"set" | | | | | 处理更新指令 | | process_update_command(conn, tokens, ntokens, NREAD_SET) | | |-->申请内存单元 | | it = item_alloc(key, nkey, flags, realtime(exptime), vlen+2);-->2表示" " | | it的引用计数会变成1 | | | | | 设置conn的成员状态和缓存指针 | | c->item_comm = comm; | | c->item = it; | | c->ritem = ITEM_data(it); | | c->rlbytes = it->nbytes; | | 修改conn的状态conn_set_state(c, conn_nread); | | (现在conn的状态转到接收key对应的数据) | | | 如果但前缓存没有合法的命令那么接收网络数据 | try_read_network(c) | |-->清除已经处理的数据-->memmove | | | 检查已读取的字节和缓冲的长度，当读取的字节大于缓存长度时对缓冲区进行 | 2倍扩展realloc(c->rbuf, c->rsize * 2); | | | 从socket中读取数据并放到缓冲中 | res = read(c->sfd, c->rbuf + c->rbytes, c->rsize - c->rbytes); | c->rbytes += res; | | | 如果read的返回值为0，表示对方socket已经关闭，这时改变conn的状态为conn_closing | 如果返回-1, 而且errno == EAGAIN 或者 errno == EWOULDBLOCK， 表示这时socket的缓存 | 已经清空，这时返回gotdata-->表示是否接收到数据 | 当conn的状态是conn_nread时(通常由存储指令:add/set/replace会转移到这个状态) 检查是否已经读取了命令中指定的字节数(包括数据的" ") 如果读取完，调用complete_nread(c)进行处理 | |-->检查item中接收到的数据的合法性(最后两个字节是" ") | | | 保存item的值store_item(it, c->item_comm) | 如果成功，调用out_string(c, "STORED")输出结果 | out_string会将str写到conn的输出缓存中,并将状态设置为conn_write, 并且c->write_and_go = conn_read | | | 删除item(这里主要是减少引用) | 需要注意的是，虽然这里已经将引用计数变成了0，但是它并不会在释放item所占的内存单元，因为 | 面已经调用了do_item_link(加入到访问历史中)，所以在后面的get指令中会找到这个item | (现在，我们已经将 |
好了，现在度于memcached的工作原理、实现细节和技巧我们都比较清楚了^_^。 memcached中对于错误的处理经常都需要把conn的状态设置为conn_closing,最后我们来看看它做了些什么清除工作：
现在我们对于memcahced的整个构思和实现都有比较清楚的理解了^_^。memcached中还有很多指令我们没有详细跟踪，不过几乎全部的函数的功能我们已经分析了^_^。 8、整体感觉 1、memcached中有很多关于内存使用的细节(如：引用计数，访问历史等)，而且它没有使用*nix系统的链表来组织数据，而是单独实现了功能非常简单的链表这是专门针对cache的功能设计的。 2、在多线程模式中，memcached使用触发器的思想(通过管道触发libevent)，而没有在工作线程中直接等待条件变量，这简化了多线程的协作，也简化了管理逻辑 3、对于pthread condiction variable的使用memcached没有注册退出清除代码，如果在pthread_cont_wait是触发了cannel信号，那么该muxter将会永久被锁定直到进程退出 4、可以说memcached没有资源重用机制(内存页被申请后是不释放的)，如果一个应用开始的时候使用了大量的小片内存，但是一段时间后又使用另一种尺寸的 内存页，那么前面分配的内存页将不会被释放，而且这种情况会持续到memcahed重启。