用Mochiweb打造百万级Comet应用，第三部分（续）

提示：如有转载请注明作者 小游戏 及出处

 

原文：A Million-user Comet Application with Mochiweb, Part 3

参考资料：Comet--基于 HTTP 长连接、无须在浏览器端安装插件的“服务器推”技术为“Comet”

               MochiWeb--建立轻量级HTTP服务器的Erlang库

 

floodtest2.erl

1. -module ( floodtest2) .
2. -compile( export_all) .
3. -define( SERVERADDR , "10.1.2.3" ) . % where mochiweb is running
4. -define( SERVERPORT , 8000 ) .
5.  
6. % Generate the config in bash like so (chose some available address space):
7. % EACH=62000; for i in `seq 1 17`; do echo "{{10,0,0,$i}, $((($i-1)*$EACH+1)), $(($i*$EACH))}, "; done
8.  
9. run( Interval ) ->
10.         Config = [
11. { { 10 ,0 ,0 ,1 } , 1 , 62000 } ,
12. { { 10 ,0 ,0 ,2 } , 62001 , 124000 } ,
13. { { 10 ,0 ,0 ,3 } , 124001 , 186000 } ,
14. { { 10 ,0 ,0 ,4 } , 186001 , 248000 } ,
15. { { 10 ,0 ,0 ,5 } , 248001 , 310000 } ,
16. { { 10 ,0 ,0 ,6 } , 310001 , 372000 } ,
17. { { 10 ,0 ,0 ,7 } , 372001 , 434000 } ,
18. { { 10 ,0 ,0 ,8 } , 434001 , 496000 } ,
19. { { 10 ,0 ,0 ,9 } , 496001 , 558000 } ,
20. { { 10 ,0 ,0 ,10 } , 558001 , 620000 } ,
21. { { 10 ,0 ,0 ,11 } , 620001 , 682000 } ,
22. { { 10 ,0 ,0 ,12 } , 682001 , 744000 } ,
23. { { 10 ,0 ,0 ,13 } , 744001 , 806000 } ,
24. { { 10 ,0 ,0 ,14 } , 806001 , 868000 } ,
25. { { 10 ,0 ,0 ,15 } , 868001 , 930000 } ,
26. { { 10 ,0 ,0 ,16 } , 930001 , 992000 } ,
27. { { 10 ,0 ,0 ,17 } , 992001 , 1054000 } ] ,
28.         start( Config , Interval ) .
29.  
30. start( Config , Interval ) ->
31.         Monitor = monitor( ) ,
32.         AdjustedInterval = Interval / length( Config ) ,
33.         [ spawn( fun start/5 , [ Lower , Upper , Ip , AdjustedInterval , Monitor ] )
34.           || { Ip , Lower , Upper }  <- Config ] ,
35.         ok.
36.  
37. start( LowerID , UpperID , _, _, _) when LowerID == UpperID -> done ;
38. start( LowerID , UpperID , LocalIP , Interval , Monitor ) ->
39.         spawn ( fun connect/5 , [ ?SERVERADDR , ?SERVERPORT , LocalIP , "/test/" ++LowerID , Monitor ] ) ,
40.         receive after Interval -> start ( LowerID + 1 , UpperID , LocalIP , Interval , Monitor ) end .
41.  
42. connect( ServerAddr , ServerPort , ClientIP , Path , Monitor ) ->
43.         Opts = [ binary, { packet, 0 } , { ip, ClientIP } , { reuseaddr, true} , { active, false} ] ,
44.         { ok, Sock } = gen_tcp:connect ( ServerAddr , ServerPort , Opts ) ,
45.         Monitor ! open,
46.         ReqL = io_lib:format ( "GET ~s\r \n Host: ~s\r \n \r \n " , [ Path , ServerAddr ] ) ,
47.         Req = list_to_binary( ReqL ) ,
48.         ok = gen_tcp:send ( Sock , [ Req ] ) ,
49.         do_recv( Sock , Monitor ) ,
50.         ( catch gen_tcp:close ( Sock ) ) ,
51.         ok.
52.  
53. do_recv( Sock , Monitor ) ->
54.         case gen_tcp:recv ( Sock , 0 ) of
55.                 { ok, B } ->
56.                         Monitor ! { bytes, size( B ) } ,
57.                         io:format ( "Recvd ~s\n " , [ binary_to_list( B ) ] ) ,
58.                         io:format ( "Recvd ~w bytes\n " , [ size( B ) ] ) ,
59.                         do_recv( Sock , Monitor ) ;
60.                 { error, closed} ->
61.                         Monitor ! closed,
62.                         closed;
63.                 Other ->
64.                         Monitor ! closed,
65.                         io:format ( "Other:~w\n " ,[ Other ] )
66.         end .
67.  
68. % Monitor process receives stats and reports how much data we received etc:
69. monitor( ) ->
70.         Pid = spawn( ?MODULE , monitor0, [ { 0 ,0 ,0 ,0 } ] ) ,
71.         timer:send_interval ( 10000 , Pid , report) ,
72.         Pid .
73.  
74. monitor0( { Open , Closed , Chunks , Bytes } =S ) ->
75.         receive
76.                 report  -> io :format ( "{Open, Closed, Chunks, Bytes} = ~w\n " ,[ S ] ) ;
77.                 open    -> monitor0 ( { Open + 1 , Closed , Chunks , Bytes } ) ;
78.                 closed  -> monitor0 ( { Open , Closed + 1 , Chunks , Bytes } ) ;
79.                 chunk   -> monitor0 ( { Open , Closed , Chunks + 1 , Bytes } ) ;
80.                 { bytes, B } -> monitor0 ( { Open , Closed , Chunks , Bytes + B } )
81.         end .

