渐进式解析 memcached 源码 - 增删改查

阅读量

作为 缓存，永远都脱离不了 增删改查 这一基本的 API，而这点儿也正是好多人刚开始看源码最感兴趣的部分，此次我们就 消息解析、增删改查、消息响应 等方面进行分析

本章承接 状态机 conn_read 逻辑（不熟悉的可以先查看 http://www.web-lovers.com/memcached-source-threads.html 状态机 部分），功能型的代码逻辑比较清晰，我们此次跟随着代码逻辑进行解析

另外由于有些方法里的代码量比较大，我们这里按照 典型的代码片段进行解析，同志们可以根据文章提示的代码位置 和 代码里面的关键词 在源码中搜素，可能数据结构一些元素 看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

常用数据结构

这次我们再次拉来 conn 连接信息的数据结构，之前在 线程机制 中也有过介绍，存储某个 socket 连接的 所有连接信息数据，特别是其中的 rbuf、rsize、rcurr、rbytes、iov*、msg* 等部分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
/* memcached.h: line 554*/
struct conn {
    // 该conn对应的 socket fd
    int    sfd;
    sasl_conn_t *sasl_conn;
    bool authenticated;
    // 当前状态机状态
    enum conn_states  state;
    enum bin_substates substate;
    // 最后一次处理客户端命令key value的时间
    rel_time_t last_cmd_time;
    // 该conn对应的event
    struct event event;
    // event当前监听的事件类型
    short  ev_flags;
    // 字面意思：触发event回调函数的原因，没看到具体哪里使用，不太明白...
    short  which;   /** which events were just triggered */
    // 读缓冲区，用于存储客户端数据报文中的命令
    char   *rbuf;   /** buffer to read commands into */
    // 有效数据的开始位置，从rbuf到rcurr之间的数据是已经处理的了，变成无效数据了
    char   *rcurr;  /** but if we parsed some already, this is where we stopped */
    // rbuf的大小
    int    rsize;   /** total allocated size of rbuf */
    // 有效数据的长度
    int    rbytes;  /** how much data, starting from rcur, do we have unparsed */
    char   *wbuf;
    char   *wcurr;
    int    wsize;
    int    wbytes;
    // 写完后的下一个状态
    /** which state to go into after finishing current write */
    enum conn_states  write_and_go;
    void   *write_and_free; /** free this memory after finishing writing */
    // 数据直通车 数据部分赋值过程
    char   *ritem;  /** when we read in an item's value, it goes here */
    // 最开始rlbytes为数据部分总长度，当把数据部分填充好后，其值就刚好为0
    int    rlbytes;
    /* data for the nread state */
    /**
     * item is used to hold an item structure created after reading the command
     * line of set/add/replace commands, but before we finished reading the actual
     * data. The data is read into ITEM_data(item) to avoid extra copying.
     */
    // 该链接当前所处理命令的item结构信息
    void   *item;     /* for commands set/add/replace  */
    
    /* data for the swallow state */
    int    sbytes;    /* how many bytes to swallow */
    /* data for the mwrite state */
    // iovec数组指针
    struct iovec *iov;
    // 数组大小
    int    iovsize;   /* number of elements allocated in iov[] */
    // 已经使用的iov数组元素个数
    int    iovused;   /* number of elements used in iov[] */
    // msghdr结构体里面的iovec结构体数组长度是有限制的，为了能传输更多的数据，只能增加msghdr结构体的个数add_msghdr函数负责增加
    struct msghdr *msglist;
    // 数组大小  默认MSG_LIST_INITIAL
    int    msgsize;   /* number of elements allocated in msglist[] */
    // 已经使用了的msghdr元素个数
    int    msgused;   /* number of elements used in msglist[] */
    // 正在用sendmsg函数传输msghdr数组中的哪一个元素
    int    msgcurr;   /* element in msglist[] being transmitted now */
    // msgcurr指向的msghdr总共多少个字节
    int    msgbytes;  /* number of bytes in current msg */
    // worker线程需要占有这个item，直至把item的数据都写回给客户端了故需要一个item指针数组记录本conn占有的item
    item   **ilist;   /* list of items to write out */
    // 数组的大小
    int    isize;
    // 当前使用到的item(在释放占用item时会用到)
    item   **icurr;
    // ilist数组中有多少个item需要释放
    int    ileft;
    char   **suffixlist;
    int    suffixsize;
    char   **suffixcurr;
    int    suffixleft;
#ifdef EXTSTORE
    int io_wrapleft;
    unsigned int recache_counter;
    io_wrap *io_wraplist; /* linked list of io_wraps */
    bool io_queued; /* FIXME: debugging flag */
#endif
    // 根据接收的第一个字符来确定是那种协议，文本ascii_prot 和 二进制binary_prot
    enum protocol protocol;   /* which protocol this connection speaks */
    enum network_transport transport; /* what transport is used by this connection */
    /* data for UDP clients */
    int    request_id; /* Incoming UDP request ID, if this is a UDP "connection" */
    // 客户端IP地址
    struct sockaddr_in6 request_addr; /* udp: Who sent the most recent request */
    socklen_t request_addr_size;
    unsigned char *hdrbuf; /* udp packet headers */
    int    hdrsize;   /* number of headers' worth of space is allocated */
    // 是否不用回复客户端信息
    bool   noreply;   /* True if the reply should not be sent. */
    /* current stats command */
    struct {
        char *buffer;
        size_t size;
        size_t offset;
    } stats;
    /* Binary protocol stuff */
    /* This is where the binary header goes */
    protocol_binary_request_header binary_header;
    uint64_t cas; /* the cas to return */
    short cmd; /* current command being processed */
    int opaque;
    int keylen;
    // 下一个conn的指针
    conn   *next;     /* Used for generating a list of conn structures */
    // 所属worker线程指针
    LIBEVENT_THREAD *thread; /* Pointer to the thread object serving this connection */
};

