Tilera多线程网络编程总结

浅浅的花香味﹌ 2022-07-25 16:57 169阅读 0赞

http://blog.csdn.net/zwleagle/article/details/8851400

http://blog.sina.com.cn/s/blog\_a574f78401015v2o.html

http://www.dssz.com/1341360.html

http://www.dssz.com/1341355.html

# [多线程网络处理服务器demo][demo] #

1.  \#include <stdio.h>  
2.  \#include <stdlib.h>  
3.  \#include <errno.h>  
4.  \#include <string.h>  
5.  \#include <sys/types.h>  
6.  \#include <netinet/in.h>  
7.  \#include <sys/socket.h>  
8.  \#include <sys/wait.h>  
9.  \#include <unistd.h>  
10. \#include <arpa/inet.h>  
11. //\#include <openssl/ssl.h>  
12. //\#include <openssl/err.h>  
13. \#include <fcntl.h>  
14. \#include <sys/epoll.h>  
15. \#include <sys/time.h>  
16. \#include <sys/resource.h>  
17. \#include <pthread.h>  
18. \#include <assert.h>  
19.   
20. //\#define DEBUG\_TILERA  
21.   
22.   
23. \#ifdef DEBUG\_TILERA  
24. \#include <tmc/alloc.h>  
25.   
26. \#include <arch/cycle.h>  
27. \#include <arch/spr.h>  
28.   
29. \#include <tmc/cpus.h>  
30. \#include <tmc/sync.h>  
31. \#include <tmc/task.h>  
32. \#endif  
33.   
34. /\* These are non-NULL pointers that will result in page faults 
35. \* under normal circumstances, used to verify that nobody uses 
36. \* non-initialized list entries. 
37. \*/  
38.   
39.   
40. \#define MAXBUF 1024  
41. \#define MAXEPOLLSIZE 500000  
42.   
43. \#define MAX\_THREAD\_NUMBER 200  
44. **int**  THREAD\_NUMBER = 50;  
45.   
46.   
47. **int** kdpfd;  
48. **struct** epoll\_event events\[MAXEPOLLSIZE\];  
49.   
50. **struct** epoll\_event thread\_events\[MAX\_THREAD\_NUMBER\]\[MAXEPOLLSIZE\];  
51. **int** fdpool\[MAX\_THREAD\_NUMBER\] = \{-1\};  
52.   
53. pthread\_t  handle\_receive\_thrdid\[MAX\_THREAD\_NUMBER\];  
54.   
55.   
56. **int** msgcount = 0;  
57. **int** timecount = 0;  
58.   
59. **int** count\_packet= 0;  
60.   
61. pthread\_mutex\_t connet\_count\_lock = PTHREAD\_MUTEX\_INITIALIZER;  
62. **int** connect\_count = 0;  
63.   
64. pthread\_mutex\_t curfds\_lock;  
65. **int** curfds;  
66.   
67. **char** buffer\[MAX\_THREAD\_NUMBER\]\[MAXBUF + 1\];  
68. pthread\_t thread\_count;  
69.   
70.   
71. cpu\_set\_t cpus;  
72.   
73.   
74.   
75. **void** BubbleSort(unsigned **char** R\[\],**int** n)  
76. \{  
77.     **int** i,j;  
78.     unsigned **char** temp;  
79.     **for** (i=0; i<n-1; i++ )  
80.     \{  
81.         **for** (j=n-2; j>=i; j--)  
82.         \{  
83.             **if** (R\[j\]>R\[j+1\])  
84.             \{  
85.                 temp=R\[j\];  
86.                 R\[j\]=R\[j+1\];  
87.                 R\[j+1\]=temp;  
88.             \}  
89.         \}  
90.     \}  
91. \}  
92.   
93. /\* 
94. setnonblocking - 设置句柄为非阻塞方式 
95. \*/  
96. **int** setnonblocking(**int** sockfd)  
97. \{  
98.     **if** (fcntl(sockfd, F\_SETFL, fcntl(sockfd, F\_GETFD, 0)|O\_NONBLOCK) == -1)  
99.     \{  
100.         **return** -1;  
101.     \}  
102.     **return** 0;  
103. \}  
104.   
105. /\* 
106. handle\_message - 处理每个 socket 上的消息收发 
107. \*/  
108.   
109.   
110.   
111. **static** **void** \*handle\_count(**void**\* arg)  
112. \{  
113.     **int** precount, speed;  
114.   
115.     **while**(1)  
116.     \{  
117.         precount = msgcount;  
118.         sleep(5);  
119.         timecount += 5;  
120.         //printf("The tcp connection count is %d\\n",count\_tcp);  
121.         //printf("The received packets count is %d, time %d\\n",msgcount, timecount);  
122.         speed = msgcount - precount ;  
123.         printf("The received speed  is %d/5seconds, connect %d, tatol packets %d\\n",speed, connect\_count, msgcount);  
124.     \}  
125.   
126.     **return** NULL;  
127. \}  
128.   
129. **static** **void** \* handl\_receive\_msg(**void** \* arg)  
130. \{  
131.     **int** fdind = 0;  
132.     **int** nfds = 0;  
133.     **int** len;  
134.     **struct** epoll\_event ev;  
135.     **int** fdi;  
136.   
137.     **char**\* buf ;  
138.   
139.     fdind = (**int**)arg;  
140.   
141.     buf = (**char**\*)&buffer\[fdind\];  
142.   
143.     // printf("fd... %d \\n", fdind);  
144.   
145.     **while**(1)  
146.     \{  
147.         nfds = epoll\_wait(fdpool\[fdind\], &thread\_events\[fdind\]\[0\], MAXEPOLLSIZE, -1);  
148.         **if** (nfds == -1)  
149.         \{  
150.             perror("epoll\_wait");  
151.             **break**;  
152.         \}  
153.   
154.         **for**( fdi = 0; fdi < nfds; fdi++)  
155.         \{  
156.             **if**((thread\_events\[fdind\]\[fdi\].events & EPOLLIN)  
157.                /\*&&(!(thread\_events\[fdind\]\[fdi\].events & EPOLLRDHUP))\*/)  
158.             \{  
159.                 **while**((-1 != (len = recv(thread\_events\[fdind\]\[fdi\].data.fd, buf, MAXBUF, 0)))  
160.                       ||((-1 == len) && (EAGAIN != errno)))  
161.                 \{  
162.                     //perror("recv error ");  
163.                     //printf("recv error %d  fd %d\\n", errno, thread\_events\[fdind\]\[fdi\].data.fd);  
164.                     //goto next;  
165.   
166.   
167.                     **if** (len > 0)  
168.                     \{  
169.   
170.                         /\*printf 
171.                         ("%d receive message success   total %d bytes data  msgcount %d\\n", 
172.                          new\_fd,   len, msgcount);\*/  
173.                         msgcount++;  
174.                         BubbleSort(buf ,len);  
175.                     \}  
176.                     **else** **if**(len == 0)  
177.                     \{  
178.                         //printf("the socket %d is closed \\n", new\_fd);  
179.                         epoll\_ctl(fdpool\[fdind\], EPOLL\_CTL\_DEL, thread\_events\[fdind\]\[fdi\].data.fd,&ev);  
180.                         close(thread\_events\[fdind\]\[fdi\].data.fd);  
181.                         pthread\_mutex\_lock(&connet\_count\_lock);  
182.                         connect\_count--;  
183.                         pthread\_mutex\_unlock (&connet\_count\_lock);  
184.   
185.                         **break**;  
186.                     \}  
187.                     **else**  
188.                     \{  
189.   
190.                         printf(" socket %d receive message fail error code: %d,  error message: '%s'\\n",  
191.                                thread\_events\[fdind\]\[fdi\].events, errno, strerror(errno));  
192.   
193.   
194.                         epoll\_ctl(fdpool\[fdind\], EPOLL\_CTL\_DEL, thread\_events\[fdind\]\[fdi\].data.fd,&ev);  
195.                         close(thread\_events\[fdind\]\[fdi\].data.fd);  
196.                         pthread\_mutex\_lock(&connet\_count\_lock);  
197.                         connect\_count--;  
198.                         pthread\_mutex\_unlock (&connet\_count\_lock);  
199.   
200.   
201.                         //pthread\_mutex\_lock (&(curfds\_lock));  
202.                         //curfds--;  
203.                         //pthread\_mutex\_unlock (&(curfds\_lock));  
204.                         **break**;  
205.                     \}  
206.   
207.                 \}  
208.             \}  
209.             /\* else if((thread\_events\[fdind\]\[fdi\].events & EPOLLRDHUP)) 
210.              \{  
211.                  epoll\_ctl(fdpool\[fdind\], EPOLL\_CTL\_DEL, thread\_events\[fdind\]\[fdi\].data.fd,&ev); 
212.                  close(thread\_events\[fdind\]\[fdi\].data.fd); 
213.                  pthread\_mutex\_lock(&connet\_count\_lock); 
214.                  connect\_count--; 
215.                  pthread\_mutex\_unlock (&connet\_count\_lock); 
216.  
217.                  printf("event.... %x \\n", thread\_events\[fdind\]\[fdi\].events); 
218.  
219.              \}\*/  
220.             **else**  
221.             \{  
222.                 printf("other event %u\\n",thread\_events\[fdind\]\[fdi\].events );  
223.             \}  
224.   
225.   
226.         \}  
227.     \}  
228.   
229.     **return** NULL;  
230.   
231. \}  
232.   
233.   
234.   
235. **int** fd\_index = 0;  
236.   
237. **int** main(**int** argc, **char** \*\*argv)  
238. \{  
239.     **int** listener, new\_fd,  nfds, n, ret;  
240.     socklen\_t len;  
241.     **struct** sockaddr\_in my\_addr, their\_addr;  
242.     unsigned **int** myport, lisnum;  
243.     **struct** epoll\_event ev;  
244.     **struct** rlimit rt;  
245.     **int**    fdind;  
246.     **int**   ind ;  
247.   
248.     **if**(5 != argc)  
249.     \{  
250.   
251.         printf("Usage: %s <thread\_number(0 ~ 200)> <port(0-65535)> <listen queue number>  <IP Address>   \\n", argv\[0\]);  
252.         exit(1);  
253.   
254.     \}  
255.   
256.     **if**(argv\[1\])  
257.         THREAD\_NUMBER = atoi(argv\[1\]);  
258.   
259.     **if** (argv\[2\])  
260.         myport = atoi(argv\[2\]);  
261.     **else**  
262.         myport = 7838;  
263.   
264.     **if** (argv\[3\])  
265.         lisnum = atoi(argv\[3\]);  
266.     **else**  
267.         lisnum = 2;  
268.   
269.   
270.   
271. \#ifdef DEBUG\_TILERA  
272.     **if** (tmc\_cpus\_get\_my\_affinity(&cpus) != 0)  
273.     \{  
274.         printf("tmc\_cpus\_get\_my\_affinity() failed.\\n");  
275.         tmc\_task\_die("tmc\_cpus\_get\_my\_affinity() failed.");  
276.     \}  
277.     **if** (tmc\_cpus\_count(&cpus) < MAX\_THREAD)  
278.     \{  
279.         printf("\\nInsufficient cpus available.\\n");  
280.         tmc\_task\_die("Insufficient cpus available.");  
281.     \}  
282. \#endif  
283.   
284.   
285.     pthread\_mutex\_init (&connet\_count\_lock, NULL);  
286.     pthread\_mutex\_init (&(curfds\_lock), NULL);  
287.   
288.     **for**( ind = 0; ind < THREAD\_NUMBER; ind++ )  
289.     \{  
290.   
291.         fdpool\[ind\] = epoll\_create(MAXEPOLLSIZE);  
292.   
293.     \}  
294.   
295.     **for**( ind = 0; ind < THREAD\_NUMBER; ind++)  
296.     \{  
297.         pthread\_create(&handle\_receive\_thrdid\[ind\], NULL, &handl\_receive\_msg, (**void**\*)ind);  
298.     \}  
299.   
300.     **if** (pthread\_create(&thread\_count, NULL, &handle\_count, NULL) != 0)  
301.     \{  
302. \#ifdef DEBUG\_TILERA  
303.         tmc\_task\_die("pthread\_create() failed.");  
304. \#endif  
305.     \}  
306.   
307.     /\* 设置每个进程允许打开的最大文件数 \*/  
308.     rt.rlim\_max = rt.rlim\_cur = MAXEPOLLSIZE;  
309.     **if** (setrlimit(RLIMIT\_NOFILE, &rt) == -1)  
310.     \{  
311.         perror("setrlimit");  
312.         exit(1);  
313.     \}  
314.     **else** printf("set the system resource success!\\n");  
315.   
316.     /\* 开启 socket 监听 \*/  
317.     **if** ((listener = socket(PF\_INET, SOCK\_STREAM, 0)) == -1)  
318.     \{  
319.         perror("socket");  
320.         exit(1);  
321.     \}  
322.     **else**  
323.         printf("socket create success!n");  
324.   
325.     setnonblocking(listener);  
326.   
327.     bzero(&my\_addr, **sizeof**(my\_addr));  
328.     my\_addr.sin\_family = PF\_INET;  
329.     my\_addr.sin\_port = htons(myport);  
330.     **if** (argv\[4\])  
331.         my\_addr.sin\_addr.s\_addr = inet\_addr(argv\[4\]);  
332.     **else**  
333.         my\_addr.sin\_addr.s\_addr = INADDR\_ANY;  
334.   
335.     **if** (bind  
336.         (listener, (**struct** sockaddr \*) &my\_addr, **sizeof**(**struct** sockaddr))  
337.         == -1)  
338.     \{  
339.         perror("bind");  
340.         exit(1);  
341.     \}  
342.     **else**  
343.         printf("IP address and port bing success!\\n");  
344.   
345.     **if** (listen(listener, lisnum) == -1)  
346.     \{  
347.         perror("listen");  
348.         exit(1);  
349.     \}  
350.     **else**  
351.         printf("start to work!\\n");  
352.   
353.   
354.   
355.   
356.   
357.     /\* 创建 epoll 句柄，把监听 socket 加入到 epoll 集合里 \*/  
358.     kdpfd = epoll\_create(MAXEPOLLSIZE);  
359.     len = **sizeof**(**struct** sockaddr\_in);  
360.     ev.events = EPOLLIN;  
361.     ev.data.fd = listener;  
362.     **if** (epoll\_ctl(kdpfd, EPOLL\_CTL\_ADD, listener, &ev) < 0)  
363.     \{  
364.         fprintf(stderr, "epoll set insertion error: fd=%d\\n", listener);  
365.         **return** -1;  
366.     \}  
367.     **else**  
368.         printf("listen socket add to epoll success\\n");  
369.   
370.   
371.   
372.   
373.     curfds = 1;  
374.     **while** (1)  
375.     \{  
376.         /\* 等待有事件发生 \*/  
377.         nfds = epoll\_wait(kdpfd, events, MAXEPOLLSIZE, -1);  
378.         **if** (nfds == -1)  
379.         \{  
380.             perror("epoll\_wait");  
381.             **break**;  
382.         \}  
383.         /\* 处理所有事件 \*/  
384.         **for** (n = 0; n < nfds; ++n)  
385.         \{  
386.             // printf("The number of fd %d \\n", nfds);  
387.             **if** (events\[n\].data.fd == listener)  
388.             \{  
389.                 new\_fd = accept(listener, (**struct** sockaddr \*) &their\_addr,  
390.                                 &len);  
391.                 **if** (new\_fd < 0)  
392.                 \{  
393.                     perror("accept");  
394.                     **continue**;  
395.                 \}  
396.                 //else  
397.                 // printf("connect from  %x:%x， allocate socket for %x\\n", inet\_ntoa(their\_addr.sin\_addr), ntohs(their\_addr.sin\_port), new\_fd);  
398.                 pthread\_mutex\_lock(&connet\_count\_lock);  
399.                 connect\_count++;  
400.                 pthread\_mutex\_unlock (&connet\_count\_lock);  
401.                 setnonblocking(new\_fd);  
402.                 ev.events = EPOLLIN | EPOLLET;  
403.                 ev.data.fd = new\_fd;  
404.   
405.                 fdind = fd\_index % THREAD\_NUMBER;  
406.   
407.                 **if** (epoll\_ctl(fdpool\[fdind\], EPOLL\_CTL\_ADD, new\_fd, &ev) < 0)  
408.                 \{  
409.                     fprintf(stderr, "add socket '%d' to epoll fail %s\\n",  
410.                             new\_fd, strerror(errno));  
411.                     // pool\_destroy ();  
412.   
413.                     **return** -1;  
414.                 \}  
415.   
416.                 fd\_index++;  
417.                 //pthread\_mutex\_lock (&(curfds\_lock));  
418.                 // curfds++;  
419.                 // pthread\_mutex\_unlock (&(curfds\_lock));  
420.   
421.   
422.             \}  
423.             **else**  
424.             \{  
425.                 printf("other event \\n");  
426.             \}  
427.         \}  
428.     \}  
429.     close(listener);  
430.   
431.     // pool\_destroy ();  
432.     **return** 0;  
433. \}

