memcached线程模型

古城微笑少年丶 2021-11-27 00:56 316阅读 0赞

1.main函数中调用thread\_init()，初始化setting.num\_threads个worker线程以及一个主线程dispatcher\_thread。  
每个worker线程用pipe创建一个管道，并注册libevent事件，当管道的读端可以读时，就调用thread\_libevent\_process()函数。  
thread\_libevent\_process()做的事情等下再说。

**\[cpp\]** [view plain][] [copy][view plain]

1.  **void** thread\_init(**int** nthreads, **struct** event\_base \*main\_base) \{ 
2.   threads = calloc(nthreads, **sizeof**(LIBEVENT\_THREAD)); 
3.  
4.  //初始化dispatcher线程
5.   dispatcher\_thread.base = main\_base; 
6.   dispatcher\_thread.thread\_id = pthread\_self(); 
7.  
8.  //初始化nthreads个worker线程
9.  **for** (i = 0; i < nthreads; i++) \{ 
10. **int** fds\[2\]; 
11. **if** (pipe(fds)) \{ 
12.  perror("Can't create notify pipe"); 
13.  exit(1); 
14.  \} 
15. 
16. //每个线程打开一个管道
17.  threads\[i\].notify\_receive\_fd = fds\[0\]; 
18.  threads\[i\].notify\_send\_fd = fds\[1\]; 
19. 
20. //注册libevent事件。当该线程的管道的读端可以读时，调用thread\_libevent\_process()
21.  setup\_thread(&threads\[i\]); 
22.  \} 
23. 
24. /\* Create threads after we've done all the libevent setup. \*/
25. /\* 创建worker线程，并让每个线程进入event\_base\_loop()循环。
26.  前面的threads数组只是创建了nthreads个LIBEVENT\_THREAD对象，
27.  并初始化了libevent以及其他信息。并没有真正开始一个新的进程。
28.  在create\_worker里才真正调用了pthread\_create，
29.  并且每个进程都进入了event\_base\_loop()
30.  \*/
31. **for** (i = 0; i < nthreads; i++) \{ 
32.  create\_worker(worker\_libevent, &threads\[i\]); 
33.  \} 
34. \} 
35. 
36. **static****void** setup\_thread(LIBEVENT\_THREAD \*me) \{ 
37.  me->base = event\_init(); 
38. 
39. /\* Listen for notifications from other threads \*/
40. //当读端可以读时，调用thread\_libevent\_process()函数
41.  event\_set(&me->notify\_event, me->notify\_receive\_fd, 
42.  EV\_READ | EV\_PERSIST, thread\_libevent\_process, me); 
43.  event\_base\_set(me->base, &me->notify\_event); 
44. 
45. **if** (event\_add(&me->notify\_event, 0) == -1) \{ 
46.  fprintf(stderr, "Can't monitor libevent notify pipe\\n"); 
47.  exit(1); 
48.  \} 
49. 
50. /\*初始化每个线程的new\_conn\_queue成员。
51.  new\_conn\_queue成员是一个conn\_queue指针，相当于一个队列，
52.  记录分配到该线程的，等待new一个conn对象的那些item的信息。
53.  每次调用thread\_libevent\_process()时，
54.  就从该队列中取出一个item，然后建立一个conn\*/
55.  me->new\_conn\_queue = malloc(**sizeof**(**struct** conn\_queue)); 
56.  cq\_init(me->new\_conn\_queue); 
57. 
58. \} 
59. 
60. **static****void** create\_worker(**void** \*(\*func)(**void** \*), **void** \*arg) \{ 
61.  pthread\_t **thread**; 
62.  pthread\_attr\_t attr; 
63. **int** ret; 
64. 
65.  pthread\_attr\_init(&attr); 
66. 
67. **if** ((ret = pthread\_create(&**thread**, &attr, func, arg)) != 0) \{ 
68.  fprintf(stderr, "Can't create thread: %s\\n", 
69.  strerror(ret)); 
70.  exit(1); 
71.  \} 
72. \} 
73. 
74. **static****void** \*worker\_libevent(**void** \*arg) \{ 
75.  LIBEVENT\_THREAD \*me = arg; 
76.  event\_base\_loop(me->base, 0); 
77. **return** NULL; 
78. \}

2.main函数中，然后调用server\_sockets，在指定的端口和ip地址上分别建立tcp和udp的监听。

然后