作为一个初始的测试，我像第一部分描述的那样连接mochiweb应用 - 它简单的每隔10秒给每个客户端发送一条消息。

erl> c(floodtest2), floodtest2:run(20).

这很快就吃掉了我的内存。

像那样用gen_tcp打开很多连接吃掉了很多内存。 在没有任何其他调整的情况下我想它需要~36GB的内存以保证正常工作。 我没有兴趣试着优化我的快速破解的erlang http 客户端(在真实世界了,这将是1M个web浏览器), 在手的有多于32GB内存的机子只有那台我们产品数据库用机， 我不能找到一个很好的理由就因为测试这个让last.fm下线：）另外，它看起来它一直只能管理打开64,500个端口 。

从这一点我决定采用值得相信的 libevent , 我很高兴发现有这么一个HTTP API。 新版已经有了一个evhttp_connection_set_local_address函数 。感觉很有希望.

这是采用libevent库用C编写的客户端：

1. #include <sys/types.h>
2. #include <sys/time.h>
3. #include <sys/queue.h>
4. #include <stdlib.h>
5. #include <err.h>
6. #include <event.h>
7. #include <evhttp.h>
8. #include <unistd.h>
9. #include <stdio.h>
10. #include <sys/socket.h>
11. #include <netinet/in.h>
12. #include <time.h>
13. #include <pthread.h>
14.  
15. #define BUFSIZE 4096
16. #define NUMCONNS 62000
17. #define SERVERADDR "10.103.1.43"
18. #define SERVERPORT 8000
19. #define SLEEP_MS 10
20.  
21. char buf[ BUFSIZE] ;
22.  
23. int bytes_recvd = 0 ;
24. int chunks_recvd = 0 ;
25. int closed = 0 ;
26. int connected = 0 ;
27.  
28. // called per chunk received
29. void chunkcb( struct evhttp_request * req, void * arg)
30. {
31.     int s = evbuffer_remove( req->input_buffer, &buf, BUFSIZE ) ;
32.     //printf("Read %d bytes: %s\n", s, &buf);
33.     bytes_recvd += s;
34.     chunks_recvd++;
35.     if ( connected >= NUMCONNS && chunks_recvd%10000 ==0 )
36.         printf ( ">Chunks: %d\t Bytes: %d\t Closed: %d\n " , chunks_recvd, bytes_recvd, closed) ;
37. }
38.  
39. // gets called when request completes
40. void reqcb( struct evhttp_request * req, void * arg)
41. {
42.     closed++;
43. }
44.  
45. int main( int argc, char **argv)
46. {
47.     event_init( ) ;
48.     struct evhttp *evhttp_connection;
49.     struct evhttp_request *evhttp_request;
50.     char addr[ 16 ] ;
51.     char path[ 32 ] ; // eg: "/test/123"
52.     int i,octet;
53.     for ( octet=1 ; octet<=17 ; octet++) {
54.         sprintf( &addr, "10.224.0.%d" , octet) ;
55.         for ( i=1 ;i<=NUMCONNS;i++) {
56.             evhttp_connection = evhttp_connection_new( SERVERADDR, SERVERPORT) ;
57.             evhttp_connection_set_local_address( evhttp_connection, &addr) ;
58.             evhttp_set_timeout( evhttp_connection, 864000 ) ; // 10 day timeout
59.             evhttp_request = evhttp_request_new( reqcb, NULL ) ;
60.             evhttp_request->chunk_cb = chunkcb;
61.             sprintf( &path, "/test/%d" , ++connected) ;
62.             if ( i%100 ==0 )  printf ( "Req: %s\t ->\t %s\n " , addr, &path) ;
63.             evhttp_make_request( evhttp_connection, evhttp_request, EVHTTP_REQ_GET, path ) ;
64.             evhttp_connection_set_timeout( evhttp_request->evcon, 864000 ) ;
65.             event_loop( EVLOOP_NONBLOCK ) ;
66.             if ( connected % 200 == 0 )
67.                 printf ( "\n Chunks: %d\t Bytes: %d\t Closed: %d\n " , chunks_recvd, bytes_recvd, closed) ;
68.             usleep( SLEEP_MS*1000 ) ;
69.         }
70.     }
71.     event_dispatch( ) ;
72.     return 0 ;
73. }