另外就是 sendmsg 使用到的 msghdr、iovc 结构体，这个可以详见 sendmsg 的使用方法：Linux msghdr结构讲解 和 UNIX网络编程读书笔记：recvmsg和sendmsg函数

代码逻辑解析

接之前 线程机制 部分状态机逻辑，接收消息以后的逻辑依次为：消息解析、增删改查、消息响应

消息解析

状态机 drive_machine 状态转移进入 conn_read 状态，调用 try_read_network 逻辑进行命令的报文读取

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
/* memcached.c: line 5436 */
// 进入命令读取逻辑的case
case conn_read:
    // 根据不同协议进行报文读取
    res = IS_UDP(c->transport) ? try_read_udp(c) : try_read_network(c);
    
    switch (res) {
    // 没有读取到数据，继续等待
    case READ_NO_DATA_RECEIVED:
        conn_set_state(c, conn_waiting);
        break;
    // 读取到了数据，进入 conn_parse_cmd 状态
    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;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
/* memcached.c: line 5080 */
static enum try_read_result try_read_network(conn *c) {
    enum try_read_result gotdata = READ_NO_DATA_RECEIVED;
    int res;
    int num_allocs = 0;
    assert(c != NULL);
    // 把已经消息指针地址移动到首部，之前的消息已经无效，后面读取到的消息直接进行覆盖
    if (c->rcurr != c->rbuf) {
        if (c->rbytes != 0) /* otherwise there's nothing to copy */
            memmove(c->rbuf, c->rcurr, c->rbytes);
        c->rcurr = c->rbuf;
    }
    while (1) {
        // 如果 buff 空间不够，进行内存扩展
        if (c->rbytes >= c->rsize) {
            // 一共只能有 4次 buff 的内容扩容机会，防止恶意一直发送报文消息刷内存
            if (num_allocs == 4) {
                return gotdata;
            }
            ++num_allocs;
            // buff内存扩展一倍
            char *new_rbuf = realloc(c->rbuf, c->rsize * 2);
            // 内存扩展失败
            if (!new_rbuf) {
                // 统计加锁
                STATS_LOCK();
                stats.malloc_fails++;
                STATS_UNLOCK();
                if (settings.verbose > 0) {
                    fprintf(stderr, "Couldn't realloc input buffer\n");
                }
                c->rbytes = 0; /* ignore what we read */
                out_of_memory(c, "SERVER_ERROR out of memory reading request");
                c->write_and_go = conn_closing;
                return READ_MEMORY_ERROR;
            }
            // buff 指向新的内存空间
            c->rcurr = c->rbuf = new_rbuf;
            c->rsize *= 2;
        }
        // 有效空间是已经 总长度 - 已经占用的有效数据长度
        int avail = c->rsize - c->rbytes;
        // 从 socket 读取报文消息
        res = read(c->sfd, c->rbuf + c->rbytes, avail);
        if (res > 0) {
            // 线程加锁，这里用来统计线程的读取的中bytes
            pthread_mutex_lock(&c->thread->stats.mutex);
            c->thread->stats.bytes_read += res;
            pthread_mutex_unlock(&c->thread->stats.mutex);
            gotdata = READ_DATA_RECEIVED;
            c->rbytes += res;
            if (res == avail) {
                // 如果读取的报文已经占满有效区域则再次循环
                continue;
            } else {
                // 读取长度不等于有效区域，证明读取完毕
                break;
            }
        }
        if (res == 0) {
            return READ_ERROR;
        }
        if (res == -1) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                break;
            }
            return READ_ERROR;
        }
    }
    return gotdata;
}

报文数据读取完毕后状态转移至 conn_parse_cmd，调用 try_read_command 、process_command、tokenize_command 等函数进行客户端报文数据进行解析，解析出命令行，具体逻辑步骤可以查看代码注释

try_read_command 对读取的报文进行简单的有效性验证

1
2
3
4
5
6
7
8
9
10
/* memcached.c: line 5455 */
// 进入命令解析逻辑的case
case conn_parse_cmd :
    if (try_read_command(c) == 0) {
        /* wee need more data! */
        conn_set_state(c, conn_waiting);
    }
    
    break;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
/* memcached.c: line 4898 */
static int try_read_command(conn *c) {
    assert(c != NULL);
    assert(c->rcurr <= (c->rbuf + c->rsize));
    assert(c->rbytes > 0);
    // 判断报文内容类型
    if (c->protocol == negotiating_prot || c->transport == udp_transport)  {
        if ((unsigned char)c->rbuf[0] == (unsigned char)PROTOCOL_BINARY_REQ) {
            c->protocol = binary_prot;
        } else {
            c->protocol = ascii_prot;
        }
        if (settings.verbose > 1) {
            fprintf(stderr, "%d: Client using the %s protocol\n", c->sfd,
                    prot_text(c->protocol));
        }
    }
    if (c->protocol == binary_prot) {
        // 二进制模式 或者 udp 模式解析，用的比较少，我们主要对 非二进制方式进行解析
        
        // ....省略了一部分代码
        
    } else {
        char *el, *cont;
        // 如果buff里面没有需要解析的报文，则返回0，则 状态机继续等待 新的报文到来
        if (c->rbytes == 0)
            return 0;
        // 用 \n 对报文进行分割，查找 第一个 \n 的位置，并获取其 指针地址
        el = memchr(c->rcurr, '\n', c->rbytes);
        if (!el) {
            // 如果切割失败，这里还会判断如果报文长度 > 1024，则去将命令用空格分隔成一个个的命令元素，如果元素个数 >100 或者 既不是get也不是gets命令，这判定为异常报文，则关闭连接
            if (c->rbytes > 1024) {
                /*
                 * We didn't have a '\n' in the first k. This _has_ to be a
                 * large multiget, if not we should just nuke the connection.
                 */
                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;
                }
            }
            return 0;
        }
        cont = el + 1;
        // 去除 \r \n
        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(c, c->rcurr);
        c->rbytes -= (cont - c->rcurr);
        c->rcurr = cont;
        assert(c->rcurr <= (c->rbuf + c->rsize));
    }
    return 1;
}

