get

返回Key对应的Value值

add

添加一个Key值，没有则添加成功并提示STORED，有则失败并提示NOT_STORED

set

无条件地设置一个Key值，没有就增加，有就覆盖，操作成功提示STORED

replace

按照相应的Key值替换数据，如果Key值不存在则会操作失败

stats

返回MemCache通用统计信息

stats items

返回各个slab中item的数目和最老的item的年龄

stats slabs

返回MemCache运行期间创建的每个slab的信息

version

返回当前MemCache版本号

flush_all

清空所有键值，但不会删除items，所以此时MemCache依旧占用内存

quit

关闭连接

1. 序号

1. 限制描述

1. 1

1. MemCache中可以保存的item数据量是没有限制的，只要内存足够

1. 2

1. MemCache单进程在32位机中最大使用内存为2G，64位机则没有限制

1. 3

1. Key最大为250个字节，超过该长度无法存储

1. 4

1. 单个item最大数据是1MB，超过1MB的数据不予存储

1. 5

1. MemCache服务端是不安全的，比如已知某个MemCache节点，可以直接telnet过去，并通过flush_all让已经存在的键值对立即失效

1. 6

1. 不能够遍历MemCache中所有的item，因为这个操作的速度相对缓慢且会阻塞其他的操作

1. 7

1. MemCache的高性能源自于两阶段哈希结构：第一阶段在客户端，通过Hash算法根据Key值算出一个节点；第二阶段在服务端，通过一个内部的Hash算法，查找真正的item并返回给客户端。从实现的角度看，MemCache是一个非阻塞的、基于事件的服务器程序

slabs_init

?????? |---> slabs_preallocate

?????????????????????????? |---> do_slabs_newslab

?????????????????????????????????????????? |--->grow_slab_list

?????????????????????????????????????????? |--->split_slab_page_into_freelist

????????

序号

描述

1

根据大小，找到合适的slabclass

2

slabclass空闲列表中是否有空闲的item节点，如果有直接分配item，用于存储内容

3

空闲列表没有空闲的item可以分配，会重新开辟一个slab(默认大小为1M)的内存块，然后切割slab并放入到空闲列表中，然后从空闲列表中获取节点

void *slabs_alloc(size_t size, unsigned int id,

??????? unsigned int flags) {

??? void *ret;

??? pthread_mutex_lock(&slabs_lock);

??? ret = do_slabs_alloc(size, id, flags);

??? pthread_mutex_unlock(&slabs_lock);

??? return ret;

}

/*@null@*/

static void *do_slabs_alloc(const size_t size, unsigned int id,

??????? unsigned int flags) {

??? slabclass_t *p;

??? void *ret = NULL;

??? item *it = NULL;

??? if (id < POWER_SMALLEST || id > power_largest) {

??????? MEMCACHED_SLABS_ALLOCATE_FAILED(size, 0);

??????? return NULL;

??? }

??? p = &slabclass[id];

??? assert(p->sl_curr == 0 || (((item *)p->slots)->it_flags & ITEM_SLABBED));

??? assert(size <= p->size);

??? /* 没有空闲slab时，需要重新分配*/

??? if (p->sl_curr == 0 && flags != SLABS_ALLOC_NO_NEWPAGE) {

??????? do_slabs_newslab(id);

??? }

??? if (p->sl_curr != 0) {

??????? /* return off our freelist */

??????? it = (item *)p->slots;

??????? p->slots = it->next;

??????? if (it->next) it->next->prev = 0;

??????? /* Kill flag and initialize refcount here for lock safety in slab

???????? * mover's freeness detection. */

??????? it->it_flags &= ~ITEM_SLABBED;

??????? it->refcount = 1;

??????? p->sl_curr--;

??????? ret = (void *)it;

??? } else {

??????? ret = NULL;

??? }

??? if (ret) {

??????? MEMCACHED_SLABS_ALLOCATE(size, id, p->size, ret);

??? } else {

??????? MEMCACHED_SLABS_ALLOCATE_FAILED(size, id);

??? }

??? return ret;

}

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);

}

static void do_slabs_free(void *ptr, const size_t size, unsigned int id) {

??? slabclass_t *p;

??? item *it;

??? assert(id >= POWER_SMALLEST && id <= power_largest);

??? if (id < POWER_SMALLEST || id > power_largest)

??????? return;

??? MEMCACHED_SLABS_FREE(size, id, ptr);

??? p = &slabclass[id];

??? it = (item *)ptr;

??? if ((it->it_flags & ITEM_CHUNKED) == 0) {

??????? it->it_flags = ITEM_SLABBED;

??????? it->slabs_clsid = id;

??????? it->prev = 0;

??????? it->next = p->slots;

??????? if (it->next) it->next->prev = it;

??????? p->slots = it;

??????? p->sl_curr++;

??? } else {

??????? do_slabs_free_chunked(it, size);

??? }

??? return;

}

1

Memcached主要是基于Libevent 网络事件库进行开发的

2

Memcached的网络模型分为两部分：主线程和工作线程。主线程主要用来接收客户端的连接信息；工作线程主要用来接管客户端连接，处理具体的业务逻辑

3

主线程和工作线程之间主要是通过pipe管道来进行通信。当主线程接收到客户端的连接的时候，会通过轮询的方式选择一个工作线程，然后向该工作线程的管道pipe写数据。工作线程监听到管道中有数据写入的时候，就会触发代码逻辑去接管客户端的连接

4

每个工作线程也是基于Libevent的事件机制，当客户端有数据写入的时候，就会触发读取的操作

typedef struct conn_queue_item CQ_ITEM;

struct conn_queue_item {

??? int?????????????? sfd;

??? enum conn_states? init_state;

??? int?????????????? event_flags;

??? int?????????????? read_buffer_size;

??? enum network_transport???? transport;

??? enum conn_queue_item_modes mode;

??? conn *c;

??? void??? *ssl;

??? io_pending_t *io; // IO when used for deferred IO handling.

??? STAILQ_ENTRY(conn_queue_item) i_next;

};

