惊群 java_(转)测试Lighttpd accept的惊群现象

lighttpd里面采用的是prefork的模型，在fork进程之前就已经创建好了listen socket

那么fork了进程池之后，所有进程都有一份自己独立的listen socket fd，

但实际上这个独立的fd 对应的确是一个文件表项，即实际上任然是一个共享的文件描述符

在阻塞模型中，各进程分别通过accept阻塞，等待连接到达，当一个连接到达时，所有的进程都会被唤醒，但只有其中一个进程可以成功accept该连接，其余的则继续投入睡眠，这就是所谓的惊群现象

lighttpd使用的是非阻塞IO复用模型，测试一下是否会有惊群现象呢？

先把结论给出：

1.比如有20个进程注册了listen socket的请求连接事件，当一个连接到达确实会有多个进程 被通知有事件要处理(但不是全部，大约只有5，6个进程)

2.被唤醒的这几个进程会调用accept函数,其中只有一个成功返回连接fd,其余进程均返回EAGAIN或者 EWOULDBLOCK错误(因为是非阻塞的)

测试方法，自己写了一个prefork进程 + epoll的非阻塞server，启动20个进程，client telnet，打印服务器日志

try to accept new connection,pid=29879

try to accept new connection,pid=29876

try to accept new connection,pid=29880

process 29879 accept connection

accept EAGAIN error pid=29876

try to accept new connection,pid=29875

accept EAGAIN error pid=29880

accept EAGAIN error pid=29875

四个进程被通知有事件处理，1个成功accept,3个返回EAGAIN

在lighttpd中，server当被通知有连接要处理时，server会通过循环执行

accept，直到返回错误，或者超过一个上限值

这样，当海量请求连接到达时，似乎惊群不会带来太多的性能损耗。

[cpp]

enum conn_states {

conn_listening, /** the socket which listens for connections */

conn_read, /** reading in a command line */

conn_write, /** writing out a simple response */

conn_nread, /** reading in a fixed number of bytes */

conn_swallow, /** swallowing unnecessary bytes w/o storing */

conn_closing, /** closing this connection */

conn_mwrite, /** writing out many items sequentially */

};

typedef struct{

int fd;

int state;

}conn;

[/cpp]

[cpp]

#include

#include

#include

#include

#include

#include

#include

#include

#include

#include "event.h"

#include "base.h"

//forward declaration

static fdevents *ev;

static conn **conns;

static int freetotal;

static int freecurr;

static int create_listen_fd(char *addr,int port);

static int conn_init();

static conn *get_conn_from_freelist();

static int add_conn_to_freelist(conn *c);

conn *conn_new(int fd,int state);

static int conn_init(){

freetotal=200;

freecurr=0;

conns=(conn **)malloc(freetotal * sizeof(*conns));

if(!conns){

return -1;

}

return 0;

}

static conn *get_conn_from_freelist(){

conn *con;

if(freecurr > 0){

con=conns[–freecurr];

conns[freecurr]=NULL;

return con;

}

return NULL;

}

static int add_conn_to_freelist(conn *c){

if(freecurr

conns[freecurr++]=c;

return 0;

}else{

conn **new_conns=(conn **)realloc(conns,sizeof(*new_conns)*2*freetotal);

if(new_conns){

freetotal*=2;

conns=new_conns;

conns[freecurr++]=c;

return 0;

}

}

return -1;

}

conn *conn_new(int fd,int state){

conn *c;

c=get_conn_from_freelist();

if(!c){

c=(conn *)malloc(sizeof(*c));

}

c->fd=fd;

c->state=state;

return c;

}