process_command 实际命令处理逻辑，判断具体命令（get、add、replace 等）的处理逻辑，我们这里主要对常用命令进行解析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
/* memcached.c: line 4582 */
static void process_command(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);
    /*
     * for commands set/add/replace, we build an item and read the data
     * directly into it, then continue in nread_complete().
     */
    c->msgcurr = 0;
    c->msgused = 0;
    c->iovused = 0;
    // 响应列表内存处理，下方有详细说明
    if (add_msghdr(c) != 0) {
        out_of_memory(c, "SERVER_ERROR out of memory preparing response");
        return;
    }
    // 命令分割，将命令分割成 每个元素组成的数组，比如 get foo，分割成 "get" "foo"，下方有详细说明
    ntokens = tokenize_command(command, tokens, MAX_TOKENS);
    // 判断命令头部，以进入相应方式的命令处理逻辑
    if (ntokens >= 3 &&
        ((strcmp(tokens[COMMAND_TOKEN].value, "get") == 0) ||
         (strcmp(tokens[COMMAND_TOKEN].value, "bget") == 0))) {
        process_get_command(c, tokens, ntokens, false, false);
    } else if ((ntokens == 6 || ntokens == 7) &&
               ((strcmp(tokens[COMMAND_TOKEN].value, "add") == 0 && (comm = NREAD_ADD)) ||
                (strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) ||
                (strcmp(tokens[COMMAND_TOKEN].value, "replace") == 0 && (comm = NREAD_REPLACE)) ||
                (strcmp(tokens[COMMAND_TOKEN].value, "prepend") == 0 && (comm = NREAD_PREPEND)) ||
                (strcmp(tokens[COMMAND_TOKEN].value, "append") == 0 && (comm = NREAD_APPEND)) )) {
        process_update_command(c, tokens, ntokens, comm, false);
    } else if ((ntokens == 7 || ntokens == 8) && (strcmp(tokens[COMMAND_TOKEN].value, "cas") == 0 && (comm = NREAD_CAS))) {
        // 乐观锁CAS控制
        process_update_command(c, tokens, ntokens, comm, true);
    }  else if (ntokens >= 3 && ntokens <= 5 && (strcmp(tokens[COMMAND_TOKEN].value, "delete") == 0)){
        // 删除
        process_delete_command(c, tokens, ntokens);
    } else if
    
        // ....省略了一部分代码
        
    } else {
        if (ntokens >= 2 && strncmp(tokens[ntokens - 2].value, "HTTP/", 5) == 0) {
            conn_set_state(c, conn_closing);
        } else {
            out_string(c, "ERROR");
        }
    }
    return;
}

add_msghdr 方法被调用的逻辑点还是比较多的，由于 消息响应的 msghdr 结构体中 iovec结构体数组长度比较有限，而这个方法主要就是对响应列表进行扩容以及消息位置的整理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
/* memcached.c: line 285 */
static int add_msghdr(conn *c)
{
    struct msghdr *msg;
    assert(c != NULL);
    // 扩容
    if (c->msgsize == c->msgused) {
        msg = realloc(c->msglist, c->msgsize * 2 * sizeof(struct msghdr));
        if (! msg) {
            STATS_LOCK();
            stats.malloc_fails++;
            STATS_UNLOCK();
            return -1;
        }
        c->msglist = msg;
        c->msgsize *= 2;
    }
    // 初始化 msg 位置
    msg = c->msglist + c->msgused;
    /* this wipes msg_iovlen, msg_control, msg_controllen, and
       msg_flags, the last 3 of which aren't defined on solaris: */
    memset(msg, 0, sizeof(struct msghdr));
    // 初始化 msg内容主体iov 的位置
    msg->msg_iov = &c->iov[c->iovused];
    // ....省略了一部分代码
        
    return 0;
}

tokenize_command 为命令切割函数，比如 “set foo 0 0 3”，切割成 “set”、”foo”、”0”、”0”、”3”，注意这里会返回比实际元素个数 +1 长度的数组

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
/* memcached.c: line 2940 */
static size_t tokenize_command(char *command, token_t *tokens, const size_t max_tokens) {
    char *s, *e;
    size_t ntokens = 0;
    size_t len = strlen(command);
    unsigned int i = 0;
    assert(command != NULL && tokens != NULL && max_tokens > 1);
    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;
}

实际的消息解析到这里已经完成，后续就根据 process_command 分发的具体 增删改查 逻辑进行运算了

增删改查

这里主要讲述入口处理，具体 增删改查逻辑的原子操作（do_item_get、assoc_insert、assoc_delete）见 上一章存储机制 http://www.web-lovers.com/memcached-source-storage.html 的 hashtable 源码 部分