/* A connection queue. */

typedef struct conn_queue CQ;

struct conn_queue {

??? STAILQ_HEAD(conn_ev_head, conn_queue_item) head;

??? pthread_mutex_t lock;

??? cache_t *cache; /* freelisted objects */

};

static void cq_init(CQ *cq) {

??? pthread_mutex_init(&cq->lock, 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;

}

/*

?* Adds an item to a connection queue.

?*/

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_mutex_unlock(&cq->lock);

}

int main (int argc, char **argv) {

?????? //检查libevent的版本是否足够新.1.3即可

??? if (!sanitycheck()) {

??????? return EX_OSERR;

??? }

?????? //对memcached的关键设置取默认值

??? settings_init();

?????? ...//解析memcached启动参数

?????? //main_base是一个struct event_base类型的全局变量

??? main_base = event_init();//为主线程创建一个event_base

??? conn_init();//先不管，后面会说到

?????? //创建settings.num_threads个worker线程，并且为每个worker线程创建一个CQ队列

?????? //并为这些worker申请各自的event_base，worker线程然后进入事件循环中??????

??? thread_init(settings.num_threads, main_base);

?????? //设置一个定时event(也叫超时event)，定时(频率为一秒)更新current_time变量

?????? //这个超时event是add到全局变量main_base里面的，所以主线程负责更新current_time(这是一个很重要的全局变量)

??? clock_handler(0, 0, 0);

??? /* create the listening socket, bind it, and init */

??? if (settings.socketpath == NULL) {

??????? FILE *portnumber_file = NULL;

????????????? //创建监听客户端的socket

??????? if (settings.port && server_sockets(settings.port, tcp_transport,//tcp_transport是枚举类型

?????????????????????????????????????????? portnumber_file)) {

??????????? vperror("failed to listen on TCP port %d", settings.port);

??????????? exit(EX_OSERR);

??????? }

???????????? ...

??? }

??? if (event_base_loop(main_base, 0) != 0) {//主线程进入事件循环

??????? retval = EXIT_FAILURE;

??? }

??? return retval;

}

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);

??? } else {

??????? // tokenize them and bind to each one of them..

??????? char *b;

??????? int ret = 0;

??????? char *list = strdup(settings.inter);

??????? for (char *p = strtok_r(list, ";,", &b);

??????????? 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;

??? for (next= ai; next; next= next->ai_next) {

??????? conn *listen_conn_add;

??????? if ((sfd = new_socket(next)) == -1) {

??????????? continue;

??????? }

??????? //todo 设置socket相关的属性，这里省略相关代码

??????? // 绑定socket，省略相关代码

??????? if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) {}

??????? // 暂时只关心TCP协议的，忽略UDP协议实现

??????? if (IS_UDP(transport)) {

??????? } else {

??????????? 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 zero iff we detected no errors in starting up connections */

??? return success == 0;

}

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, void *ssl) {

??? conn *c;

??? assert(sfd >= 0 && sfd < max_fds);

c = conns[sfd];

。。。。。。。。。。。

? ??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;

??? }

??? return c;

}

void event_handler(const evutil_socket_t fd, const short which, void *arg) {

??? conn *c;

??? c = (conn *)arg;

??? assert(c != NULL);

??? c->which = which;

??? /* sanity */

??? 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);

??? /* wait for next event */

??? return;

}

void memcached_thread_init(int nthreads, void *arg) {

??? int???????? i;

??? int???????? power;

??? threads = calloc(nthreads, sizeof(LIBEVENT_THREAD));

? ??for (i = 0; i < nthreads; i++) {

#ifdef HAVE_EVENTFD

??????? threads[i].notify_event_fd = eventfd(0, EFD_NONBLOCK);

??????? if (threads[i].notify_event_fd == -1) {

??????????? perror("failed creating eventfd for worker thread");

??????????? exit(1);

??? ????}

#else

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

#endif

#ifdef EXTSTORE

???? ???threads[i].storage = arg;

#endif

??????? setup_thread(&threads[i]);

??????? /* Reserve three fds for the libevent base, and two for the pipe */

??????? stats_state.reserved_fds += 5;

??? }

??? /* Create threads after we've done all the libevent setup. */

??? for (i = 0; i < nthreads; i++) {

??????? create_worker(worker_libevent, &threads[i]);

??? }

??? /* Wait for all the threads to set themselves up before returning. */

??? pthread_mutex_lock(&init_lock);

??? wait_for_thread_registration(nthreads);

??? pthread_mutex_unlock(&init_lock);

}

static void create_worker(void *(*func)(void *), void *arg) {

??? pthread_attr_t? attr;

??? int???????????? ret;

??? pthread_attr_init(&attr);

??? if ((ret = pthread_create(&((LIBEVENT_THREAD*)arg)->thread_id, &attr, func, arg)) != 0) {

??????? fprintf(stderr, "Can't create thread: %s\n",

??????????????? strerror(ret));

??????? exit(1);

??? }

}

static void *worker_libevent(void *arg) {

??? LIBEVENT_THREAD *me = arg;

??? /* Any per-thread setup can happen here; memcached_thread_init() will block until

???? * all threads have finished initializing.

???? */

??? me->l = logger_create();

??? me->lru_bump_buf = item_lru_bump_buf_create();

??? if (me->l == NULL || me->lru_bump_buf == NULL) {

??????? abort();

??? }

??? if (settings.drop_privileges) {

??????? drop_worker_privileges();

??? }

?? ?register_thread_initialized();

? ??event_base_loop(me->base, 0);

??? // same mechanism used to watch for all threads exiting.

?? ?register_thread_initialized();

??? event_base_free(me->base);

??? return NULL;

}

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;

#ifdef HAVE_ACCEPT4

??? static int? use_accept4 = 1;

#else

??? static int? use_accept4 = 0;

#endif