static int create_listen_fd(char *addr,int port){

int fd,val,flags;

struct sockaddr_in sockaddr;

fd=socket(AF_INET,SOCK_STREAM,0);

if(fd==-1){

fprintf(stderr,"socket()\n");

return -1;

}

val=1;

if(setsockopt(fd,SOL_SOCKET,SO_REUSEADDR,&val,sizeof(val))<0){

fprintf(stderr,"reuseaddr\n");

return -1;

}

if((flags=fcntl(fd,F_GETFL,0)<0) || fcntl(fd,F_SETFL,flags | O_NONBLOCK) < 0){

fprintf(stderr,"nonblocking\n");

return -1;

}

bzero(&sockaddr,sizeof(sockaddr));

sockaddr.sin_family=AF_INET;

sockaddr.sin_port=htons(port);

inet_pton(AF_INET,addr,&sockaddr.sin_addr);

if(bind(fd,(struct sockaddr *)&sockaddr,sizeof(sockaddr))<0){

fprintf(stderr,"bind error %s",strerror(errno));

return -1;

}

if(listen(fd,2048)<0){

fprintf(stderr,"listen %s",strerror(errno));

return -1;

}

return fd;

}

void event_handler(int fd,void *ctx, int revents){

struct sockaddr_in addr;

socklen_t sock_len;

int done=0,connfd;

conn *c;

c=(conn *)ctx;

while(!done){

switch(c->state){

case conn_listening:

printf("try to accept new connection,pid=%d\n",getpid());

sock_len=sizeof(addr);

connfd=accept(fd,(struct sockaddr *)&addr,&sock_len);

if(connfd>0){

printf("process %d accept connection\n",getpid());

c = conn_new(connfd,conn_read);

fdevent_register(ev,connfd,event_handler,c);

fdevent_event_add(ev,connfd,FDEVENT_IN);

}else{

if(errno== EAGAIN || errno == EWOULDBLOCK){

printf("accept EAGAIN error pid=%d\n",getpid());

}

if(errno==EINTR){

printf("accept EINTR error pid=%d\n",getpid());

}

if(errno==ECONNABORTED){ /* this is a FreeBSD thingy */

printf("accept EABORTED error pid=%d\n",getpid());

}

if(errno==EMFILE){

printf("accept EMFILE error pid=%d\n",getpid());

}

}

done=1;

break;

case conn_read:

printf("on read");

break;

}

}

}

int main(int argc,char **argv){

int fd,o;

char *listen_addr;

int port,num_childs,max_fds;

struct rlimit rlim;

conn *c;

port=0;

num_childs=5;

while(-1!=(o=getopt(argc,argv,"l:p:f:h"))){

switch(o){

case ‘l’:

listen_addr=strdup(optarg);

break;

case ‘p’:

port=atoi(optarg);

break;

case ‘f’:

num_childs=atoi(optarg);

break;

case ‘h’:

printf("Usage -l listen addr\n");

printf("Usage -p listen port \n");

printf("Usage -f fork num\n");

exit(1);

}

}

if(!listen_addr){

listen_addr=strdup("127.0.0.1");

}

if(!port){

printf("port is unknown\n");

exit(1);

}

if(0 != getrlimit(RLIMIT_NOFILE,&rlim)){

fprintf(stderr,"getrlimit failed.reason %s\n",strerror(errno));

exit(1);

}

max_fds=rlim.rlim_cur;

//create listen socket

if(-1==(fd=create_listen_fd(listen_addr,port))){

fprintf(stderr,"create listen fd failed\n");

exit(1);

}

//prefork child

if(num_childs > 0){

int child=0;

while(!child){

if(num_childs >0){

switch(fork()){

case -1:

return -1;

case 0:

child=1;

break;

default:

num_childs–;

break;

}

}else{

int status;

if(-1 !=wait(&status)){

num_childs++;

}else{

//ignore

}

}

}

}

//child process event

conn_init();

c=conn_new(fd,conn_listening);

ev=fdevent_init(max_fds);

if(!ev){

fprintf(stderr,"fdevent_init()\n");

exit(1);

}

fdevent_register(ev,fd,event_handler,c);

fdevent_event_add(ev,fd,FDEVENT_IN);

fdevent_poll(ev,1000);

}

[/cpp]