更多的参数用#define’s硬编码，这样你通过编辑源码来配置他然后重新编译。

编印运行：
$ gcc -o httpclient httpclient.c -levent
$ ./httpclient

这样还是不能打开多余64,500个端口 . 尽管他用了很少的内存。

尽管我指定了本地地址端口还是会超出限制, 临时端口无论是在内核或tcp栈上分配的都会超出2^16。 因此，为了能打开多于64,500个连接, 你需要指定本地地址和本地端口，相应的管理它们。不幸的libevent HTTP API没有选项指定本地端口。我 为 libevent打了补丁 加了一个合适的函数：
void evhttp_connection_set_local_port(struct evhttp_connection *evcon, u_short port); .

这相当不错； libevent编写的被很好, 文档也相当友好。

安装我修改过的libevent, 我可以在the set_local_address下添加如下代码:
evhttp_connection_set_local_port(evhttp_connection, 1024+i);

用他替换后, 从不同的地址的多个连接就能用同一端口号，指定本地地址。我重编译客户端让它运行一段时间以验证他能大多2^16限制。

Netstat验证 :
# netstat -n | awk '/^tcp/ {t[$NF]++}END{for(state in t){print state, t[state]}}’
TIME_WAIT 8
ESTABLISHED 118222

这显示多少端口在不同状态被打开。我们最后能够打开多于2^16个连接.

现在我们有了在一台机子上打开百万http连接的工具。它看起来每个连接消耗2KB内存, 加上内核占用的。是时候用它测试百万连接用户的我们的mochiweb comet服务器了。

C1024K测试-1 百万comet连接

为了这个测试我用了4台不同配置的服务器。这样的配置可能比测试用的高，但是它是有用的将来会用为产品，这能做一个很变态的测试. 所有这四个服务器都在同一个千兆局域网上，中间用了3个交换机和一个路由器。

一百万测试就像第一、二部分的10k测试，主要不同是更改了客户端，现在是用libevent,c编写，我在一个用了多台机子的正式的分布式erlang设置中运行的a。

服务器1 - 四核 2GHz CPU, 16GB 内存

• 启动订阅管理器
• 调入好友数据
• 启动路由器

服务器2 - 双通道四核 2.8GHz CPU, 32GB 内存

• 启动mochiweb应用

服务器3 - 四核 2GHz CPU, 16GB 内存

• 创建17个真实ip
• 安装打了补丁的libevent
• 运行客户端: ./httpclient 每秒建立100个连接直到1M

服务器4 - 双核 2GHz, 2GB内存

• 运行msggen程序, 向路由器发送大量的消息

在猛增到一百万连接期间我测量了mochiweb的内存用量，还有在剩下的时间里：

httpclient在每个连接见加了10ms延时,因此打开一百万连接用了将近3个小时。打开1M连接的mochiweb进程固定地内存用大约为25GB. 运行的服务器都由Ganglia监控, 它测量CPU, 网络和内存用量并且生成漂亮的图片：

你可以看到它需要大约38GB内存且开始了swap。我猜想这个不同是因为内核为保持打开的连接而基本被耗光的 。当我开始发送消息是就达到了定点。

消息有1000个进程产生，每个进程平均60ms一条消息，总共每秒大约16,666条消息：

erl> [ spawn( fun()->msggen:start(1000000, 10+random:uniform(100), 1000000) end) || I <- lists:seq(1,1000) ].

服务器 (server-4) 产生消息看起来如下图所示（Ganglia）:

每秒有10MB的消息发出 - 每秒16,666条消息. 典型的这些消息来自消息总线，应用服务器，或者已存在架构的一部分。