??? assert(c != NULL);

??? while (!stop) {

??????? switch(c->state) {

??????? case conn_listening:

??????????? addrlen = sizeof(addr);

??????????? sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen);

?????????? // 中间省略一系列的socket相关的初始化工作???????????

??????????? if (settings.maxconns_fast &&

??????????? } else {

??????????????? dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST,

???????????????????????????????????? DATA_BUFFER_SIZE, c->transport);

????????? ??}

??????????? stop = true;

??????????? break;

void dispatch_conn_new(int sfd, enum conn_states init_state, int event_flags,

?????????????????????? int read_buffer_size, enum network_transport transport, void *ssl) {

??? CQ_ITEM *item = NULL;

??? LIBEVENT_THREAD *thread;

??? if (!settings.num_napi_ids)

??????? thread = select_thread_round_robin();

??? else

??????? thread = select_thread_by_napi_id(sfd);

??? item = cqi_new(thread->ev_queue);

??? 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;

??? }

??? item->sfd = sfd;

??? item->init_state = init_state;

??? item->event_flags = event_flags;

??? item->read_buffer_size = read_buffer_size;

??? item->transport = transport;

??? item->mode = queue_new_conn;

??? item->ssl = ssl;

??? MEMCACHED_CONN_DISPATCH(sfd, (int64_t)thread->thread_id);

??? notify_worker(thread, item);

}

static void notify_worker(LIBEVENT_THREAD *t, CQ_ITEM *item) {

??? cq_push(t->ev_queue, item);

#ifdef HAVE_EVENTFD

??? uint64_t u = 1;

??? if (write(t->notify_event_fd, &u, sizeof(uint64_t)) != sizeof(uint64_t)) {

??????? perror("failed writing to worker eventfd");

??????? /* TODO: This is a fatal problem. Can it ever happen temporarily? */

??? }

#else

??? char buf[1] = "c";

??? if (write(t->notify_send_fd, buf, 1) != 1) {

??????? perror("Failed writing to notify pipe");

??????? /* TODO: This is a fatal problem. Can it ever happen temporarily? */

??? }

#endif

}

