|
{ if ( !ppool ) { PRINTINFO( LEVEL4, "ppool is null!!!" ); return; } init_signal( ppool ); ppool->epollfd = Epoll_create( 5 ); PRINTINFO( LEVEL0, "child cur process:%d", ppool->cur_process_num ); int pipefd = ppool->sub_process[ppool->index].pipefd[1]; Epoll_add_fd( ppool->epollfd, pipefd ); Epoll_add_fd( ppool->epollfd, sig_pipefd[1] ); struct epoll_event *events = (struct epoll_event *) calloc( sizeof(struct epoll_event), ppool->max_epoll_event ); if ( !events ) { PRINTINFO( LEVEL5, "calloc error!!!" ); goto _free_source; } struct client_param *cparam = NULL; cparam = (struct client_param *) calloc( sizeof(struct client_param), ppool->max_user_num ); if ( !cparam ) { PRINTINFO( LEVEL5, "calloc error!!!" ); goto _free_source; } int event_num, event_fd; int i, j, ret, onebyte; while ( !ppool->stop ) { event_num = Epoll_wait( ppool->epollfd, events, ppool->max_epoll_event, -1); //PRINTINFO( LEVEL0, "child event num:%d", event_num ); if ( (event_num < 0) && (errno != EINTR) ) { PRINTINFO_ERR( LEVEL4, "Epoll_wait error!!!" ); ppool->stop = 1; break; } for ( i = 0; i < event_num; i++ ) { event_fd = events[i].data.fd; //parent process notify that there is a new client connect to. if ( event_fd == pipefd && events[i].events & EPOLLIN ) { PRINTINFO( LEVEL0, "receive signal from parent there is a new client connection" ); ret = recv( event_fd, (char *)&onebyte, 1, 0 ); if ( ret <= 0 ) { continue; } else { struct sockaddr_in clientaddr; socklen_t addrlen = sizeof(clientaddr); bzero( &clientaddr, addrlen ); int connfd = accept( ppool->listenfd, (struct sockaddr *)&clientaddr, &addrlen ); if ( connfd < 0 ) { PRINTINFO_ERR( LEVEL3, "accept a new client error!!!" ); continue; } PRINTINFO( LEVEL0, "one client conntect(fd:%d)", connfd ); Epoll_add_fd( ppool->epollfd, connfd ); client_param_init( &cparam[connfd], connfd, &clientaddr ); } } //process catch a signal else if ( event_fd == sig_pipefd[1] && events[i].events & EPOLLIN ) { int sig; char signals[1024] = {0}; ret = recv( sig_pipefd[1], signals, sizeof(signals), 0 ); if ( ret <= 0 ) { continue; } client_signal_handle( ppool, signals, ret ); } //client socket fd has readable event,maybe a //request else if ( events[i].events & EPOLLIN ) { client_handle( &cparam[event_fd] ); } else { continue; } } } _free_source: free( events ); events = NULL; free( cparam ); cparam = NULL; close( pipefd ); close( ppool->epollfd ); } static int parent_signal_handle( processpool *ppool, char *signals, int signals_num ) { int i = 0, j = 0; for ( ; i < signals_num; i++ ) { switch( signals[i] ) { case SIGCHLD: { PRINTINFO( LEVEL0, "parent receive SIGCHLD signal" ); pid_t pid; int stat; while ( (pid = waitpid(-1, &stat, WNOHANG)) > 0 ) { PRINTINFO( LEVEL0, "parent receive SIGCHLD signal(pid:%d)", pid ); for ( j = 0; j < ppool->cur_process_num; j++ ) { if ( ppool->sub_process[j].pid == pid ) { PRINTINFO( LEVEL3, "child process(%d) exit.", pid ); close( ppool->sub_process[j].pipefd[1] ); ppool->sub_process[j].pid = -1; } } } ppool->stop = 1; for ( j = 0; j < ppool->cur_process_num; j++ ) { PRINTINFO( LEVEL0, "pid:%d", ppool->sub_process[j].pid ); if ( ppool->sub_process[j].pid != -1 ) { ppool->stop = 0; break; } } break; } case SIGTERM: case SIGINT: { PRINTINFO( LEVEL2, "recv SIGINT/SIGTERM, kill all child process now." ); //PRINTINFO( LEVEL0, "cur_process_num:%d", ppool->cur_process_num ); for ( i = 0; i < ppool->cur_process_num; i++ ) { int pid = ppool->sub_process[i].pid; if ( pid != -1 ) { PRINTINFO( LEVEL0, "kill process:%d", pid ); ppool->sub_process[i].pid = -1; kill( pid, SIGTERM ); } } ppool->stop = 1; break; } default: { break; } } } } static void run_parent( processpool *ppool ) { if ( !ppool ) { PRINTINFO( LEVEL4, "ppool is null!!!" ); return; } init_signal( ppool ); ppool->epollfd = Epoll_create( 5 ); Epoll_add_fd( ppool->epollfd, ppool->listenfd ); Epoll_add_fd( ppool->epollfd, sig_pipefd[1] ); struct epoll_event *events = NULL; events = (struct epoll_event *) calloc( sizeof(struct epoll_event), ppool->max_epoll_event ); if ( !events ) { PRINTINFO( LEVEL5, "calloc error!!!" ); char sig = SIGINT; parent_signal_handle( ppool, &sig, 1 ); goto _free_source; } int event_num; int i, onebyte = 1, ret, j; //p_idx specifies current dispatched child process. //roll_index specifies the next child process. int p_idx , roll_index = 0; //PRINTINFO( LEVEL0, "epollfd:%d", ppool->epollfd ); while ( !ppool->stop ) { //Specifying a timeout of -1 causes epoll_wait() //to block indefinitely. event_num = Epoll_wait( ppool->epollfd, events, ppool->max_epoll_event, -1 ); if ( event_num < 0 && errno != EINTR ) { PRINTINFO_ERR( LEVEL5, "Epoll_wait error!!!" ); ppool->stop = 1; break; } //PRINTINFO( LEVEL0, "event_num:%d", event_num ); for ( i = 0; i < event_num; i++ ) { int event_fd = events[i].data.fd; //listenfd,there is a new client connection. //notify child process to accept if ( event_fd == ppool->listenfd ) { PRINTINFO( LEVEL0, "event:parent listenfd" ); //round robin dispatch //easily roll polling p_idx = ro |