process_get_command 查询逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
/* memcached.c: line 3769 */
static inline void process_get_command(conn *c, token_t *tokens, size_t ntokens, bool return_cas, bool should_touch) {
    char *key;
    size_t nkey;
    int i = 0;
    int si = 0;
    item *it;
    // KEY_TOKEN宏 的值为1，get操作中 0对应get(gets)  1对应key值
    token_t *key_token = &tokens[KEY_TOKEN];
    char *suffix;
    int32_t exptime_int = 0;
    rel_time_t exptime = 0;
    assert(c != NULL);
    // 这个只有 gat/gats 命令才能用到，gat命令为getAndTouch，即查询数据的同时更新过期时间
    if (should_touch) {
        // For get and touch commands, use first token as exptime
        if (!safe_strtol(tokens[1].value, &exptime_int)) {
            out_string(c, "CLIENT_ERROR invalid exptime argument");
            return;
        }
        key_token++;
        exptime = realtime(exptime_int);
    }
    do {
        while(key_token->length != 0) {
            // 获取key的值以及长度
            key = key_token->value;
            nkey = key_token->length;
            // 如果key长度大于最大长度
            if (nkey > KEY_MAX_LENGTH) {
                out_string(c, "CLIENT_ERROR bad command line format");
                while (i-- > 0) {
                    // 去除之前可能已经写入的消息列表数据
                    item_remove(*(c->ilist + i));
                    if (return_cas || !settings.inline_ascii_response) {
                        do_cache_free(c->thread->suffix_cache, *(c->suffixlist + i));
                    }
                }
                return;
            }
            // limited_get 里逻辑主要是 判断 需要should_touch则调用 item_touch，反之调用 item_get；而 item_touch 和 item_get 的相同点是最终都是使用 do_item_get 函数，不同点是前者获取后更新 exptime。do_item_get详解 见 上一章存储机制 http://www.web-lovers.com/memcached-source-storage.html 的 hashtable源码部分
            it = limited_get(key, nkey, c, exptime, should_touch);
            // 判断是否自动搜集状态信息
            if (settings.detail_enabled) {
                stats_prefix_record_get(key, nkey, NULL != it);
            }
            if (it) {
                // 超出ilist范围，则进行扩展并移动it指针地址
                if (_ascii_get_expand_ilist(c, i) != 0) {
                    item_remove(it);
                    break; // FIXME: Should bail down to error.
                }
                /*
                 * Construct the response. Each hit adds three elements to the
                 * outgoing data list:
                 *   "VALUE "
                 *   key
                 *   " " + flags + " " + data length + "\r\n" + data (with \r\n)
                 */
                if (return_cas || !settings.inline_ascii_response)
                {
                  MEMCACHED_COMMAND_GET(c->sfd, ITEM_key(it), it->nkey,
                                        it->nbytes, ITEM_get_cas(it));
                  int nbytes;
                  // 获取后缀，比如set的时候是"set foo 0 0 10"，这里就是获取这个"0 10\r\n" 
                  suffix = _ascii_get_suffix_buf(c, si);
                  if (suffix == NULL) {
                      item_remove(it);
                      break;
                  }
                  si++;
                  nbytes = it->nbytes;
                  int suffix_len = make_ascii_get_suffix(suffix, it, return_cas, nbytes);
                  // 添加消息，如果添加成功则返回 0，比如 命令为 get foo，则逐次添加需要返回的消息为：
                  // VALUE foo 0 10
                  // 1234567890
                  // END
                  if (add_iov(c, "VALUE ", 6) != 0 ||
                      add_iov(c, ITEM_key(it), it->nkey) != 0 ||
                      (settings.inline_ascii_response && add_iov(c, ITEM_suffix(it), it->nsuffix - 2) != 0) ||
                      add_iov(c, suffix, suffix_len) != 0)
                      {
                          item_remove(it);
                          break;
                      }
#ifdef EXTSTORE
                  if (it->it_flags & ITEM_HDR) {
                      if (_get_extstore(c, it, c->iovused-3, 4) != 0) {
                          item_remove(it);
                          break;
                      }
                  } else if ((it->it_flags & ITEM_CHUNKED) == 0) {
#else
                  // key对应内容添加至消息主体
                  if ((it->it_flags & ITEM_CHUNKED) == 0) {
#endif
                      add_iov(c, ITEM_data(it), it->nbytes);
                  } else if (add_chunked_item_iovs(c, it, it->nbytes) != 0) {
                      item_remove(it);
                      break;
                  }
                }
                else
                {
                  MEMCACHED_COMMAND_GET(c->sfd, ITEM_key(it), it->nkey,
                                        it->nbytes, ITEM_get_cas(it));
                  if (add_iov(c, "VALUE ", 6) != 0 ||
                      add_iov(c, ITEM_key(it), it->nkey) != 0)
                      {
                          item_remove(it);
                          break;
                      }
                  if ((it->it_flags & ITEM_CHUNKED) == 0)
                      {
                          if (add_iov(c, ITEM_suffix(it), it->nsuffix + it->nbytes) != 0)
                          {
                              item_remove(it);
                              break;
                          }
                      } else if (add_iov(c, ITEM_suffix(it), it->nsuffix) != 0 ||
                                 add_chunked_item_iovs(c, it, it->nbytes) != 0) {
                          item_remove(it);
                          break;
                      }
                }
                if (settings.verbose > 1) {
                    int ii;
                    fprintf(stderr, ">%d sending key ", c->sfd);
                    for (ii = 0; ii < it->nkey; ++ii) {
                        fprintf(stderr, "%c", key[ii]);
                    }
                    fprintf(stderr, "\n");
                }
                /* item_get() has incremented it->refcount for us */
                // 数据统计
                pthread_mutex_lock(&c->thread->stats.mutex);
                if (should_touch) {
                    c->thread->stats.touch_cmds++;
                    c->thread->stats.slab_stats[ITEM_clsid(it)].touch_hits++;
                } else {
                    c->thread->stats.lru_hits[it->slabs_clsid]++;
                    c->thread->stats.get_cmds++;
                }
                pthread_mutex_unlock(&c->thread->stats.mutex);
#ifdef EXTSTORE
                /* If ITEM_HDR, an io_wrap owns the reference. */
                if ((it->it_flags & ITEM_HDR) == 0) {
                    *(c->ilist + i) = it;
                    i++;
                }
#else
                // 结果写入 ilist 消息列表中
                *(c->ilist + i) = it;
                i++;
#endif
            } else {
                // 查找不存在，进行miss等相关统计
                pthread_mutex_lock(&c->thread->stats.mutex);
                if (should_touch) {
                    c->thread->stats.touch_cmds++;
                    c->thread->stats.touch_misses++;
                } else {
                    c->thread->stats.get_misses++;
                    c->thread->stats.get_cmds++;
                }
                MEMCACHED_COMMAND_GET(c->sfd, key, nkey, -1, 0);
                pthread_mutex_unlock(&c->thread->stats.mutex);
            }
            key_token++;
        }
        /*
         * If the command string hasn't been fully processed, get the next set
         * of tokens.
         */
        // 如果 get 的 key 比较多超过了 MAX_TOKENS，这里还会继续进行切分
        if(key_token->value != NULL) {
            ntokens = tokenize_command(key_token->value, tokens, MAX_TOKENS);
            key_token = tokens;
        }
    } while(key_token->value != NULL);
    // 当前使用到item的列表 指向 消息列表
    c->icurr = c->ilist;
    // 有多少个 item 需要释放
    c->ileft = i;
    if (return_cas || !settings.inline_ascii_response) {
        c->suffixcurr = c->suffixlist;
        c->suffixleft = si;
    }
    if (settings.verbose > 1)
        fprintf(stderr, ">%d END\n", c->sfd);
    /*
        If the loop was terminated because of out-of-memory, it is not
        reliable to add END\r\n to the buffer, because it might not end
        in \r\n. So we send SERVER_ERROR instead.
    */
    // 添加消息 END\r\n
    if (key_token->value != NULL || add_iov(c, "END\r\n", 5) != 0
        || (IS_UDP(c->transport) && build_udp_headers(c) != 0)) {
        out_of_memory(c, "SERVER_ERROR out of memory writing get response");
    }
    else {
        // 进入状态机 conn_mwrite 状态进行消息响应
        conn_set_state(c, conn_mwrite);
        c->msgcurr = 0;
    }
}

process_update_command 写入逻辑，无论 set、add、append、replace、cas 用的都是 此逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
/* memcached.c: line 3968 */
static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas) {
    // key值
    char *key;
    // key值长度
    size_t nkey;
    unsigned int flags;
    int32_t exptime_int = 0;
    time_t exptime;
    int vlen;
    uint64_t req_cas_id=0;
    item *it;
    assert(c != NULL);
    // 服务端不需要返回客户端消息，减少网络IO，增加效率
    set_noreply_maybe(c, tokens, ntokens);
    // key太大
    if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) {
        out_string(c, "CLIENT_ERROR bad command line format");
        return;
    }
    // key值
    key = tokens[KEY_TOKEN].value;
    // key值长度
    nkey = tokens[KEY_TOKEN].length;
    // 将命令的每个元素进行转换，分别获取到 flags、exptime_int、vlen 三项值
    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;
    }
    /* Ubuntu 8.04 breaks when I pass exptime to safe_strtol */
    exptime = exptime_int;
    /* Negative exptimes can underflow and end up immortal. realtime() will
       immediately expire values that are greater than REALTIME_MAXDELTA, but less
       than process_started, so lets aim for that. */
    // 如果 exptime 传递的小于0，则直接失效了，这个设置一下默认失效时间 30天 + 1
    if (exptime < 0)
        exptime = REALTIME_MAXDELTA + 1;
    // does cas value exist?
    // 是否解析 cas 命令
    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;
    }
    // 长度 + 2，是因为数据后面要加上 \r\n
    vlen += 2;
    // 详情统计
    if (settings.detail_enabled) {
        stats_prefix_record_set(key, nkey);
    }
    // 申请 item，需要分配 大小相当的 空闲 item，最终调用的是 do_slabs_alloc，do_slabs_alloc详解 见 上一章存储机制 http://www.web-lovers.com/memcached-source-storage.html 的 slabs源码部分
    it = item_alloc(key, nkey, flags, realtime(exptime), vlen);
    if (it == 0) {
        // 获取 item 失败，一般是因为内存不足
        enum store_item_type status;
        if (! item_size_ok(nkey, flags, vlen)) {
            // 单数据过大
            out_string(c, "SERVER_ERROR object too large for cache");
            status = TOO_LARGE;
        } else {
            // 没有内存了
            out_of_memory(c, "SERVER_ERROR out of memory storing object");
            status = NO_MEMORY;
        }
        LOGGER_LOG(c->thread->l, LOG_MUTATIONS, LOGGER_ITEM_STORE,
                NULL, status, comm, key, nkey, 0, 0);
        /* swallow the data line */
        c->write_and_go = conn_swallow;
        c->sbytes = vlen;
        /* Avoid stale data persisting in cache because we failed alloc.
         * Unacceptable for SET. Anywhere else too? */
        // 如果 命令是 set，而且 查找 item 不成功，则删除 原来存储的 item
        if (comm == NREAD_SET) {
            it = item_get(key, nkey, c, DONT_UPDATE);
            if (it) {
                item_unlink(it);
                STORAGE_delete(c->thread->storage, it);
                item_remove(it);
            }
        }
        return;
    }
    // 设置 cas
    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);
}