static void thread_libevent_process(evutil_socket_t fd, short which, void *arg) {

??? LIBEVENT_THREAD *me = arg;

??? CQ_ITEM *item;

??? conn *c;

??? uint64_t ev_count = 0; // max number of events to loop through this run.

#ifdef HAVE_EVENTFD

??? if (read(fd, &ev_count, sizeof(uint64_t)) != sizeof(uint64_t)) {

??????? if (settings.verbose > 0)

??????????? fprintf(stderr, "Can't read from libevent pipe\n");

??????? return;

??? }

#else

??? char buf[MAX_PIPE_EVENTS];

??? ev_count = read(fd, buf, MAX_PIPE_EVENTS);

??? if (ev_count == 0) {

??????? if (settings.verbose > 0)

??????????? fprintf(stderr, "Can't read from libevent pipe\n");

??????? return;

??? }

#endif

??? for (int x = 0; x < ev_count; x++) {

??????? item = cq_pop(me->ev_queue);

??????? if (item == NULL) {

??????????? return;

??????? }

??????? switch (item->mode) {

??????????? case queue_new_conn:

?????? ?????????c = conn_new(item->sfd, item->init_state, item->event_flags,

?????????????????????????????????? item->read_buffer_size, item->transport,

?????????????????????????????????? me->base, item->ssl);

序号

描述

1

当客户端和Memcached建立TCP连接后，Memcached会基于Libevent的event事件来监听客户端是否有可以读取的数据

2

当客户端有命令数据报文上报的时候，就会触发drive_machine方法中的conn_read这个Case，在进入这个状态之前经过conn_new_cmd->conn_waiting->conn_read的流程

3

memcached通过try_read_network方法读取客户端的报文。如果读取失败，则返回conn_closing，去关闭客户端的连接；如果没有读取到任何数据，则会返回conn_waiting，继续等待客户端的事件到来，并且退出drive_machine的循环；如果数据读取成功，则会将状态转交给conn_parse_cmd处理，读取到的数据会存储在c->rbuf容器中

4

conn_parse_cmd主要的工作就是用来解析命令。主要通过try_read_command这个方法来读取c->rbuf中的命令数据，通过\n来分隔数据报文的命令。如果c->buf内存块中的数据匹配不到\n，则返回继续等待客户端的命令数据报文到来conn_waiting；否则就会转交给process_command方法，来处理具体的命令（命令解析会通过\0符号来分隔）

5

process_command主要用来处理具体的命令。其中tokenize_command这个方法非常重要，将命令拆解成多个元素（KEY的最大长度250）。例如我们以get命令为例，最终会跳转到process_get_command这个命令 process_*_command这一系列就是处理具体的命令逻辑的

6

我们进入process_get_command，当获取数据处理完毕之后，会转交到conn_mwrite这个状态。如果获取数据失败，则关闭连接

7

进入conn_mwrite后，主要是通过transmit方法来向客户端提交数据。如果写数据失败，则关闭连接或退出drive_machine循环；如果写入成功，则又转交到conn_new_cmd这个状态

8

conn_new_cmd这个状态主要是处理c->rbuf中剩余的命令。主要看一下reset_cmd_handler这个方法，这个方法会去判断c->rbytes中是否还有剩余的报文没处理，如果未处理，则转交到conn_parse_cmd（第四步）继续解析剩余命令；如果已经处理了，则转交到conn_waiting，等待新的事件到来。在转交之前，每次都会执行一次conn_shrink方法

9

conn_shrink方法主要用来处理命令报文容器c->rbuf和输出内容的容器是否数据满了？是否需要扩大buffer的大小，是否需要移动内存块。接受命令报文的初始化内存块大小2048，最大8192

状态机

说明

conn_new_cmd

主线程通过调用 dispatch_conn_new，把新连接调度给工作线程后，worker 线程创建 conn 对象，这个连接初始状态就是 conn_new_cmd。除了通过新建连接进入 conn_new_cmd 状态之外，如果连接命令处理完毕，准备接受新指令时，也会将连接的状态设置为 conn_new_cmd 状态。

进入 conn_new_cmd 后，工作线程会调用 reset_cmd_handler 函数，重置 conn 的 cmd 和 substate 字段，并在必要时对连接 buf 进行收缩。因为连接在处理 client 来的命令时，对于写指令，需要分配较大的读 buf 来存待更新的 key value，而对于读指令，则需要分配较大的写 buf 来缓冲待发送给 client 的 value 结果。持续运行中，随着大 size value 的相关操作，这些缓冲会占用很多内存，所以需要设置一个阀值，超过阀值后就进行缓冲内存收缩，避免连接占用太多内存。在后端服务以及中间件开发中，这个操作很重要，因为线上服务的连接很容易达到万级别，如果一个连接占用几十 KB 以上的内存，后端系统仅连接就会占用数百 MB 甚至数 GB 以上的内存空间。

工作线程处理完 conn_new_cmd 状态的主要逻辑后，如果读缓冲区有数据可以读取，则进入 conn_parse_cmd 状态，否则就会进入到 conn_waiting 状态，等待网络数据进来。

conn_waiting

连接进入 conn_waiting 状态后，处理逻辑很简单，直接通过 update_event 函数注册读事件即可，之后会将连接状态更新为 conn_read。

conn_read

当工作线程监听到网络数据进来，连接就进入 conn_read 状态。对 conn_read 的处理，是通过 try_read_network 从 socket 中读取网络数据。如果读取失败，则进入 conn_closing 状态，关闭连接。如果没有读取到任何数据，则会返回 conn_waiting，继续等待 client 端的数据到来。如果读取数据成功，则会将读取的数据存入 conn 的 rbuf 缓冲，并进入 conn_parse_cmd 状态，准备解析 cmd

conn_parse_cmd

conn_parse_cmd 状态的处理逻辑就是解析命令。工作线程首先通过 try_read_command 读取连接的读缓冲，并通过 \n 来分隔数据报文的命令。如果命令首行长度大于 1024，关闭连接，这就意味着 key 长度加上其他各项命令字段的总长度要小于 1024 字节。当然对于 key，Mc 有个默认的最大长度，key_max_length，默认设置为 250 字节。校验完毕首行报文的长度，接下来会在 process_command 函数中对首行指令进行处理。

备注：

conn_parse_cmd 的状态处理，只有读取到 \n，有了完整的命令首行协议，才会进入 process_command，否则会跳转到 conn_waiting，继续等待客户端的命令数据报文。在 process_command 处理中，如果是获取类命令，在获取到 key 对应的 value 后，则跳转到 conn_mwrite，准备写响应给连接缓冲。而对于 update 变更类型的指令，则需要继续读取 value 数据，此时连接会跳转到 conn_nread 状态。在 conn_parse_cmd 处理过程中，如果遇到任何失败，都会跳转到 conn_closing 关闭连接

conn_write

连接 conn_write 状态处理逻辑很简单，直接进入 conn_mwrite 状态。或者当 conn 的 iovused 为 0 或对于 udp 协议，将响应写入 conn 消息缓冲后，再进入 conn_mwrite 状态。

conn_mwrite

进入 conn_mwrite 状态后，工作线程将通过 transmit 来向客户端写数据。如果写数据失败，跳转到 conn_closing，关闭连接退出状态机。如果写数据成功，则跳转到 conn_new_cmd，准备下一次新指令的获取

conn_closing

最后一个 conn_closing 状态，前面提到过很多次，在任何状态的处理过程中，如果出现异常，就会进入到这个状态，关闭连接。

??????? case conn_new_cmd:

??????????? /* Only process nreqs at a time to avoid starving other

?????????????? connections */

??????????? --nreqs;

??????????? if (nreqs >= 0) {

??????????????? reset_cmd_handler(c);

??????????? } else if (c->resp_head) {

??????????????? // flush response pipe on yield.

??????????????? conn_set_state(c, conn_mwrite);

??????????? } else {

??????????????? pthread_mutex_lock(&c->thread->stats.mutex);

??????????????? c->thread->stats.conn_yields++;

??????????????? pthread_mutex_unlock(&c->thread->stats.mutex);

??????????????? if (c->rbytes > 0) {

????????? ??????????/* We have already read in data into the input buffer,

?????????????????????? so libevent will most likely not signal read events

?????????????????????? on the socket (unless more data is available. As a

?????????????????????? hack we should just put in a request to write data,

?????????????????????? because that should be possible ;-)

??????????????????? */

??????????????????? 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);

??????????????????????? break;

??????????????????? }

??????????????? }

??????????????? stop = true;

??????????? }

static void reset_cmd_handler(conn *c) {

??? c->cmd = -1;

??? c->substate = bin_no_state;

??? if (c->item != NULL) {

??????? // TODO: Any other way to get here?

??????? // SASL auth was mistakenly using it. Nothing else should?

??????? if (c->item_malloced) {

??????????? free(c->item);

???? ???????c->item_malloced = false;

??????? } else {

??????????? item_remove(c->item);

??????? }

??????? c->item = NULL;

??? }

??? if (c->rbytes > 0) {

??????? conn_set_state(c, conn_parse_cmd);

??? } else if (c->resp_head) {

??????? conn_set_state(c, conn_mwrite);

??? } else {

??????? conn_set_state(c, conn_waiting);

??? }

}

??????? case conn_waiting:

??????????? rbuf_release(c);

??????????? if (!update_event(c, EV_READ | EV_PERSIST)) {

??????????????? if (settings.verbose > 0)

??????????????????? fprintf(stderr, "Couldn't update event\n");

??????????????? conn_set_state(c, conn_closing);

??????????????? break;

??????????? }

??????????? conn_set_state(c, conn_read);

??????????? stop = true;

??????????? break;

??????? case conn_read:

??????????? if (!IS_UDP(c->transport)) {

??????????????? // Assign a read buffer if necessary.

??????????????? if (!rbuf_alloc(c)) {

??????????????????? // TODO: Some way to allow for temporary failures.

??????????????????? conn_set_state(c, conn_closing);

??????????????????? break;

??????????????? }

??????????????? res = try_read_network(c);

??????????? } else {

??????????????? // UDP connections always have a static buffer.

??????????????? res = try_read_udp(c);

??????????? }

??????????? switch (res) {

??????????? case READ_NO_DATA_RECEIVED:

??????????????? conn_set_state(c, conn_waiting);

??????????????? break;

??????????? case READ_DATA_RECEIVED:

??????????????? conn_set_state(c, conn_parse_cmd);

??????????????? break;

??????????? case READ_ERROR:

? ??????????????conn_set_state(c, conn_closing);

??????????????? break;

??????????? case READ_MEMORY_ERROR: /* Failed to allocate more memory */

??????????????? /* State already set by try_read_network */

??????????????? break;

??????????? }

??????????? break;

??????? case conn_parse_cmd:

??????????? c->noreply = false;

??????????? if (c->try_read_command(c) == 0) {

??????????????? /* we need more data! */

??????????????? if (c->resp_head) {

??????????????????? // Buffered responses waiting, flush in the meantime.

??????????????????? conn_set_state(c, conn_mwrite);

??????????????? } else {

???? ???????????????conn_set_state(c, conn_waiting);

??????????????? }

??????????? }

??????????? break;

int try_read_command_ascii(conn *c) {

??? char *el, *cont;

??? if (c->rbytes == 0)

??????? return 0;

??? el = memchr(c->rcurr, '\n', c->rbytes);

??? if (!el) {

??????? if (c->rbytes > 2048) {

??????????? char *ptr = c->rcurr;

??????????? while (*ptr == ' ') { /* ignore leading whitespaces */

??????????????? ++ptr;

??????????? }

??????????? if (ptr - c->rcurr > 100 ||

??????????????? (strncmp(ptr, "get ", 4) && strncmp(ptr, "gets ", 5))) {

??????????????? conn_set_state(c, conn_closing);

??????????????? return 1;

??????????? }

??????????? if (!c->rbuf_malloced) {

??????????????? if (!rbuf_switch_to_malloc(c)) {

??????????????????? conn_set_state(c, conn_closing);

??????????????????? return 1;

??????????????? }

??????????? }

??????? }

??????? return 0;

??? }

??? cont = el + 1;

??? if ((el - c->rcurr) > 1 && *(el - 1) == '\r') {

??????? el--;

??? }

??? *el = '\0';

??? assert(cont <= (c->rcurr + c->rbytes));

??? c->last_cmd_time = current_time;

??? process_command_ascii(c, c->rcurr);

??? c->rbytes -= (cont - c->rcurr);

??? c->rcurr = cont;

??? assert(c->rcurr <= (c->rbuf + c->rsize));

??? return 1;

}

static size_t tokenize_command(char *command, token_t *tokens, const size_t max_tokens) {

??? char *s, *e;

??? size_t ntokens = 0;

??? assert(command != NULL && tokens != NULL && max_tokens > 1);

??? size_t len = strlen(command);

??? unsigned int i = 0;

??? s = e = command;

??? for (i = 0; i < len; i++) {

??????? if (*e == ' ') {

??????????? if (s != e) {

??????????????? tokens[ntokens].value = s;

??????????????? tokens[ntokens].length = e - s;

??????????????? ntokens++;

??????????????? *e = '\0';

??????????????? if (ntokens == max_tokens - 1) {

??????????????????? e++;

??????????????????? s = e; /* so we don't add an extra token */

??????????????????? break;

??????????????? }

??????????? }

??????????? s = e + 1;

??????? }

??????? e++;

??? }

??? if (s != e) {

??????? tokens[ntokens].value = s;

??????? tokens[ntokens].length = e - s;

??????? ntokens++;

??? }

??? /*

???? * If we scanned the whole string, the terminal value pointer is null,

???? * otherwise it is the first unprocessed character.

???? */

??? tokens[ntokens].value =? *e == '\0' ? NULL : e;

??? tokens[ntokens].length = 0;

??? ntokens++;

??? return ntokens;

}

void process_command_ascii(conn *c, char *command) {

??? token_t tokens[MAX_TOKENS];

??? size_t ntokens;

??? int comm;

??? assert(c != NULL);

??? MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes);

??? if (settings.verbose > 1)

??????? fprintf(stderr, "<%d %s\n", c->sfd, command);

??? // Prep the response object for this query.

??? if (!resp_start(c)) {

??????? conn_set_state(c, conn_closing);

??????? return;

??? }

??? ntokens = tokenize_command(command, tokens, MAX_TOKENS);

??? // All commands need a minimum of two tokens: cmd and NULL finalizer

??? // There are also no valid commands shorter than two bytes.

??? if (ntokens < 2 || tokens[COMMAND_TOKEN].length < 2) {

??????? out_string(c, "ERROR");

??????? return;

??? }

??? // Meta commands are all 2-char in length.

??? char first = tokens[COMMAND_TOKEN].value[0];

??? if (first == 'm' && tokens[COMMAND_TOKEN].length == 2) {

??????? switch (tokens[COMMAND_TOKEN].value[1]) {

????????? …………………………………………..

??? } else if (first == 'g') {

??????? // Various get commands are very common.

??????? WANT_TOKENS_MIN(ntokens, 3);

??????? if (strcmp(tokens[COMMAND_TOKEN].value, "get") == 0) {

??????????? process_get_command(c, tokens, ntokens, false, false);

??????? } else if (strcmp(tokens[COMMAND_TOKEN].value, "gets") == 0) {

??????????? process_get_command(c, tokens, ntokens, true, false);

??????? } else if (strcmp(tokens[COMMAND_TOKEN].value, "gat") == 0) {

??????????? process_get_command(c, tokens, ntokens, false, true);

??????? } else if (strcmp(tokens[COMMAND_TOKEN].value, "gats") == 0) {

??????????? process_get_command(c, tokens, ntokens, true, true);

??????? } else {

??????????? out_string(c, "ERROR");

??????? }

??? } else if (first == 's') {

??????? if (strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) {

??????????? WANT_TOKENS_OR(ntokens, 6, 7);

??????????? process_update_command(c, tokens, ntokens, comm, false);

??????? } else if (strcmp(tokens[COMMAND_TOKEN].value, "stats") == 0) {

??????????? process_stat(c, tokens, ntokens);

??????? } else if (strcmp(tokens[COMMAND_TOKEN].value, "shutdown") == 0) {

??????????? process_shutdown_command(c, tokens, ntokens);

??????? } else if (strcmp(tokens[COMMAND_TOKEN].value, "slabs") == 0) {

??????????? process_slabs_command(c, tokens, ntokens);

??????? } else {

??????????? out_string(c, "ERROR");

??????? }

………………………………………..

static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas) {

? …………………………………………………….

??? item *it;

??? assert(c != NULL);

??? set_noreply_maybe(c, tokens, ntokens);

??? if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) {

??????? out_string(c, "CLIENT_ERROR bad command line format");

??????? return;

??? }

??? key = tokens[KEY_TOKEN].value;

??? nkey = tokens[KEY_TOKEN].length;

??? if (! (safe_strtoul(tokens[2].value, (uint32_t *)&flags)

?????????? && safe_strtol(tokens[3].value, &exptime_int)

?????????? && safe_strtol(tokens[4].value, (int32_t *)&vlen))) {

??????? out_string(c, "CLIENT_ERROR bad command line format");

??????? return;

??? }

??? exptime = realtime(EXPTIME_TO_POSITIVE_TIME(exptime_int));

??? // does cas value exist?

??? if (handle_cas) {

??????? if (!safe_strtoull(tokens[5].value, &req_cas_id)) {

??????????? out_string(c, "CLIENT_ERROR bad command line format");

??????????? return;

??????? }

??? }

??? if (vlen < 0 || vlen > (INT_MAX - 2)) {

??????? out_string(c, "CLIENT_ERROR bad command line format");

??????? return;

??? }

??? vlen += 2;

it = item_alloc(key, nkey, flags, exptime, vlen);

…………………………………………………

? ??ITEM_set_cas(it, req_cas_id);

??? c->item = it;

??? c->ritem = ITEM_data(it);

??? c->rlbytes = it->nbytes;

??? c->cmd = comm;

??? conn_set_state(c, conn_nread);

}

??????? case conn_nread:

??????????? if (c->rlbytes == 0) {

??????????????? complete_nread(c);

??????????????? break;

??????????? }

??????????? /* Check if rbytes < 0, to prevent crash */

??????????? if (c->rlbytes < 0) {

??????????????? if (settings.verbose) {

??????????????????? fprintf(stderr, "Invalid rlbytes to read: len %d\n", c->rlbytes);

??????????????? }

??????????????? conn_set_state(c, conn_closing);

??????????????? break;

??????????? }

??????????? if (c->item_malloced || ((((item *)c->item)->it_flags & ITEM_CHUNKED) == 0) ) {

??????????????? /* first check if we have leftovers in the conn_read buffer */

??????????????? if (c->rbytes > 0) {

??????????????????? int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes;

??????????????????? memmove(c->ritem, c->rcurr, tocopy);

??????????????????? c->ritem += tocopy;

??????????????????? c->rlbytes -= tocopy;

??????????????????? c->rcurr += tocopy;

??????????????????? c->rbytes -= tocopy;

??????????????????? if (c->rlbytes == 0) {

??????????????????????? break;

??????????????????? }

??????????????? }

??????????????? /*? now try reading from the socket */

??????????????? res = c->read(c, c->ritem, c->rlbytes);

??????????????? if (res > 0) {

??????????????????? pthread_mutex_lock(&c->thread->stats.mutex);

??????????????????? c->thread->stats.bytes_read += res;

??????????????????? pthread_mutex_unlock(&c->thread->stats.mutex);

??????????????????? if (c->rcurr == c->ritem) {

??????????????????? ????c->rcurr += res;

??????????????????? }

??????????????????? c->ritem += res;

??????????????????? c->rlbytes -= res;

??????????????????? break;

??????????????? }

??????????? } else {

??????????????? res = read_into_chunked_item(c);

??????????????? if (res > 0)

??????????????????? break;

??????????? }

??????????? if (res == 0) { /* end of stream */

??????????????? c->close_reason = NORMAL_CLOSE;

??????????????? conn_set_state(c, conn_closing);

??????????????? break;

??????????? }

??????????? if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {

??????????????? if (!update_event(c, EV_READ | EV_PERSIST)) {

??????????????????? if (settings.verbose > 0)

??????????????????????? fprintf(stderr, "Couldn't update event\n");

??????????????????? conn_set_state(c, conn_closing);

??????????????????? break;

??????????????? }

??????????????? stop = true;

??????????????? break;

??????????? }

??????????? /* Memory allocation failure */

??????????? if (res == -2) {

??????????????? out_of_memory(c, "SERVER_ERROR Out of memory during read");

??????????????? c->sbytes = c->rlbytes;

??????????????? conn_set_state(c, conn_swallow);

??????????????? // Ensure this flag gets cleared. It gets killed on conn_new()

??????????? ????// so any conn_closing is fine, calling complete_nread is

??????????????? // fine. This swallow semms to be the only other case.

??????????????? c->set_stale = false;

??????????????? c->mset_res = false;

??????????????? break;

??????????? }

?????????? ?/* otherwise we have a real error, on which we close the connection */

??????????? if (settings.verbose > 0) {

??????????????? fprintf(stderr, "Failed to read, and not due to blocking:\n"

??????????????????????? "errno: %d %s \n"

??????????????????????? "rcurr=%p ritem=%p rbuf=%p rlbytes=%d rsize=%d\n",

??????????????????????? errno, strerror(errno),

??????????????????????? (void *)c->rcurr, (void *)c->ritem, (void *)c->rbuf,

??????????????????????? (int)c->rlbytes, (int)c->rsize);

??????????? }

????????? ??conn_set_state(c, conn_closing);

??????????? break;

void complete_nread_ascii(conn *c) {

??? assert(c != NULL);

??? item *it = c->item;

??? int comm = c->cmd;

??? enum store_item_type ret;

??? bool is_valid = false;

??? pthread_mutex_lock(&c->thread->stats.mutex);

??? c->thread->stats.slab_stats[ITEM_clsid(it)].set_cmds++;

??? pthread_mutex_unlock(&c->thread->stats.mutex);

??? if ((it->it_flags & ITEM_CHUNKED) == 0) {

??????? if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) == 0) {

??????????? is_valid = true;

??????? }

??? }

