Memcached源码分析之memcached.c

memcached.c 由于代码太多，在此省略了部分代码，例如UPD连接，二进制协议，某些错误输出和调试输出等，建议从main函数开始看起。

1. #include "memcached.h"
2. //尝试从socket中读取数据的结果枚举
3. enum try_read_result {
4. READ_DATA_RECEIVED,
5. READ_NO_DATA_RECEIVED,
6. READ_ERROR, /** an error occured (on the socket) (or client closed connection) */
7. READ_MEMORY_ERROR /** failed to allocate more memory */
8. };
9. struct stats stats; //全局统计变量
10. struct settings settings; //全局配置变量
11. static conn *listen_conn = NULL;  //全局的监听连接对象
12. static int max_fds; //fds上限
13. static struct event_base *main_base; //主线程的event_base 对象
14. //向客户端发送数据的结果枚举
15. enum transmit_result {
16. TRANSMIT_COMPLETE, /** All done writing. */
17. TRANSMIT_INCOMPLETE, /** More data remaining to write. */
18. TRANSMIT_SOFT_ERROR, /** Can‘t write any more right now. */
19. TRANSMIT_HARD_ERROR /** Can‘t write (c->state is set to conn_closing) */
20. };
21. extern pthread_mutex_t conn_lock; //连接锁
22. /**
23. 创建连接函数，参数为：
24. sfd 要监听的socket fd
25. init_state 连接的初始化状态conn_states
26. event_flags 监听的事件
27. read_buffer_size 读缓存大小
28. transport 监听的socket 类型
29. base event_base
30. 每监听一个fd（listen fd和client fd）都会创建这样一个conn来保存相关信息，表示一个“连接”，
31. 无论连接是已经连接上了，还是仅仅处于listen状态。
32. */
33. conn *conn_new(const int sfd, enum conn_states init_state,
34. const int event_flags,
35. const int read_buffer_size, enum network_transport transport,
36. struct event_base *base) {
37. conn *c;
38. assert(sfd >= 0 && sfd < max_fds);
39. c = conns[sfd];
40. if (NULL == c) {
41. if (!(c = (conn *)calloc(1, sizeof(conn)))) {
42. STATS_LOCK();
43. stats.malloc_fails++;
44. STATS_UNLOCK();
45. fprintf(stderr, "Failed to allocate connection object\n");
46. return NULL;
47. }
48. MEMCACHED_CONN_CREATE(c);
49. c->rbuf = c->wbuf = 0;
50. //c->xx省略部分初始化代码
51. c->hdrsize = 0;
52. c->rbuf = (char *)malloc((size_t)c->rsize);
53. c->wbuf = (char *)malloc((size_t)c->wsize);
54. c->ilist = (item **)malloc(sizeof(item *) * c->isize);
55. c->suffixlist = (char **)malloc(sizeof(char *) * c->suffixsize);
56. c->iov = (struct iovec *)malloc(sizeof(struct iovec) * c->iovsize);
57. c->msglist = (struct msghdr *)malloc(sizeof(struct msghdr) * c->msgsize);
58. if (c->rbuf == 0 || c->wbuf == 0 || c->ilist == 0 || c->iov == 0 ||
59. c->msglist == 0 || c->suffixlist == 0) {
60. conn_free(c);
61. STATS_LOCK();
62. stats.malloc_fails++;
63. STATS_UNLOCK();
64. fprintf(stderr, "Failed to allocate buffers for connection\n");
65. return NULL;
66. }
67. STATS_LOCK();
68. stats.conn_structs++;
69. STATS_UNLOCK();
70. c->sfd = sfd;
71. conns[sfd] = c;
72. }
73. c->transport = transport;
74. c->protocol = settings.binding_protocol;
75. if (!settings.socketpath) {
76. c->request_addr_size = sizeof(c->request_addr);
77. } else {
78. c->request_addr_size = 0;
79. }
80. if (transport == tcp_transport && init_state == conn_new_cmd) {
81. if (getpeername(sfd, (struct sockaddr *) &c->request_addr,
82. &c->request_addr_size)) {
83. perror("getpeername");
84. memset(&c->request_addr, 0, sizeof(c->request_addr));
85. }
86. }
87. if (settings.verbose > 1) {
88. //省略向终端输出调试的代码
89. }
90. c->state = init_state;
91. //c->xxx 省略部分初始化代码
92. c->noreply = false;
93. //创建事件，处理函数为event_handler，并把conn 连接对象传入event_handler中。
94. //（主线程调用conn_new时：）在主线程，创建完listenfd后，调用此函数监听网络连接事件，此时conn对象的conn_state为conn_listening
95. /**（建议看到worker线程调用conn_new时才看以下解析）
96. （worker线程调用conn_new时）在worker线程，收到主线程丢过来的client fd时，调用此函数监听来自client fd的网络事件。
97. 也就是说，无论是主线程还是worker线程，都会调用此函数conn_new，创建conn连接对象，同时监听各自的fd。
98. 而且都是调用event_handler作处理，只是不一样的fd, 不一样的conn对象（即下面的（void *） c）
99. 进入 event_handler看看都做了啥？
100. */
101. event_set(&c->event, sfd, event_flags, event_handler, (void *)c);
102. event_base_set(base, &c->event); //为事件设置事件基地
103. c->ev_flags = event_flags;
104. if (event_add(&c->event, 0) == -1) { //把事件加入监听
105. perror("event_add");
106. return NULL;
107. }
108. STATS_LOCK();
109. stats.curr_conns++;
110. stats.total_conns++;
111. STATS_UNLOCK();
112. MEMCACHED_CONN_ALLOCATE(c->sfd);
113. return c;
114. }
115. //向客户端输出字符串
116. static void out_string(conn *c, const char *str) {
117. size_t len;
118. assert(c != NULL);
119. if (c->noreply) {
120. if (settings.verbose > 1)
121. fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str);
122. c->noreply = false;
123. conn_set_state(c, conn_new_cmd);
124. return;
125. }
126. if (settings.verbose > 1)
127. fprintf(stderr, ">%d %s\n", c->sfd, str);
128. /* Nuke a partial output... */
129. c->msgcurr = 0;
130. c->msgused = 0;
131. c->iovused = 0;
132. add_msghdr(c); //添加一个msghdr
133. len = strlen(str);
134. if ((len + 2) > c->wsize) {
135. /* ought to be always enough. just fail for simplicity */
136. str = "SERVER_ERROR output line too long";
137. len = strlen(str);
138. }
139. memcpy(c->wbuf, str, len); //把要发送的字符串写入wbuf字段中
140. memcpy(c->wbuf + len, "\r\n", 2); //添加换行回车符
141. c->wbytes = len + 2;
142. c->wcurr = c->wbuf;
143. conn_set_state(c, conn_write); //把连接状态设置为conn_write，则状态机进入conn_write分支执行输出
144. c->write_and_go = conn_new_cmd; //当状态机完成输出后要切换到的状态为conn_new_cmd
145. return;
146. }
147. /**
148. 无法分配内存时经常会调用此函数，例如内存空间已满，但又无法淘汰旧的item时，则向客户端响应一个错误
149. */
150. static void out_of_memory(conn *c, char *ascii_error) {
151. const static char error_prefix[] = "SERVER_ERROR ";
152. const static int error_prefix_len = sizeof(error_prefix) - 1;
153. if (c->protocol == binary_prot) {
154. /* Strip off the generic error prefix; it‘s irrelevant in binary */
155. if (!strncmp(ascii_error, error_prefix, error_prefix_len)) {
156. ascii_error += error_prefix_len;
157. }
158. write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, ascii_error, 0);
159. } else {
160. out_string(c, ascii_error);
161. }
162. }
163. /*
164. * 解析文本协议时，执行set/add/replace 命令并把value数据塞到item之后，调用此函数。
165. * 这个函数的作用比较简单，主要是做一些收尾工作。
166. *
167. */
168. static void complete_nread_ascii(conn *c) {
169. assert(c != NULL);
170. item *it = c->item;
171. int comm = c->cmd;
172. enum store_item_type ret;
173. pthread_mutex_lock(&c->thread->stats.mutex);
174. c->thread->stats.slab_stats[it->slabs_clsid].set_cmds++;
175. pthread_mutex_unlock(&c->thread->stats.mutex);
176. if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) {
177. out_string(c, "CLIENT_ERROR bad data chunk");
178. } else {
179. ret = store_item(it, comm, c);
180. }
181. switch (ret) {
182. case STORED:
183. out_string(c, "STORED");
184. break;
185. //省略其它
186. default:
187. out_string(c, "SERVER_ERROR Unhandled storage type.");
188. }
189. }
190. //释放引用，注意仅仅是一个释放引用的逻辑，不一定把内存空间给释放掉
191. //具体是否要把内存空间释放，说见item_remove和do_item_remove函数。而在此处，
192. //一般都只是释放引用而已。
193. item_remove(c->item); /* release the c->item reference */
194. c->item = 0;
195. }
196. //重置命令处理
197. static void reset_cmd_handler(conn *c) {
198. c->cmd = -1;
199. c->substate = bin_no_state;
200. if(c->item != NULL) {
201. item_remove(c->item);
202. c->item = NULL;
203. }
204. conn_shrink(c);
205. /**
206. 不知道你有没有在下面这个地方困惑过。。反正我是纠结了很久。。。
207. 这个c->rbytes到底什么时候开始不为0了？为0的话，怎么执行这次的命令？
208. 回顾一下我们是怎么会调用到reset_cmd_handler这个地方：
209. 1）主线程监听到listen fd有连接上来，进入drive_machine的conn_listen状态，
210. 然后accept连接得到client fd，再dispatch_conn_new把client fd丢给某个worker线程
211. 2）某个worker线程收到主线程丢过来的client fd之后，把它加到那个worker线程自己的
212. event_base，然后注册event_handler（event_handler主要是调drive_machine）作为事件处理函数，
213. 在注册event_handler的同时，初始化了一个conn对象作为drive_machine的参数，
214. 而这个对象中的rbytes为0。
215. 3）当worker线程监听的client fd第一次有命令请求过来时（注意是第一次），例如set key 0 0 4\r\n，
216. worker线程的收到通知，然后陷入了event_handler再到drive_machine中去。。
217. 而此时，传过来的conn* c，c->state 为conn_new_cmd，而c->rbytes确实为0！！
218. （再次强调是第一次有命令请求过来时）
219. 就在这个时候就进入reset_cmd_handler，当前你看到的这个地方了。继续分析之后做了些啥：
220. 4）第3）点得知，c->rbytes由此确实为0，所以在这种情况，必然会进入下面的else分支，
221. 状态机进入conn_waiting状态。。。。
222. 5）进入conn_waiting状态做了啥？就是把连接状态由conn_waiting变成conn_read，然后stop = true，退出
223. 状态机。没错，退出状态机了！结束本次event_handler了！这就是比较纠结之处，这次命令请求触发的
224. event_handler居然只做了把状态由conn_new_cmd变成conn_read，然后再等待下次事件通知。
225. （详见conn_waiting分支代码）
226. 6）那刚才那个命令请求怎么办？压根没有去读？例如set key 0 0 4\r\n这个命令的实质作用被忽略了吗？
227. 其实没有。。原因是libevent默认的是水平触发，也就是说，这个命令还没被读，下次继续触发。。。
228. 下次event_base会因为刚才那个命令再触发通知告诉worker线程，再次进入drive_machine，只是此时
229. c->state是conn_read状态，conn_read分支才真正把这个命令执行！
230. 也就是说，当worker线程监听的client fd "第一次"有命令过来的时候，会触发两次event_base的通知！！
231. */
232. //这个rbytes大于0的情况，是当一次event中包含多个命令，
233. //或者说多个\r\n时候，程序执行到第二个及以后命令的时候出现。
234. if (c->rbytes > 0) {
235. conn_set_state(c, conn_parse_cmd);
236. } else {
237. conn_set_state(c, c
238. onn_waiting);
239. }
240. }
241. enum store_item_type do_store_item(item *it, int comm, conn *c, const uint32_t hv) {
242. char *key = ITEM_key(it);
243. item *old_it = do_item_get(key, it->nkey, hv); //拿出旧的item
244. enum store_item_type stored = NOT_STORED;
245. item *new_it = NULL;
246. int flags;
247. if (old_it != NULL && comm == NREAD_ADD) {
248. do_item_update(old_it); //更新item信息，其实主要是更新最近使用信息，即lru链表，。
249. } else if (!old_it && (comm == NREAD_REPLACE
250. || comm == NREAD_APPEND || comm == NREAD_PREPEND))
251. {
252. } else if (comm == NREAD_CAS) { //省略
253. } else {
254. if (comm == NREAD_APPEND || comm == NREAD_PREPEND) {
255. if (ITEM_get_cas(it) != 0) {
256. // CAS much be equal
257. if (ITEM_get_cas(it) != ITEM_get_cas(old_it)) {
258. stored = EXISTS;
259. }
260. }
261. if (stored == NOT_STORED) {
262. flags = (int) strtol(ITEM_suffix(old_it), (char **) NULL, 10);
263. new_it = do_item_alloc(key, it->nkey, flags, old_it->exptime, it->nbytes + old_it->nbytes - 2 /* CRLF */, hv);
264. if (new_it == NULL) {
265. if (old_it != NULL)
266. do_item_remove(old_it);
267. return NOT_STORED;
268. }
269. if (comm == NREAD_APPEND) {
270. memcpy(ITEM_data(new_it), ITEM_data(old_it), old_it->nbytes);
271. memcpy(ITEM_data(new_it) + old_it->nbytes - 2 /* CRLF */, ITEM_data(it), it->nbytes);
272. } else {
273. memcpy(ITEM_data(new_it), ITEM_data(it), it->nbytes);
274. memcpy(ITEM_data(new_it) + it->nbytes - 2 /* CRLF */, ITEM_data(old_it), old_it->nbytes);
275. }
276. it = new_it;
277. }
278. }
279. //SET 命令会直接跑来这里，不仔细看还真不知道 -_-||
280. /**
281. 会跑来这里的，NOT_STORED的都是还没执行"link"操作的情况。
282. 像NREAD_APPEND这些命令都是有link过的。
283. 至于什么是link，具体往下看。
284. */
285. if (stored == NOT_STORED) {
286. if (old_it != NULL)
287. /**
288. 如果要SET 的key已经存在，则用新的item覆盖新的。
289. 详见thread::item_replace和items::do_item_replace
290. */
291. item_replace(old_it, it, hv);
292. else
293. /**
294. 如果要SET 的不存在，则把item链接起来，这链接主要是做了一些关于这个item杂七
295. 杂八的工作，其实比较重要的两点：
296. 1）把item放到hash table ，作用就是为了能够很快通过key拿到item啦。
297. 2）插入到lru链表
298. 详见items::do_item_link
299. */
300. do_item_link(it, hv);
301. c->cas = ITEM_get_cas(it);
302. stored = STORED;
303. }
304. }
305. if (old_it != NULL)
306. do_item_remove(old_it); /* release our reference */
307. if (new_it != NULL)
308. do_item_remove(new_it);
309. if (stored == STORED) {
310. c->cas = ITEM_get_cas(it);
311. }
312. return stored;
313. }
314. static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas) {
315. char *key;
316. size_t nkey;
317. unsigned int flags;
318. int32_t exptime_int = 0;
319. time_t exptime;
320. int vlen;
321. uint64_t req_cas_id=0;
322. item *it;
323. assert(c != NULL);
324. set_noreply_maybe(c, tokens, ntokens);
325. if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) {
326. out_string(c, "CLIENT_ERROR bad command line format"); //key过长，out_string函数的作用是输出响应，
327. //详见out_string定义处
328. return;
329. }
330. key = tokens[KEY_TOKEN].value; //键名
331. nkey = tokens[KEY_TOKEN].length; //键长度
332. if (! (safe_strtoul(tokens[2].value, (uint32_t *)&flags)
333. && safe_strtol(tokens[3].value, &exptime_int)
334. && safe_strtol(tokens[4].value, (int32_t *)&vlen))) {
335. out_string(c, "CLIENT_ERROR bad command line format");
336. return;
337. }
338. exptime = exptime_int;
339. if (exptime < 0)
340. exptime = REALTIME_MAXDELTA + 1;
341. if (handle_cas) {
342. if (!safe_strtoull(tokens[5].value, &req_cas_id)) {
343. out_string(c, "CLIENT_ERROR bad command line format");
344. return;
345. }
346. }
347. vlen += 2;
348. if (vlen < 0 || vlen - 2 < 0) {
349. out_string(c, "CLIENT_ERROR bad command line format");
350. return;
351. }
352. if (settings.detail_enabled) {
353. stats_prefix_record_set(key, nkey);
354. }
355. it = item_alloc(key, nkey, flags, realtime(exptime), vlen); //在这里执行内存分配工作。详见item_alloc
356. if (it == 0) {
357. //略
358. return;
359. }
360. ITEM_set_cas(it, req_cas_id);
361. c->item = it; //将item指针指向分配的item空间
362. c->ritem = ITEM_data(it); //将 ritem 指向 it->data中要存放 value 的位置
363. c->rlbytes = it->nbytes; //data的大小
364. c->cmd = comm; //命令类型
365. conn_set_state(c, conn_nread); //继续调用状态机，执行命令的另一半工作。
366. }
367. /**
368. 这里就是对命令的解析和执行了，准确来说，这里只是执行了命令的一半，
369. 然后根据命令类型再次改变conn_state使程序再次进入状态机，完成命令的
370. 另一半工作
371. command此时的指针值等于conn的rcurr
372. */
373. static void process_command(conn *c, char *command) {
374. token_t tokens[MAX_TOKENS];
375. size_t ntokens;
376. int comm; //命令类型
377. assert(c != NULL);
378. MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes);
379. if (settings.verbose > 1)
380. fprintf(stderr, "<%d %s\n", c->sfd, command);
381. c->msgcurr = 0;
382. c->msgused = 0;
383. c->iovused = 0;
384. if (add_msghdr(c) != 0) {
385. out_of_memory(c, "SERVER_ERROR out of memory preparing response");
386. return;
387. }
388. /**
389. 下面这个tokenize_command是一个词法分析，把command分解成一个个token
390. */
391. ntokens = tokenize_command(command, tokens, MAX_TOKENS);
392. //下面是对上面分解出来的token再进行语法分析，解析命令，下面的comm变量为最终解析出来命令类型
393. if (ntokens >= 3 &&
394. ((strcmp(tokens[COMMAND_TOKEN].value, "get") == 0) ||
395. (strcmp(tokens[COMMAND_TOKEN].value, "bget") == 0))) {
396. process_get_command(c, tokens, ntokens, false);
397. } else if ((ntokens == 6 || ntokens == 7) &&xx//略
398. //add/set/replace/prepend/append为“更新”命令，调用同一个函数执行命令。详见process_update_command定义处
399. process_update_command(c, tokens, ntokens, comm, false);
400. } else if ((ntokens == 7 || ntokens == 8) && (strcmp(tokens[COMMAND_TOKEN].value, "cas") == 0 && (comm = NREAD_CAS))) {
401. process_update_command(c, tokens, ntokens, comm, true);
402. } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "incr") == 0)) {
403. process_arithmetic_command(c, tokens, ntokens, 1);
404. } else if (ntokens >= 3 && (strcmp(tokens[COMMAND_TOKEN].value, "gets") == 0)) {
405. //执行get 命令
406. process_get_command(c, tokens, ntokens, true);
407. } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "decr") == 0)) {
408. process_arithmetic_command(c, tokens, ntokens, 0);
409. } else if (ntokens >= 3 && ntokens <= 5 && (strcmp(tokens[COMMAND_TOKEN].value, "delete") == 0)) {
410. //执行delete 删除命令
411. process_delete_command(c, tokens, ntokens);
412. }elseif{//省略其它命令。。。
413. }
414. return;
415. }
416. /*
417. * if we have a complete line in the buffer, process it.
418. */
419. static int try_read_command(conn *c) {
420. assert(c != NULL);
421. assert(c->rcurr <= (c->rbuf + c->rsize));
422. assert(c->rbytes > 0);
423. //省略掉UDP和二进制协议的逻辑
424. if (c->protocol == binary_prot) {
425. } else {
426. char *el, *cont;
427. if (c->rbytes == 0) //读buffer没有待解析的数据
428. return 0;
429. el = memchr(c->rcurr, ‘\n‘, c->rbytes); //找第一个命令的末尾，即换行符
430. if (!el) {
431. if (c->rbytes > 1024) {
432. char *ptr = c->rcurr;
433. while (*ptr == ‘ ‘) { /* ignore leading whitespaces */
434. ++ptr;
435. }
436. if (ptr - c->rcurr > 100 ||
437. (strncmp(ptr, "get ", 4) && strncmp(ptr, "gets ", 5))) {
438. conn_set_state(c, conn_closing);
439. return 1;
440. }
441. }
442. /*
443. 如果没有找到换行符，则说明读到的数据还不足以成为一个完整的命令，
444. 返回0
445. */
446. return 0;
447. }
448. cont = el + 1; //下一个命令的开头
449. /*
450. 下面这个if的作用是把el指向当前命令最后一个有效字符的下一个字符，即\r
451. 目的是为了在命令后面插上一个\0，字符串结束符。
452. 例如 GET abc\r\n******，变成GET abc\0\n*****，这样以后读出的字符串就是一个命令。
453. */
454. if ((el - c->rcurr) > 1 && *(el - 1) == ‘\r‘) {
455. el--;
456. }
457. *el = ‘\0‘;
458. assert(cont <= (c->rcurr + c->rbytes));
459. c->last_cmd_time = current_time;
460. process_command(c, c->rcurr); //执行命令。分析详见process_command
461. //当前命令执行完之后，把当前指针rcurr指向 下一个命令的开头，并调用rbytes（剩余未处理字节数大小）
462. //逻辑上相当于把已处理的命令去掉。
463. c->rbytes -= (cont - c->rcurr);
464. c->rcurr = cont;
465. assert(c->rcurr <= (c->rbuf + c->rsize));
466. }
467. return 1;
468. }
469. static enum try_read_result try_read_network(conn *c) {
470. enum try_read_result gotdata = READ_NO_DATA_RECEIVED;
471. int res;
472. int num_allocs = 0;
473. assert(c != NULL);
474. if (c->rcurr != c->rbuf) {
475. if (c->rbytes != 0) /* otherwise there‘s nothing to copy */
476. memmove(c->rbuf, c->rcurr, c->rbytes);
477. c->rcurr = c->rbuf;
478. }
479. while (1) {
480. if (c->rbytes >= c->rsize) {//读buffer空间扩充
481. if (num_allocs == 4) {
482. return gotdata;
483. }
484. ++num_allocs;
485. char *new_rbuf = realloc(c->rbuf, c->rsize * 2);
486. if (!new_rbuf) {
487. //失败的情况，省略
488. }
489. c->rcurr = c->rbuf = new_rbuf;
490. c->rsize *= 2;
491. }
492. int avail = c->rsize - c->rbytes; //读buffer的空间还剩余多少大小可以用
493. res = read(c->sfd, c->rbuf + c->rbytes, avail); //往剩下的可用的地方里塞
494. if (res > 0) {
495. pthread_mutex_lock(&c->thread->stats.mutex);
496. c->thread->stats.bytes_read += res;
497. pthread_mutex_unlock(&c->thread->stats.mutex);
498. gotdata = READ_DATA_RECEIVED;
499. /**
500. rbytes是当前指针rcurr至读buffer末尾的数据大小，这里可简单地理解为对rbytes的初始化。
501. */
502. c->rbytes += res;
503. if (res == avail) { //可能还没读完，此时读buffer可用空间满了，那么下次循环会进行读buffer空间扩充
504. continue;
505. } else {
506. break; //socket的可读数据都读完了
507. }
508. }
509. if (res == 0) {
510. return READ_ERROR;
511. }
512. if (res == -1) {
513. if (errno == EAGAIN || errno == EWOULDBLOCK) {
514. break;
515. }
516. return READ_ERROR;
517. }
518. }
519. return gotdata;
520. }
521. /**
522. 通过状态机的方式处理网络请求/命令，后面的代码可以看到，
523. 实质这里是根据
524. 1）不同的conn连接对象，包括来自主线程的监听listen fd的连接或来自worker线程监听的client fd的连接
525. 2）或者说是conn连接不同的状态，例如主线程监听listen fd连接通常是conn_listenning状态，
526. 而worker线程监听的client fd的连接 可能是conn_new_cmd，conn_read等状态
527. 来进行不同的业务处理
528. */
529. static void drive_machine(conn *c) {
530. bool stop = false;
531. int sfd;
532. socklen_t addrlen;
533. struct sockaddr_storage addr;
534. int nreqs = settings.reqs_per_event; //每个连接可处理的最大请求数
535. int res;
536. const char *str;
537. #ifdef HAVE_ACCEPT4
538. static int use_accept4 = 1;
539. #else
540. static int use_accept4 = 0;
541. #endif
542. assert(c != NULL);
543. /**
544. 这里是事件被触发后处理业务逻辑的核心
545. 这个while循环里面有一个巨大的switch case，根据连接对象 conn当前
546. 的连接状态conn_state，进入不同的case，而每个case可能会改变conn的连接状态，
547. 也就是说在这个while+switch中，conn会不断的发生状态转移，最后被分发到合适的case上作处理。
548. 可以理解为，这里是一个有向图，每个case是一个顶点，有些case通过改变conn对象的连接状态让程序在
549. 下一次循环中进入另一个case，几次循环后程序最终进入到“无出度的顶点”然后结束状态机，
550. 这里的无出度的顶点就是带设置stop=true的case分支
551. */
552. while (!stop) {
553. switch(c->state) {
554. case conn_listening: //此case只有当listen fd有事件到达后触发主线程执行
555. sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen); //accept，得到client fd
556. if (sfd == -1) {
557. } else {
558. /**
559. accept成功后，调用dispatch_conn_new，把client fd交给 worker线程处理
560. 必须注意dispatch_conn_new 函数第二个参数：init_state，也就是
561. 创建连接对象的初始化状态，通过主线程分发给worker线程的client fd，最终
562. 建立的连接对象初始化状态为conn_new_cmd （当然这里只说的是TCP socket的情况，UDP socket暂不作分析）
563. */
564. dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST,
565. DATA_BUFFER_SIZE, tcp_transport);
566. }
567. stop = true;
568. break;
569. case conn_waiting: //等待数据
570. if (!update_event(c, EV_READ | EV_PERSIST)) {
571. if (settings.verbose > 0)
572. fprintf(stderr, "Couldn‘t update event\n");
573. conn_set_state(c, conn_closing);
574. break;
575. }
576. conn_set_state(c, conn_read);
577. stop = true;
578. break;
579. case conn_read: //执行读数据
580. res = IS_UDP(c->transport) ? try_read_udp(c) : try_read_network(c); //从网络上读取数据
581. switch (res) {
582. case READ_NO_DATA_RECEIVED:
583. conn_set_state(c, conn_waiting); //没有数据再继续等待
584. break;
585. case READ_DATA_RECEIVED: //成功接收数据则进入conn_parse_cmd分支，解析命令
586. conn_set_state(c, conn_parse_cmd);
587. break;
588. case READ_ERROR:
589. conn_set_state(c, conn_closing);
590. break;
591. case READ_MEMORY_ERROR: /* Failed to allocate more memory */
592. /* State already set by try_read_network */
593. break;
594. }
595. break;
596. case conn_parse_cmd :
597. /**
598. try_read_network后，到达conn_parse_cmd状态，但try_read_network并不确保每次到达
599. 的数据都足够一个完整的cmd（ascii协议情况下往往是没有"\r\n"，即回车换行），
600. 所以下面的try_read_command之所以叫try就是这个原因，
601. 当读到的数据还不够成为一个cmd的时候，返回0，conn继续进入conn_waiting状态等待更多的数据到达。
602. */
603. if (try_read_command(c) == 0) {
604. /* wee need more data! */
605. conn_set_state(c, conn_waiting);
606. }
607. break;
608. case conn_new_cmd:
609. /* Only process nreqs at a time to avoid starving other
610. connections */
611. /*
612. 这里的reqs是请求的意思，其实叫“命令”更准确。一次event发生，有可能包含多个命令，
613. 从client fd里面read到的一次数据，不能保证这个数据只是包含一个命令，有可能是多个
614. 命令数据堆在一起的一次事件通知。这个nreqs是用来控制一次event最多能处理多少个命令。
615. */
616. --nreqs;
617. if (nreqs >= 0) {
618. /**
619. 准备执行命令。为什么叫reset cmd，reset_cmd_handler其实做了一些解析执行命令之前
620. 的初始化动下一个，都会重新进入这个case作。而像上面说的，一次event有可能有多个命令，每执行一个命令，如果还有
621. conn_new_cmd，reset一下再执行下一个命令。
622. */
623. reset_cmd_handler(c);
624. } else {
625. //省略
626. stop = true;
627. }
628. break;
629. case conn_nread:
630. /**
631. 由process_update_command执行后进入此状态，process_update_command函数只执行了add/set/replace 等命令的一半，
632. 剩下的一半由这里完成。
633. 在这插一下memcached的命令解析协议，add/set/replace等这种写类型的命令，分为两块，或者说要有两次的“回车”，
634. 如：
635. 1)set key 0 0 4\r\n
636. 2)haha\r\n
637. 第1)行是命令，第2)行是数据块。
638. 例如如果是上面的set命令，process_update_command只完成了第1)行，分配了item空间，但还没有把value塞到对应的
639. item中去。因此，在这一半要完成的动作就是把第2)行代表value的数据读出来，塞到刚拿到的item空间中去
640. */
641. /*
642. 下面的rlbytes字段表示要读的value数据还剩下多少字节
643. 如果是第一次由process_update_command进入到此，rlbytes此时在process_update_command中被初始化为item->nbytes，
644. 即value的总字节数，由第1)行命令中声明，即上面例子第1)行的"4"
645. */
646. if (c->rlbytes == 0) {
647. /**
648. 如果要读value数据的都读完了，就调用complete_nread完成收尾工作，程序会跟着complete_nread进入下一个
649. 状态。所以执行完complete_nread会break;
650. */
651. complete_nread(c);
652. break;
653. }
654. //如果还有数据没读完，继续往下执行。可知，下面的动作就是继续从buffer中读value数据往item中的data的value位置塞。
655. /* Check if rbytes < 0, to prevent crash */
656. if (c->rlbytes < 0) {
657. if (settings.verbose) {
658. fprintf(stderr, "Invalid rlbytes to read: len %d\n", c->rlbytes);
659. }
660. conn_set_state(c, conn_closing);
661. break;
662. }
663. /* first check if we have leftovers in the conn_read buffer */
664. if (c->rbytes > 0) {
665. /**
666. 取rbytes与rlbytes中最小的值。
667. 为啥？
668. 因为这里我们的目的是剩下的还没读的value的字节，而rlbytes代表的是还剩下的字节数
669. 如果rlbytes比rbytes小，只读rlbytes长度就够了，rbytes中多出来的部分不是我们这个时候想要的
670. 如果rbytes比rlbytes小，即使你要rlbytes这么多，但buffer中没有这么多给你读。
671. */
672. int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes;
673. if (c->ritem != c->rcurr) {
674. memmove(c->ritem, c->rcurr, tocopy); //往分配的item中塞，即为key设置value的过程
675. }
676. c->ritem += tocopy;
677. c->rlbytes -= tocopy;
678. c->rcurr += tocopy;
679. c->rbytes -= tocopy;
680. if (c->rlbytes == 0) {
681. break;
682. }
683. }
684. //这里往往是我们先前读到buffer的数据还没足够的情况下，从socket中读。
685. /* now try reading from the socket */
686. res = read(c->sfd, c->ritem, c->rlbytes);//往分配的item中塞，即为key设置value的过程
687. if (res > 0) {
688. pthread_mutex_lock(&c->thread->stats.mutex);
689. c->thread->stats.bytes_read += res;
690. pthread_mutex_unlock(&c->thread->stats.mutex);
691. if (c->rcurr == c->ritem) {
692. c->rcurr += res;
693. }
694. c->ritem += res;
695. c->rlbytes -= res;
696. break;
697. }
698. if (res == 0) { /* end of stream */
699. conn_set_state(c, conn_closing);
700. break;
701. }
702. if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
703. if (!update_event(c, EV_READ | EV_PERSIST)) {
704. if (settings.verbose > 0)
705. fprintf(stderr, "Couldn‘t update event\n");
706. conn_set_state(c, conn_closing);
707. break;
708. }
709. stop = true;
710. break;
711. }
712. /* otherwise we have a real error, on which we close the connection */
713. if (settings.verbose > 0) {
714. }
715. conn_set_state(c, conn_closing);
716. break;
717. case conn_swallow:
718. //省略
719. conn_set_state(c, conn_closing);
720. break;
721. case conn_write:
722. if (c->iovused == 0 || (IS_UDP(c->transport) && c->iovused == 1)) {
723. if (add_iov(c, c->wcurr, c->wbytes) != 0) {
724. }
725. }
726. /* fall through... */ //直接进入conn_mwrite分支
727. case conn_mwrite:
728. if (IS_UDP(c->transport) && c->msgcurr == 0 && build_udp_headers(c) != 0) {
729. if (settings.verbose > 0)
730. fprintf(stderr, "Failed to build UDP headers\n");
731. conn_set_state(c, conn_closing);
732. break;
733. }
734. switch (transmit(c)) { //执行transmit发送数据到客户端
735. case TRANSMIT_COMPLETE:
736. if (c->state == conn_mwrite) {
737. conn_release_items(c);
738. if(c->protocol == binary_prot) {
739. conn_set_state(c, c->write_and_go);
740. } else {
741. conn_set_state(c, conn_new_cmd); //重新回到conn_new_cmd分支等待新命令
742. }
743. } else if (c->state == conn_write) {
744. if (c->write_and_free) {
745. free(c->write_and_free);
746. c->write_and_free = 0;
747. }
748. conn_set_state(c, c->write_and_go);
749. } else {
750. if (settings.verbose > 0)
751. fprintf(stderr, "Unexpected state %d\n", c->state);
752. conn_set_state(c, conn_closing);
753. }
754. break;
755. case TRANSMIT_INCOMPLETE:
756. case TRANSMIT_HARD_ERROR:
757. break; /* Continue in state machine. */
758. case TRANSMIT_SOFT_ERROR:
759. stop = true;
760. break;
761. }
762. break;
763. case conn_closing:
764. if (IS_UDP(c->transport))
765. conn_cleanup(c);
766. else
767. conn_close(c);
768. stop = true;
769. break;
770. case conn_closed:
771. /* This only happens if dormando is an idiot. */
772. abort();
773. break;
774. case conn_max_state:
775. assert(false);
776. break;
777. }
778. }
779. return;
780. }
781. //事件处理器主要调用drive_machine状态机执行事件处理
782. void event_handler(const int fd, const short which, void *arg) {
783. conn *c;
784. c = (conn *)arg;
785. assert(c != NULL);
786. c->which = which;
787. if (fd != c->sfd) {
788. if (settings.verbose > 0)
789. fprintf(stderr, "Catastrophic: event fd doesn‘t match conn fd!\n");
790. conn_close(c);
791. return;
792. }
793. drive_machine(c); //调用drive_machine处理事件发生后的业务逻辑。
794. return;
795. }
796. //创建socket
797. static int new_socket(struct addrinfo *ai) {
798. int sfd;
799. int flags;
800. if ((sfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) == -1) {
801. return -1;
802. }
803. if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 ||
804. fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
805. perror("setting O_NONBLOCK");
806. close(sfd);
807. return -1;
808. }
809. return sfd;
810. }
811. static int server_sockets(int port, enum network_transport transport,
812. FILE *portnumber_file) {
813. if (settings.inter == NULL) { //如果没有指定ip
814. return server_socket(settings.inter, port, transport, portnumber_file);
815. } else {
816. //省略指定的情况
817. return ret;
818. }
819. }
820. //main函数
821. int main (int argc, char **argv) {
822. //省略一些变量定义，下面使用到时会说明
823. settings_init(); //初始化配置
824. setbuf(stderr, NULL);
825. //memcached启动时指定的启动参数配置，在此省略
826. while (-1 != (c = getopt(argc, argv, xxx
827. }
828. if (hash_init(hash_type) != 0) { //哈希算法初始化
829. fprintf(stderr, "Failed to initialize hash_algorithm!\n");
830. exit(EX_USAGE);
831. }
832. //身份验证的代码
833. if (getuid() == 0 || geteuid() == 0) {
834. }
835. //以常驻进程方式运行
836. if (do_daemonize) {
837. if (daemonize(maxcore, settings.verbose) == -1) {
838. }
839. }
840. main_base = event_init(); //全局的main_base变量
841. stats_init(); //初始化全局统计
842. assoc_init(settings.hashpower_init); //初始化哈希表
843. conn_init(); //初始化连接对象，配置
844. slabs_init(settings.maxbytes, settings.factor, preallocate); //初始化slabs，这里会对一些内存管理进行初始化
845. //初始化主线程，参数是worker线程个数，和当前主线程的event_base
846. thread_init(settings.num_threads, main_base);
847. if (start_assoc_maintenance_thread() == -1) {//启动哈希表维护线程，详见assoc::start_assoc_maintenance_thread
848. exit(EXIT_FAILURE);
849. }
850. /**
851. 如果开启了slab可重分配，则启动slab维护线程
852. 注意slab维护线程（下面简称s）与哈希表维护线程（下面简称哈）逻辑上不一样的地方：
853. s是只要用户开启了slab_reassign，那么启动memcached服务的时候这个线程就同时启动的
854. 而哈虽然一开始就启动了，但是处理睡眠态，在需要的时候，memcached才发送信息唤醒。
855. 详见assoc::start_assoc_maintenance_thread和slabs::start_slab_maintenance_thread
856. */
857. if (settings.slab_reassign &&
858. start_slab_maintenance_thread() == -1) {
859. exit(EXIT_FAILURE);
860. }
861. init_lru_crawler(); //初始化item爬虫
862. //这里省略unix socket监听方式和UDP的代码
863. /*
864. 睡眠一下，只是提供一定的时候让socket打开而已
865. */
866. usleep(1000);
867. if (stats.curr_conns + stats.reserved_fds >= settings.maxconns - 1) {
868. fprintf(stderr, "Maxconns setting is too low, use -c to increase.\n");
869. exit(EXIT_FAILURE);
870. }
871. if (pid_file != NULL) {
872. save_pid(pid_file); //创建pid文件
873. }
874. drop_privileges();
875. //进入事件循环
876. if (event_base_loop(main_base, 0) != 0) {
877. retval = EXIT_FAILURE;
878. }
879. stop_assoc_maintenance_thread(); //进程退出之前停止哈希表维护线程
880. //进程退出的收尾工作
881. return retval;
882. }