当我开始发送消息时，服务器1的负载(运行订阅管理器和路由器)一直低于1,CPU占用率从0增到5%。

服务器2的CPU (运行mochiweb应用, 有1M个连接) 增长的比较显著：

自然的，进程当要处理消息是不得不离开休眠状态,内存用量将轻微增加 . 没有消息且所有连接处于打开状态是内存用量的最好时候- 可想的,实际工作时需要更多内存。

从安全方面, mochiweb机器需要40GB内存一打开1M活跃comet连接。30GB用于mochiweb应用,剩下的10GB用于内核.换句话说,每个连接你需要分配40KB。

当用大量连接做各种测试时,我最终对我的sysctl.conf文件做了些修改. 只是部分试错,我真的不知道更多以做出明智的决定关于那个值需要修改的内部原因 . 我的策略是等待问题发生，检测 /var/log/kern.log 看什么神秘的错误被报告, 然后添加听起来很合理的数据. 这是上面测试使用的设置信息：

net.core.rmem_max = 33554432
net.core.wmem_max = 33554432
net.ipv4.tcp_rmem = 4096 16384 33554432
net.ipv4.tcp_wmem = 4096 16384 33554432
net.ipv4.tcp_mem = 786432 1048576 26777216
net.ipv4.tcp_max_tw_buckets = 360000
net.core.netdev_max_backlog = 2500
vm.min_free_kbytes = 65536
vm.swappiness = 0
net.ipv4.ip_local_port_range = 1024 65535

我将很喜欢学习更多关于linux tcp调优的知识这样我能对这些设置做更明智的决策. 这些可与确定不是优化的, 但是最少它们足够应付1M的连接. 这些更改运行在一个64bit的elang虚拟机上, 字长是8bytes而不是4, 可能可以解释为什么内存用量比我在第二部分做c10k测试时高的多。

一个用Libevent实现的Erlang C-Node

在为libevent加入HTTP api后, 它看起来完全合理做1M连接测试相对于用c写的http服务器，因此我们有了比较的基础。

我猜打开内核的poll模型意味这erlang虚拟机能够用epol(或类似的),但是即使是这样显然也需要解决负载问题，我们通过委派连接处理给用libevent实现的c程序或许能减轻负载. 我想重用更多的Erlang代码, 因此让我们尽可能少的用c - 只是在连接处理和HTTP部分。 我也寻找了试用Erlang C 接口的一个理由,因此下面的程序组合了两者。他是一个用C和libevent写的comet http服务器用用整数id标志用户(向我们的mochiweb应用), 且扮演一个Erlang C节点.

它连接一个指定的erlang节点, 监听像{123, <<"Hello user 123">>}的消息然后向用户 123分派 “Hello user 123″ , 假如已连接. 那些没有连接用户的消息被丢弃,就像前面的例子。

httpdcnode.c