??? if (!is_valid) {

??????? // metaset mode always returns errors.

??????? if (c->mset_res) {

??????????? c->noreply = false;

??????? }

??????? out_string(c, "CLIENT_ERROR bad data chunk");

??? } else {

???? ?ret = store_item(it, comm, c);

………………………………………….

????? if (c->mset_res) {

????????? _finalize_mset(c, ret);

????? } else {

????????? switch (ret) {

????????? case STORED:

????????????? out_string(c, "STORED");

????????????? break;

????????? case EXISTS:

????????????? out_string(c, "EXISTS");

????????????? break;

????????? case NOT_FOUND:

????????????? out_string(c, "NOT_FOUND");

????????? ????break;

????????? case NOT_STORED:

????????????? out_string(c, "NOT_STORED");

????????????? break;

????????? default:

????????????? out_string(c, "SERVER_ERROR Unhandled storage type.");

????????? }

????? }

??? }

??? c->set_stale = false; /* force flag to be off just in case */

??? c->mset_res = false;

??? item_remove(c->item);?????? /* release the c->item reference */

??? c->item = 0;

}

void out_string(conn *c, const char *str) {

??? size_t len;

??? assert(c != NULL);

??? mc_resp *resp = c->resp;

??? // if response was original filled with something, but we're now writing

??? // out an error or similar, have to reset the object first.

??? // TODO: since this is often redundant with allocation, how many callers

?? ?// are actually requiring it be reset? Can we fast test by just looking at

??? // tosend and reset if nonzero?

??? resp_reset(resp);

??? if (c->noreply) {

??????? // TODO: just invalidate the response since nothing's been attempted

??????? // to send yet?

??????? resp->skip = true;

??????? if (settings.verbose > 1)

??????????? fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str);

??????? conn_set_state(c, conn_new_cmd);

??????? return;

??? }