conn_nread 状态继续读取后续操作内容报文

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
/* memcached.c: line 5492 */
case conn_nread:
    // 数据读取完毕，这里是循环判断的，循环读取，直到 读取完毕
    if (c->rlbytes == 0) {
        // 这里执行写入 item 操作，complete_nread->complete_nread_ascii(ascii文本读取)->store_item(根据key值获取hv数值，用于查找key所在位置)->do_store_item(具体存储逻辑)
        // complete_nread 在 store 操作执行成功后会根据结果进行调用 out_string 进行内容输出，比如 STORED、EXISTS、NOT_FOUND、NOT_STORED 以及一些错误信息 等
        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 || (((item *)c->item)->it_flags & ITEM_CHUNKED) == 0) {
        /* first check if we have leftovers in the conn_read buffer */
        // 检查一下之前 buff 里面有剩余数据，则 将 读取新数据的指针往后移动
        if (c->rbytes > 0) {
            int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes;
            if (c->ritem != c->rcurr) {
                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 = read(c->sfd, 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 */
        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;
        c->write_and_go = conn_swallow;
        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=%lx ritem=%lx rbuf=%lx rlbytes=%d rsize=%d\n",
                errno, strerror(errno),
                (long)c->rcurr, (long)c->ritem, (long)c->rbuf,
                (int)c->rlbytes, (int)c->rsize);
    }
    conn_set_state(c, conn_closing);
    break;

do_store_item 最终最主要的存储逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
/* memcached.c: line 2785 */
enum store_item_type do_store_item(item *it, int comm, conn *c, const uint32_t hv) {
    char *key = ITEM_key(it);
    // 查询已经存在的key对应的item信息
    item *old_it = do_item_get(key, it->nkey, hv, c, DONT_UPDATE);
    enum store_item_type stored = NOT_STORED;
    item *new_it = NULL;
    uint32_t flags;
    if (old_it != NULL && comm == NREAD_ADD) {
        // 如果原本key对应的item已经存在，即使 comm 是添加命令，则 不会添加和替换，而是 更新老数据的访问时间以及lru位置
        /* add only adds a nonexistent item, but promote to head of LRU */
        do_item_update(old_it);
    } else if (!old_it && (comm == NREAD_REPLACE
        || comm == NREAD_APPEND || comm == NREAD_PREPEND))
    {
        // 如果本身item不存在，而命令去 replace、append 就有点儿扯淡了，啥也不干
        /* replace only replaces an existing value; don't store */
    } else if (comm == NREAD_CAS) {
        // cas 逻辑
        /* validate cas operation */
        // 需要保证 item 已经存在
        if(old_it == NULL) {
            // LRU expired
            stored = NOT_FOUND;
            pthread_mutex_lock(&c->thread->stats.mutex);
            c->thread->stats.cas_misses++;
            pthread_mutex_unlock(&c->thread->stats.mutex);
        }
        else if (ITEM_get_cas(it) == ITEM_get_cas(old_it)) {
            // 验证 cas 是有效的
            // cas validates
            // it and old_it may belong to different classes.
            // I'm updating the stats for the one that's getting pushed out
            pthread_mutex_lock(&c->thread->stats.mutex);
            c->thread->stats.slab_stats[ITEM_clsid(old_it)].cas_hits++;
            pthread_mutex_unlock(&c->thread->stats.mutex);
            // 更新老数据
            STORAGE_delete(c->thread->storage, old_it);
            item_replace(old_it, it, hv);
            stored = STORED;
        } else {
            // cas版本不一致，仅仅进行统计，不做实质性的操作
            pthread_mutex_lock(&c->thread->stats.mutex);
            c->thread->stats.slab_stats[ITEM_clsid(old_it)].cas_badval++;
            pthread_mutex_unlock(&c->thread->stats.mutex);
            if(settings.verbose > 1) {
                fprintf(stderr, "CAS:  failure: expected %llu, got %llu\n",
                        (unsigned long long)ITEM_get_cas(old_it),
                        (unsigned long long)ITEM_get_cas(it));
            }
            stored = EXISTS;
        }
    } else {
        int failed_alloc = 0;
        /*
         * Append - combine new and old record into single one. Here it's
         * atomic and thread-safe.
         */
        if (comm == NREAD_APPEND || comm == NREAD_PREPEND) {
            /*
             * Validate CAS
             */
            // 如果有 cas 信息则 验证cas
            if (ITEM_get_cas(it) != 0) {
                // CAS much be equal
                if (ITEM_get_cas(it) != ITEM_get_cas(old_it)) {
                    stored = EXISTS;
                }
            }
#ifdef EXTSTORE
            if ((old_it->it_flags & ITEM_HDR) != 0) {
                /* block append/prepend from working with extstore-d items.
                 * also don't replace the header with the append chunk
                 * accidentally, so mark as a failed_alloc.
                 */
                failed_alloc = 1;
            } else
#endif
            // 如果中间没有出现异常，stored 到目前为止应该是 NOT_STORED 状态
            if (stored == NOT_STORED) {
                /* we have it and old_it here - alloc memory to hold both */
                /* flags was already lost - so recover them from ITEM_suffix(it) */
                if (settings.inline_ascii_response) {
                    flags = (uint32_t) strtoul(ITEM_suffix(old_it), (char **) NULL, 10);
                } else if (old_it->nsuffix > 0) {
                    flags = *((uint32_t *)ITEM_suffix(old_it));
                } else {
                    flags = 0;
                }
                // 申请新的item，因为最佳的话空间不够了
                new_it = do_item_alloc(key, it->nkey, flags, old_it->exptime, it->nbytes + old_it->nbytes - 2 /* CRLF */);
                /* copy data from it and old_it to new_it */
                // copy 新的item至老的item地址，这里其实不只是 copy 功能，而是 老item和需要保存的item 根据追加情况 链表组装好 再copy 到新的 item地址
                if (new_it == NULL || _store_item_copy_data(comm, old_it, new_it, it) == -1) {
                    failed_alloc = 1;
                    stored = NOT_STORED;
                    // failed data copy, free up.
                    // copy 失败后，要将新申请的item释放掉，还给 freelist
                    if (new_it != NULL)
                        item_remove(new_it);
                } else {
                    // copy 成功
                    it = new_it;
                }
            }
        }
        // 如果中间没有出现异常，stored 到目前为止应该是 NOT_STORED 状态；而且 没有出现扩展失败的情况
        if (stored == NOT_STORED && failed_alloc == 0) {
            // 如果原本key对应item存在，这直接替换掉，即使是 append，这里的it 也是追加好的item，详见前几行的逻辑
            if (old_it != NULL) {
                STORAGE_delete(c->thread->storage, old_it);
                item_replace(old_it, it, hv);
            } else {
                // 如果原本key对应item不存在，直接将 it 链接到对应 hashtable 下面
                do_item_link(it, hv);
            }
            c->cas = ITEM_get_cas(it);
            stored = STORED;
        }
    }
    // 下面的逻辑就是开始释放用过的 item
    if (old_it != NULL)
        do_item_remove(old_it);         /* release our reference */
    if (new_it != NULL)
        do_item_remove(new_it);
    if (stored == STORED) {
        c->cas = ITEM_get_cas(it);
    }
    LOGGER_LOG(c->thread->l, LOG_MUTATIONS, LOGGER_ITEM_STORE, NULL,
            stored, comm, ITEM_key(it), it->nkey, it->exptime, ITEM_clsid(it));
    return stored;
}

process_delete_command 删除逻辑，不过相对来说删除逻辑还是比较简单的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
/* memcached.c: line 4279 */
static void process_delete_command(conn *c, token_t *tokens, const size_t ntokens) {
    char *key;
    size_t nkey;
    item *it;
    assert(c != NULL);
    // 命令参数合法性
    if (ntokens > 3) {
        bool hold_is_zero = strcmp(tokens[KEY_TOKEN+1].value, "0") == 0;
        bool sets_noreply = set_noreply_maybe(c, tokens, ntokens);
        bool valid = (ntokens == 4 && (hold_is_zero || sets_noreply))
            || (ntokens == 5 && hold_is_zero && sets_noreply);
        if (!valid) {
            out_string(c, "CLIENT_ERROR bad command line format.  "
                       "Usage: delete <key> [noreply]");
            return;
        }
    }
    // 获取key值和key值长度
    key = tokens[KEY_TOKEN].value;
    nkey = tokens[KEY_TOKEN].length;
    if(nkey > KEY_MAX_LENGTH) {
        out_string(c, "CLIENT_ERROR bad command line format");
        return;
    }
    if (settings.detail_enabled) {
        stats_prefix_record_delete(key, nkey);
    }
    // 查找key值对应的item
    it = item_get(key, nkey, c, DONT_UPDATE);
    if (it) {
        MEMCACHED_COMMAND_DELETE(c->sfd, ITEM_key(it), it->nkey);
        // 删除命中统计
        pthread_mutex_lock(&c->thread->stats.mutex);
        c->thread->stats.slab_stats[ITEM_clsid(it)].delete_hits++;
        pthread_mutex_unlock(&c->thread->stats.mutex);
        // 将item从hashtable中解下来
        item_unlink(it);
        // 这行的操作只有在 EXTSTORE 宏定义的时候才会生效，即只有 外部存储设置的时候
        STORAGE_delete(c->thread->storage, it);
        // 移除空闲的item，收回至 freelist
        item_remove(it);      /* release our reference */
        out_string(c, "DELETED");
    } else {
        // 找不到需要删除的数据，miss统计
        pthread_mutex_lock(&c->thread->stats.mutex);
        c->thread->stats.delete_misses++;
        pthread_mutex_unlock(&c->thread->stats.mutex);
        out_string(c, "NOT_FOUND");
    }
}

消息响应

add_iov 添加消息内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
/* memcached.c: line 957 */
static int add_iov(conn *c, const void *buf, int len) {
    struct msghdr *m;
    int leftover;
    assert(c != NULL);
    // UDP
    if (IS_UDP(c->transport)) {
        
        // ....省略了一部分代码
        
    } else {
        /* Optimized path for TCP connections */
        // 消息列表msglist存储消息，这里获取重新可以写入的指针
        m = &c->msglist[c->msgused - 1];
        // 如果返回的结果列表已经大于最大值，则重新更新 add_msghdr，add_msghdr方法就是对 msghdr 进行扩容
        if (m->msg_iovlen == IOV_MAX) {
            add_msghdr(c);
            m = &c->msglist[c->msgused - 1];
        }
        // ensure_iov_space 是检测 iov 空间是否足够，如果不够则 扩容两倍
        if (ensure_iov_space(c) != 0)
            return -1;
        // 添加数据至iov
        m->msg_iov[m->msg_iovlen].iov_base = (void *)buf;
        m->msg_iov[m->msg_iovlen].iov_len = len;
        c->msgbytes += len;
        c->iovused++;
        m->msg_iovlen++;
    }
    return 0;
}

add_msghdr msghdr 扩容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
/* memcached.c: line 285 */
static int add_msghdr(conn *c)
{
    struct msghdr *msg;
    assert(c != NULL);
    // 空间是否足够
    if (c->msgsize == c->msgused) {
        // 扩容一倍
        msg = realloc(c->msglist, c->msgsize * 2 * sizeof(struct msghdr));
        if (! msg) {
            STATS_LOCK();
            stats.malloc_fails++;
            STATS_UNLOCK();
            return -1;
        }
        c->msglist = msg;
        c->msgsize *= 2;
    }
    msg = c->msglist + c->msgused;
    /* this wipes msg_iovlen, msg_control, msg_controllen, and
       msg_flags, the last 3 of which aren't defined on solaris: */
    memset(msg, 0, sizeof(struct msghdr));
    // 重新指定最新指针位置
    msg->msg_iov = &c->iov[c->iovused];
    if (IS_UDP(c->transport) && c->request_addr_size > 0) {
        msg->msg_name = &c->request_addr;
        msg->msg_namelen = c->request_addr_size;
    }
    c->msgbytes = 0;
    c->msgused++;
    if (IS_UDP(c->transport)) {
        /* Leave room for the UDP header, which we'll fill in later. */
        return add_iov(c, NULL, UDP_HEADER_SIZE);
    }
    return 0;
}

进入状态机 conn_mwrite 状态进行消息响应

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
/* memcached.c: line 5639 */
// 进入消息响应逻辑的case
case conn_mwrite:
    // ....省略了一部分代码
 
    // 调用 transmit 函数对 客户端写入 响应消息
    switch (transmit(c)) {
      // 传输完成
      case TRANSMIT_COMPLETE:
          if (c->state == conn_mwrite) {
              // 释放 item 的占用
              conn_release_items(c);
              /* XXX:  I don't know why this wasn't the general case */
              if(c->protocol == binary_prot) {
                  conn_set_state(c, c->write_and_go);
              } else {
                  // 重新进入等待命令阶段
                  conn_set_state(c, conn_new_cmd);
              }
          } else if (c->state == conn_write) {
              if (c->write_and_free) {
                  free(c->write_and_free);
                  c->write_and_free = 0;
              }
              conn_set_state(c, c->write_and_go);
          } else {
              if (settings.verbose > 0)
                  fprintf(stderr, "Unexpected state %d\n", c->state);
              conn_set_state(c, conn_closing);
          }
          break;
      // 传输一部分
      case TRANSMIT_INCOMPLETE:
      // 有可能没有传输，send返回的 ret 为0，继续循环 conn_mwrite 状态
      case TRANSMIT_HARD_ERROR:
          break;                   /* Continue in state machine. */
      // 传输错误
      case TRANSMIT_SOFT_ERROR:
          stop = true;
          break;
      }
      break;

transmit 函数是枚举类型，主要往 客户端写入响应数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
/* memcached.c: line 5208 */
static enum transmit_result transmit(conn *c) {
    assert(c != NULL);
    // 当前指向的 msg 已经发送完毕 并且 消息列表未发送长度是 0
    if (c->msgcurr < c->msgused &&
            c->msglist[c->msgcurr].msg_iovlen == 0) {
        /* Finished writing the current msg; advance to the next. */
        c->msgcurr++;
    }
    if (c->msgcurr < c->msgused) {
        ssize_t res;
        // 需要发送的消息结构体
        struct msghdr *m = &c->msglist[c->msgcurr];
        // 写入客户端sfd
        res = sendmsg(c->sfd, m, 0);
        if (res > 0) {
            // 发送成功统计
            pthread_mutex_lock(&c->thread->stats.mutex);
            c->thread->stats.bytes_written += res;
            pthread_mutex_unlock(&c->thread->stats.mutex);
            /* We've written some of the data. Remove the completed
               iovec entries from the list of pending writes. */
            // 根据返回的res写入长度，循环iovec数组，循环减去每一个数组的元素长度
            while (m->msg_iovlen > 0 && res >= m->msg_iov->iov_len) {
                res -= m->msg_iov->iov_len;
                m->msg_iovlen--;
                m->msg_iov++;
            }
            /* Might have written just part of the last iovec entry;
               adjust it so the next write will do the rest. */
            // 写入了部分数据
            if (res > 0) {
                m->msg_iov->iov_base = (caddr_t)m->msg_iov->iov_base + res;
                m->msg_iov->iov_len -= res;
            }
            return TRANSMIT_INCOMPLETE;
        }
        // 写入失败
        if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
            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);
                return TRANSMIT_HARD_ERROR;
            }
            return TRANSMIT_SOFT_ERROR;
        }
        /* if res == 0 or res == -1 and error is not EAGAIN or EWOULDBLOCK,
           we have a real error, on which we close the connection */
        if (settings.verbose > 0)
            perror("Failed to write, and not due to blocking");
        if (IS_UDP(c->transport))
            conn_set_state(c, conn_read);
        else
            conn_set_state(c, conn_closing);
        return TRANSMIT_HARD_ERROR;
    } else {
        return TRANSMIT_COMPLETE;
    }
}

本文作者： wettper
本文链接： http://www.web-lovers.com/memcached-source-add-del-review.html
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

wettper <br /><br /> 现 离职状态 程序猿 <br /><br /> 叫醒我的不止阳光，还有梦想！<br /><br /> 喵喵、念念、健身、吉他！