1)如果是TCP，就调用conn\_new(sfd, conn\_listening, EV\_READ|EV+PERSIST, 1, tcp\_transport, main\_base)，该函数在main\_base上建立libevent事件，当sfd可读时，调用event\_handler(). 而event\_handler()又调用drive\_machine().

2)如果是UDP，就调用dispatch\_conn\_new(sfd, conn\_read, EV\_READ | EV\_PERSIST, UDP\_READ\_BUFFER\_SIZE, udp\_transport)，该函数用round-robin的方法找到一个worker线程，然后new一个CQ\_ITEM对象，把该item对象放到worker线程的new\_conn\_queue队列中。然后，通知该worker线程。通知的方法是，往该线程的写端写一个字节，这样，该线程的读端就可读了。由于之前已经讲过，每个worker线程注册了libevent事件，当读端可读时，就调用thread\_libevent\_process()。  
server\_sockets()调用server\_socket(), server\_socket()先在指定端口和ip上建立socket，得到文件描述符sfd，然后bind，然后

**\[cpp\]** [view plain][] [copy][view plain]

1.  **static****int** server\_socket(**const****char** \*interface, 
2.  **int** port, 
3.  **enum** network\_transport transport, 
4.  **FILE** \*portnumber\_file) \{ 
5.  **if** ((sfd = new\_socket(next)) == -1) \{ 
6.   ... 
7.   \} 
8.  **if** (bind(sfd, next->ai\_addr, next->ai\_addrlen) == -1) \{ 
9.   ... 
10.  \} 
11. **if** (!IS\_UDP(transport) && listen(sfd, settings.backlog) == -1) \{ 
12.  ... 
13.  \} 
14. 
15. **if** (IS\_UDP(transport)) \{ 
16. **int** c; 
17. 
18. **for** (c = 0; c < settings.num\_threads\_per\_udp; c++) \{ 
19. /\* this is guaranteed to hit all threads because we round-robin \*/
20.  dispatch\_conn\_new(sfd, conn\_read, EV\_READ | EV\_PERSIST, 
21.  UDP\_READ\_BUFFER\_SIZE, transport); 
22.  \} 
23.  \} **else** \{ 
24. **if** (!(listen\_conn\_add = conn\_new(sfd, conn\_listening, 
25.  EV\_READ | EV\_PERSIST, 1, 
26.  transport, main\_base))) \{ 
27.  ... 
28.  \} 
29.  \} 
30. \}

3. 继续2.1

如果是TCP，把conn的状态设为conn\_listening, 设置libevent事件。当一个新的tcp connection到达时，进入drive\_machine。

**\[cpp\]** [view plain][] [copy][view plain]

1.  **static****void** drive\_machine(conn \*c) \{ 
2.  **switch**(c->state) \{ 
3.  **case** conn\_listening: 
4.   addrlen = **sizeof**(addr); 
5.   sfd = accept(c->sfd, (**struct** sockaddr \*)&addr, &addrlen)); 
6.   dispatch\_conn\_new(sfd, conn\_new\_cmd, EV\_READ | EV\_PERSIST, 
7.   DATA\_BUFFER\_SIZE, tcp\_transport); 
8.  
9.   ... 
10. 
11. \}

调用dispatch\_conn\_new，把新的tcp connection的sfd指派到某个worker线程上。

4. 继续2.2

注意server\_socket中，当协议是UDP时，调用dispatch\_new\_conn，但是是在一个for循环中。就是说，对于一个udp的描述符，指派了settings.num\_threads\_per\_udp个线程来监控该udp描述符。

这叫做“惊群”。当一个udp连接到达时，所有的libevent都能检测到该事件，但是只有一个线程调用recvfrom能返回数据。其他线程都返回失败。

udp用惊群而tcp不用的原因可能是：对于tcp，主线程（即dispatch\_thread）会一直监控是否有新的tcp connection到达。如果到达，就会指派一个worker thread处理它。而对于udp，如果不用惊群，那么只有一个worker 线程监控udp 请求。因此，当该线程在处理一个udp请求时，其他的udp请求就不能得到及时处理。

另一个原因是：

“TCP是定然不能这样设计的,所以只能accept之后分发到特定线程,因为是字节流.  
而UDP是包,随便哪个线程处理包1,再换个线程处理包2也是没有问题的.”

[view plain]: http://blog.csdn.net/orzlzro/article/details/7532625#