??? if (settings.verbose > 1)

??????? fprintf(stderr, ">%d %s\n", c->sfd, str);

??? // Fill response object with static string.

??? len = strlen(str);

??? if ((len + 2) > WRITE_BUFFER_SIZE) {

??????? /* ought to be always enough. just fail for simplicity */

??????? str = "SERVER_ERROR output line too long";

??????? len = strlen(str);

??? }

??? memcpy(resp->wbuf, str, len);

??? memcpy(resp->wbuf + len, "\r\n", 2);

??? resp_add_iov(resp, resp->wbuf, len + 2);

??? conn_set_state(c, conn_new_cmd);

??? return;

}

??????? case conn_write:

??????? case conn_mwrite:

? ????????????for (io_queue_t *q = c->io_queues; q->type != IO_QUEUE_NONE; q++) {

??????????????? if (q->stack_ctx != NULL) {

??????????????????? io_queue_cb_t *qcb = thread_io_queue_get(c->thread, q->type);

??????????????????? qcb->submit_cb(q);

??????????????????? c->io_queues_submitted++;

??????????????? }

??????????? }

??????????? if (c->io_queues_submitted != 0) {

??????????????? conn_set_state(c, conn_io_queue);

????? ??????????event_del(&c->event);

?? ?????????????stop = true;

??????????????? break;

??????????? }