## tilera试用 ##

tilera处理器架构

前段时间同事搞来了一台使用tilera处理器的服务器，没错，就是那个由MIT专家做的、64个核的处理器，我非常好奇，所以也登上去体验体验。

tilera在硬件之上做了一个薄薄的软件层，叫hypervisior，再在这个hypervisior上装了一个linux，然后是gcc/g++/gdb等一套工具链。当然，这个linux是改过的，在内核arch/tile/目录里增加了东西，但是即使是增加了kernel代码，也只能跑在hypervisior上，不能直接跑在tilera硬件上。

我们用的是tilepro处理器，32位，64个核，每个核却只有863MHZ，所以多进程/多线程的程序有福了，我试了一下make -j60，感觉上确实比较快，但是，configure就慢得夸张。

在tilera上安装apache+php的过程中遇到几个小问题，主要是因为这个linux环境比较荒芜：

1. 从源码安装软件时，运行./configure遇到“configure: error: C compiler cannot create executables”,解决方法：

export CC=”gcc”

export CPP=”gcc -E”

2. ./configure还会遇到不能识别当前机器的处理器类型，解决方法：

./configure --build=i386

不用担心这个欺骗性的i386选项，对于可移植的c代码软件，这样不会造成什么问题。

我安装php最后还是失败了，所以拉倒，改装nginx做测试，起了20个nginx进程，通过千兆网卡做压力，却只能达到3000左右的QPS，这显然太低了。于是问了tilera的技术支持，他反馈说他们做过memcached的实验（据他说，facebook已经在用tilera机器专跑memcached），能到60万QPS，但是nginx多进程却很慢，他们也很疑惑，目前还在研究为什么。

看来tilera的软件层目前还偏薄弱，等一段时间吧，等linux-2.6.36稳定，且tilera的3.0版本的开发工具套件变为stable，我们再来关注关注。

====== 2011.01.06 ======

今天tilera公司的顾冉同学发来新消息，他们就用和我们一样的硬件环境（一片tilepro）和软件环境（MDE-2.1.0），10个左右的并发nginx，达到了将近2万QPS，比x86上的apache差一些，但是比我的测试结果已经好很多了。

也许是我的配置有问题，有待以后研究了。

[demo]: http://blog.csdn.net/zwleagle/article/details/8851400