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的回收