??????????? switch (!IS_UDP(c->transport) ? transmit(c) : transmit_udp(c)) {

??????????? case TRANSMIT_COMPLETE:

??????????????? if (c->state == conn_mwrite) {

??????????????????? // Free up IO wraps and any half-uploaded items.

??????????????????? conn_release_items(c);

??????????????????? conn_set_state(c, conn_new_cmd);

??????????????????? if (c->close_after_write) {

??????????????????????? conn_set_state(c, conn_closing);

????????????????? ??}

??????????????? } else {

??????????????????? if (settings.verbose > 0)

??????????????????????? fprintf(stderr, "Unexpected state %d\n", c->state);

??????????????????? conn_set_state(c, conn_closing);

??????????????? }

??????????????? break;

序号

描述

1

Memcached在启动的时候，会默认初始化一个HashTable，这个table的默认长度为65536

2

我们将这个HashTable中的每一个元素称为桶，每个桶就是一个item结构的单向链表

3

Memcached会将key值hash成一个变量名称为hv的uint32_t类型的值

4

通过hv与桶的个数之间的按位与计算，hv & hashmask(hashpower)，就可以得到当前的key会落在哪个桶上面

5

然后会将item挂到这个桶的链表上面。链表主要是通过item结构中的h_next实现

void assoc_init(const int hashtable_init) {

??? if (hashtable_init) {

??????? hashpower = hashtable_init;

??? }

??? primary_hashtable = calloc(hashsize(hashpower), sizeof(void *));

??? if (! primary_hashtable) {

??????? fprintf(stderr, "Failed to init hashtable.\n");

??????? exit(EXIT_FAILURE);

??? }

??? STATS_LOCK();

??? stats_state.hash_power_level = hashpower;

? ??stats_state.hash_bytes = hashsize(hashpower) * sizeof(void *);

??? STATS_UNLOCK();

}

typedef struct _stritem {

??? /* Protected by LRU locks */

??? struct _stritem *next;

??? struct _stritem *prev;

??? /* Rest are protected by an item lock */

??? struct _stritem *h_next;??? /* hash chain next */

??? rel_time_t????? time;?????? /* least recent access */

??? rel_time_t????? exptime;??? /* expire time */

??? int???????????? nbytes;???? /* size of data */

??? unsigned short? refcount;

??? uint16_t??????? it_flags;?? /* ITEM_* above */

??? uint8_t???????? slabs_clsid;/* which slab class we're in */

??? uint8_t???????? nkey;?????? /* key length, w/terminating null and padding */

??? /* this odd type prevents type-punning issues when we do

???? * the little shuffle to save space when not using CAS. */

??? union {

??????? uint64_t cas;

??????? char end;

??? } data[];

??? /* if it_flags & ITEM_CAS we have 8 bytes CAS */

??? /* then null-terminated key */

??? /* then " flags length\r\n" (no terminating null) */

??? /* then data with terminating \r\n (no terminating null; it's binary!) */

} item;

item *assoc_find(const char *key, const size_t nkey, const uint32_t hv) {

??? item *it;

??? uint64_t oldbucket;

??? if (expanding &&

??????? (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket)

??? {

??????? it = old_hashtable[oldbucket];

??? } else {

??????? it = primary_hashtable[hv & hashmask(hashpower)];

? ??}

??? item *ret = NULL;

??? int depth = 0;

??? while (it) {

??????? if ((nkey == it->nkey) && (memcmp(key, ITEM_key(it), nkey) == 0)) {

??????????? ret = it;

??????????? break;

??????? }

??????? it = it->h_next;

??????? ++depth;

??? }

??? MEMCACHED_ASSOC_FIND(key, nkey, depth);

??? return ret;

}

int assoc_insert(item *it, const uint32_t hv) {

??? uint64_t oldbucket;

//??? assert(assoc_find(ITEM_key(it), it->nkey) == 0);? /* shouldn't have duplicately named things defined */

??? if (expanding &&

??????? (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket)

??? {

??????? it->h_next = old_hashtable[oldbucket];

??????? old_hashtable[oldbucket] = it;

??? } else {

??????? it->h_next = primary_hashtable[hv & hashmask(hashpower)];

??????? primary_hashtable[hv & hashmask(hashpower)] = it;

??? }

??? MEMCACHED_ASSOC_INSERT(ITEM_key(it), it->nkey);

??? return 1;

}

void assoc_delete(const char *key, const size_t nkey, const uint32_t hv) {

??? item **before = _hashitem_before(key, nkey, hv);

??? if (*before) {

??????? item *nxt;

??????? /* The DTrace probe cannot be triggered as the last instruction

???????? * due to possible tail-optimization by the compiler

???????? */

??????? MEMCACHED_ASSOC_DELETE(key, nkey);

??????? nxt = (*before)->h_next;

??????? (*before)->h_next = 0;?? /* probably pointless, but whatever. */

??????? *before = nxt;

??????? return;

??? }

??? /* Note:? we never actually get here.? the callers don't delete things

?????? they can't find. */

??? assert(*before != 0);

}

HOT_LRU

在HOT LRU队列中的数据绝不会到HOT_LRU队列的前面，只会往更冷的队列中放。规则是：如果元素变得活跃，就放到WARM队列中；否则如果不活跃，就直接放到COLD队列中

WARM_LRU

如果WARM队列的元素变的活跃，就会移动到WARM队列头；否则往COLD队列移动

COLD_LRU

COLD队列中的元素，都是不太活跃的了，所以当需要淘汰元素时都会首先到COLD LRU队列中找可以淘汰的数据。当一个在COLD队列的元素重新变成活跃元素时，并不会移动到COLD队列的头部，而是直接移动回去WARM队列

PS：任何操作都不能将一个元素从WARM和COLD队列中移动回去HOT队列了，也就是从HOT队列中移动元素出去的操作是单向操作

惰性删除

memcached一般不会主动去清除已经过期或者失效的缓存，当get请求一个item才会去检查item是否失效

flush命令

flush命令会将所有的item设置为失效

创建的时候检查

Memcached会在创建ITEM的时候去LRU的链表尾部开始检查，是否有失效的ITEM，如果没有的话就重新创建

LRU爬虫

memcached默认是关闭LRU爬虫的。LRU爬虫是一个单独的线程，会去清理失效的ITEM

LRU淘汰

当缓存没有内存可以分配给新的元素的时候，memcached会从LRU链表的尾部开始淘汰一个ITEM，不管这个ITEM是否还在有效期都将会面临淘汰。LRU链表插入缓存ITEM的时候有先后顺序，所以淘汰一个ITEM也是从尾部进行 也就是先淘汰最早的ITEM。

int item_is_flushed(item *it) {

??? rel_time_t oldest_live = settings.oldest_live;

??? uint64_t cas = ITEM_get_cas(it);

??? uint64_t oldest_cas = settings.oldest_cas;

??? if (oldest_live == 0 || oldest_live > current_time)

??????? return 0;

??? if ((it->time <= oldest_live)

??????????? || (oldest_cas != 0 && cas != 0 && cas < oldest_cas)) {

??????? return 1;

??? }

??? return 0;

}

序号

描述

1

do_item_alloc进入新增item的内存申请流程

2

do_item_alloc_pull进入item申请的逻辑处理，最多处理10次

3

do_item_alloc_pull内部逻辑是尝试通过slabs_alloc申请内存，失败则尝试通过lru_pull_tail方法释放LRU队列中的item变成可用item

4

lru_pull_tail执行释放LRU队列中item的过程，内部包括各种过期item的回收