1. #include <sys/types.h>
2. #include <sys/time.h>
3. #include <sys/queue.h>
4. #include <stdlib.h>
5. #include <err.h>
6. #include <event.h>
7. #include <evhttp.h>
8. #include <stdio.h>
9. #include <sys/socket.h>
10. #include <netinet/in.h>
11.  
12. #include "erl_interface.h"
13. #include "ei.h"
14.  
15. #include <pthread.h>
16.  
17. #define BUFSIZE 1024
18. #define MAXUSERS (17*65536) // C1024K
19.  
20. // List of current http requests by uid:
21. struct evhttp_request * clients[ MAXUSERS+1 ] ;
22. // Memory to store uids passed to the cleanup callback:
23. int slots[ MAXUSERS+1 ] ;
24.  
25. // called when user disconnects
26. void cleanup( struct evhttp_connection *evcon, void *arg)
27. {
28.     int *uidp = ( int *) arg;
29.     fprintf( stderr, "disconnected uid %d\n " , *uidp) ;
30.     clients[ *uidp] = NULL ;
31. }
32.  
33. // handles http connections, sets them up for chunked transfer,
34. // extracts the user id and registers in the global connection table,
35. // also sends a welcome chunk.
36. void request_handler( struct evhttp_request *req, void *arg)
37. {
38.         struct evbuffer *buf;
39.         buf = evbuffer_new( ) ;
40.         if ( buf == NULL ) {
41.             err( 1 , "failed to create response buffer" ) ;
42.         }
43.  
44.         evhttp_add_header( req->output_headers, "Content-Type" , "text/html; charset=utf-8" ) ;
45.  
46.         int uid = -1 ;
47.         if ( strncmp( evhttp_request_uri( req) , "/test/" , 6 ) == 0 ) {
48.             uid = atoi( 6 +evhttp_request_uri( req) ) ;
49.         }
50.  
51.         if ( uid <= 0 ) {
52.             evbuffer_add_printf( buf, "User id not found, try /test/123 instead" ) ;
53.             evhttp_send_reply( req, HTTP_NOTFOUND, "Not Found" , buf) ;
54.             evbuffer_free( buf) ;
55.             return ;
56.         }
57.  
58.         if ( uid > MAXUSERS) {
59.             evbuffer_add_printf( buf, "Max uid allowed is %d" , MAXUSERS) ;
60.             evhttp_send_reply( req, HTTP_SERVUNAVAIL, "We ran out of numbers" , buf) ;
61.             evbuffer_free( buf) ;
62.             return ;
63.         }
64.  
65.         evhttp_send_reply_start( req, HTTP_OK, "OK" ) ;
66.         // Send welcome chunk:
67.         evbuffer_add_printf( buf, "Welcome, Url: ‘%s’ Id: %d\n " , evhttp_request_uri( req) , uid) ;
68.         evhttp_send_reply_chunk( req, buf) ;
69.         evbuffer_free( buf) ;
70.  
71.         // put reference into global uid->connection table:
72.         clients[ uid] = req;
73.         // set close callback
74.         evhttp_connection_set_closecb( req->evcon, cleanup, &slots[ uid] ) ;
75. }
76.  
77.  
78. // runs in a thread - the erlang c-node stuff
79. // expects msgs like {uid, msg} and sends a a ‘msg’ chunk to uid if connected
80. void cnode_run( )
81. {
82.     int fd;                                  /* fd to Erlang node */
83.     int got;                                 /* Result of receive */
84.     unsigned char buf[ BUFSIZE] ;              /* Buffer for incoming message */
85.     ErlMessage emsg;                         /* Incoming message */
86.  
87.     ETERM *uid, *msg;
88.  
89.     erl_init( NULL , 0 ) ;
90.  
91.     if ( erl_connect_init( 1 , "secretcookie" , 0 ) == -1 )
92.         erl_err_quit( "erl_connect_init" ) ;
93.  
94.     if ( ( fd = erl_connect( "httpdmaster@localhost" ) ) < 0 )
95.         erl_err_quit( "erl_connect" ) ;
96.  
97.     fprintf( stderr, "Connected to httpdmaster@localhost\n \r " ) ;
98.  
99.     struct evbuffer *evbuf;
100.  
101.     while ( 1 ) {
102.         got = erl_receive_msg( fd, buf, BUFSIZE, &emsg) ;
103.         if ( got == ERL_TICK) {
104.             continue ;
105.         } else if ( got == ERL_ERROR) {
106.             fprintf( stderr, "ERL_ERROR from erl_receive_msg.\n " ) ;
107.             break ;
108.         } else {
109.             if ( emsg.type == ERL_REG_SEND) {
110.                 // get uid and body data from eg: {123, <<"Hello">>}
111.                 uid = erl_element( 1 , emsg.msg ) ;
112.                 msg = erl_element( 2 , emsg.msg ) ;
113.                 int userid = ERL_INT_VALUE( uid) ;
114.                 char *body = ( char *) ERL_BIN_PTR( msg) ;
115.                 int body_len = ERL_BIN_SIZE( msg) ;
116.                 // Is this userid connected?
117.                 if ( clients[ userid] ) {
118.                     fprintf( stderr, "Sending %d bytes to uid %d\n " , body_len, userid) ;                
119.                     evbuf = evbuffer_new( ) ;
120.                     evbuffer_add( evbuf, ( const void *) body, ( size_t) body_len) ;
121.                     evhttp_send_reply_chunk( clients[ userid] , evbuf) ;
122.                     evbuffer_free( evbuf) ;
123.                 } else {
124.                     fprintf( stderr, "Discarding %d bytes to uid %d - user not connected\n " ,
125.                             body_len, userid) ;                
126.                     // noop
127.                 }
128.                 erl_free_term( emsg.msg ) ;
129.                 erl_free_term( uid) ;
130.                 erl_free_term( msg) ;
131.             }
132.         }
133.     }
134.     // if we got here, erlang connection died.
135.     // this thread is supposed to run forever

评论

2 楼 suifeng 2008-12-30  

看了你的文章, 明白了什么叫专业
  

1 楼 idisc 2008-12-04  

由于长度限制，还得再续，呵呵