c++ libevent 多进程socket惊群怎么处理？ - 知否问答 - 万象云+社区

struct event_base * base = NULL; pthread_mutex_t *g_lock=NULL; //多进程锁 int listen_fd = -1;//多进程监听socket struct event ev_accept; ...... int main(int argc,char** argv){ if(!init_lock()){ err(1,"init lock failed"); return -1; } if(!init_listener()){ err(1,"init listener failed"); return -1; } pid_t pid =1; int fork_i=0; for(;fork_i<FORK_NUM;fork_i++){ pid = fork(); if(-1 == pid || 0 == pid){ break; } } if(-1 == pid){ err(1,"this children process: %d fail to fork",fork_i); exit(1); } else if(0 == pid){ //子进程 local_worker(fork_i); exit(0); } else{ //主进程 manager(); exit(0); } return 0; } bool init_lock(){ g_lock = (pthread_mutex_t*)mmap(NULL,sizeof(pthread_mutex_t),PROT_READ|PROT_WRITE,MAP_SHARED|MAP_ANON,-1,0); pthread_mutexattr_t attr; pthread_mutexattr_init(&attr); pthread_mutexattr_setpshared(&attr,PTHREAD_PROCESS_SHARED); pthread_mutex_init(g_lock,&attr); if(g_lock == MAP_FAILED){ return false; } return true; } bool try_lock(){ if(0 == pthread_mutex_trylock(g_lock)){ return true; } else{ return false; } } void unlock(){ pthread_mutex_unlock(g_lock); } void* local_worker(int worker_id,pid_t pid){ ...... struct event_base * base = p_work_info->base = event_base_new(); /*初始化监听客户端连接事件*/ event_set(&ev_accept,listen_fd,EV_READ|EV_PERSIST,on_accept,NULL); event_base_set(base,&ev_accept); event_add(&ev_accept,NULL); ...... event_base_dispatch(base); } void on_accept(int fd, short ev, void *arg){ if(!try_lock()){ return; } int client_fd; struct sockaddr_in client_addr; socklen_t client_len = sizeof(client_addr); client_fd = accept(fd,(struct sockaddr*)&client_addr,&client_len); unlock(); warn("~~~~~~~~~~~"); warn("%d accept",getpid()); if(client_fd < 0){ warn("accept failed"); return; } ...... }

...... client disconnected #客户端断开连接时打印消息，之后客户端重新发起连接请求。 libevent_worker: ~~~~~~~~~~~: Success libevent_worker: 12486 accept: Success libevent_worker: ~~~~~~~~~~~: Resource temporarily unavailable libevent_worker: 12485 accept: Resource temporarily unavailable libevent_worker: accept failed: Resource temporarily unavailable client disconnected libevent_worker: ~~~~~~~~~~~: Success libevent_worker: 12486 accept: Success ......

//libevent server sample on linux /*socket*/ #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> /**/ #include <iostream> #include <err.h> //err() #include <string.h> //memset() #include <fcntl.h> //fcntl() #include <pthread.h> // #include <cstdlib> //calloc() #include <unistd.h> //fork(),close() #include <sys/queue.h> //链表 #include <errno.h> //errno,ENTIR #include <stdio.h> #include <vector> #include <string> #include <sys/mman.h>//mmap #include <sys/wait.h> /*libevent*/ #include <event.h> /*内存泄露*/ #include <mcheck.h> /**/ #include "manage_protocol.h" using namespace std; //#define CONTROLLEN CMSG_LEN(sizeof(int)) #define SERVER_PORT 55541 #define LISTEN_NUM 32 #define FORK_NUM 4 #define BUF_LEN 1024 #define WRITE_BUF_LEN 1024*4 struct event_accepted{ int fd;//socket fd int worker_id; //记录所连客户端socket所属的工作者编号 struct event* ev_read; struct event* ev_write; char* write_buf; int len; //write_buf的长度 int offset; //write_buf已写入后的偏移 event_accepted(){ ev_read = NULL; ev_write = NULL; write_buf = NULL; worker_id = -1; fd = -1; len = 0; offset = 0; } }; struct socket_pair{ int connecter; //管理者监听工作者信息的socket int accepter; //工作者监听管理者消息的socket socket_pair(){ connecter =-1; accepter =-1; } }; /*工作者结构*/ struct worker_info{ int worker_id; //工作者编号 //pthread_mutex_t g_clock; //线程锁 struct socket_pair socket_pairs; //管理者与工作者通讯的事件socket对。 struct event socket_pair_read_event; //用于监听socket pair的事件(多进程中，管理者与工作者该事件对应的socket不同) struct event socket_pair_write_event; //用于通知socket pair对端的事件(多进程中，管理者与工作者该事件对应的socket不同) struct event_base *base; //工作者的event_base实例(多进程中，管理者与工作者event_base不同) char* write_buf; int len; //write_buf的长度 int offset; //write_buf已写入后的偏移 pid_t pid;//记录子进程pid // TAILQ_ENTRY(event_accepted) entries;//工作者上注册的event的链表，用于回收event分配的内存 worker_info(){ worker_id = -1; write_buf = NULL; len = 0; offset = 0; } }; // vector<worker_info> v_global_workers; struct event_base * base = NULL; pthread_mutex_t *g_lock=NULL; //多进程锁 int listen_fd = -1;//多进程监听socket struct event ev_accept;//多进程连接事件 struct event signal_child; int setnonblock(int fd); void on_accept(int fd, short ev, void *arg); void worker_read(int fd, short ev, void *arg); void worker_write(int fd, short ev, void *arg); void* local_worker(int worker_id, pid_t pid = -1); int socketpair_init(); //初始化本地socket连接用于通知本地工作者注册事件 void socketpair_worker_read(int fd, short ev, void *arg);//本地工作者接收管理者命令 void socketpair_worker_write(int fd, short ev, void *arg);//本地工作者向管理者发送状态等 void socketpair_manager_read(int fd, short ev, void *arg); void socketpair_manager_write(int fd, short ev, void *arg); void proc_socketpair_worker_read(int fd, char* data, int len);// void proc_socketpair_manager_read(worker_info *worker,int fd, char* data, int len);// void destroy(); void manager();//管理者监听连接事件(在管理者注册完所有与工作者相关事件后进行) bool init_lock(); bool try_lock(); void unlock(); bool init_listener(); void signal_child_cb(int fd,short ev,void *arg); int main(int argc,char** argv){ setenv("MALLOC_TRACE","mtrace.out",1); mtrace(); //初始化本地工作者注册通知socket if(socketpair_init() <0){ err(1,"init socketpair_init failed"); } if(!init_lock()){ err(1,"init lock failed"); return -1; } if(!init_listener()){ err(1,"init listener failed"); return -1; } pid_t pid =1; int fork_i=0; for(;fork_i<FORK_NUM;fork_i++){ //v_global_workers[fork_i].worker_id=fork_i; pid = fork(); if(-1 == pid || 0 == pid){ v_global_workers[fork_i].pid=getpid(); break; } v_global_workers[fork_i].pid=pid; } if(-1 == pid){ err(1,"this children process: %d fail to fork",fork_i); exit(1); } else if(0 == pid){ //子进程 local_worker(fork_i); exit(0); } else{ //主进程 manager(); exit(0); } return 0; } bool init_lock(){ g_lock = (pthread_mutex_t*)mmap(NULL,sizeof(pthread_mutex_t),PROT_READ|PROT_WRITE,MAP_SHARED|MAP_ANON,-1,0); pthread_mutexattr_t attr; pthread_mutexattr_init(&attr); pthread_mutexattr_setpshared(&attr,PTHREAD_PROCESS_SHARED); pthread_mutex_init(g_lock,&attr); if(g_lock == MAP_FAILED){ return false; } return true; } bool try_lock(){ if(0 == pthread_mutex_trylock(g_lock)){ return true; } else{ return false; } } void unlock(){ pthread_mutex_unlock(g_lock); } bool init_listener(){ /*初始化监听socket*/ struct sockaddr_in listen_addr; int reuseaddr_on =1; listen_fd =socket(AF_INET,SOCK_STREAM,0); if(listen_fd<0){ err(1,"listen failed"); return false; } if(setsockopt(listen_fd,SOL_SOCKET,SO_REUSEADDR,&reuseaddr_on,sizeof(reuseaddr_on)) <0){ err(1,"setsockopt failed"); return false; } memset(&listen_addr,0,sizeof(listen_addr)); listen_addr.sin_family = AF_INET; listen_addr.sin_addr.s_addr = INADDR_ANY; listen_addr.sin_port =htons(SERVER_PORT); if(bind(listen_fd,(struct sockaddr*)&listen_addr,sizeof(listen_addr))<0){ err(1,"bind failed"); return false; } if(listen(listen_fd,LISTEN_NUM)<0){ err(1,"listen failed"); return false; } if(setnonblock(listen_fd)<0){ err(1,"set server socket to non-blocking failed"); return false; } return true; } void manager(){ base = event_init(); for(int i=0;i<v_global_workers.size();i++){ /*初始化管理者写事件内存*/ v_global_workers[i].base = base; v_global_workers[i].write_buf = new char[WRITE_BUF_LEN]; v_global_workers[i].len = 0; v_global_workers[i].offset = 0; /*初始化管理者读写事件并注册读事件（写事件发送时再注册）*/ event_set( &v_global_workers[i].socket_pair_read_event, v_global_workers[i].socket_pairs.connecter, EV_READ|EV_PERSIST, socketpair_manager_read, &v_global_workers[i]); event_add(&v_global_workers[i].socket_pair_read_event,NULL); event_set( &v_global_workers[i].socket_pair_write_event, v_global_workers[i].socket_pairs.connecter, EV_WRITE, socketpair_manager_write, &v_global_workers[i]); } // /*初始化监听客户端连接事件*/ // event_set(&ev_accept,listen_fd,EV_READ|EV_PERSIST,on_accept,NULL); // event_add(&ev_accept,NULL); //注册监听子进程退出信号事件 event_set(&signal_child,SIGCHLD,EV_SIGNAL|EV_PERSIST,signal_child_cb,&signal_child); event_add(&signal_child,NULL); event_dispatch(); } void on_accept(int fd, short ev, void *arg){ if(!try_lock()){ //warn("%d:lock failed",getpid()); return; } //warn("%d:lock success",getpid()); int worker_id =0; int client_fd; struct sockaddr_in client_addr; socklen_t client_len = sizeof(client_addr); struct worker_info *p_work_info; struct event_accepted *client; client_fd = accept(fd,(struct sockaddr*)&client_addr,&client_len); unlock(); warn("~~~~~~~~~~~"); warn("%d accept",getpid()); if(client_fd < 0){ warn("accept failed"); return; } p_work_info = &v_global_workers[0]; client = (struct event_accepted*)calloc(1,sizeof(event_accepted)); if(NULL == client){ err(1,"on_accept malloc event_accepted failed"); } client->fd = client_fd; client->ev_write = new event; client->ev_read = new event; client->write_buf = new (nothrow) char[WRITE_BUF_LEN]; client->len = 0; client->offset = 0; if(client->ev_read == NULL || NULL == client->ev_write || NULL == client->write_buf){ if(!client->ev_read){ delete client->ev_read; } if(!client->ev_write){ delete client->ev_write; } if(!client->write_buf){ delete[] client->write_buf; } free(client); err(1,"alloc read event failed"); } if(setnonblock(client_fd)<0){ warn("failed to set client socket non-blocking"); close(client_fd); } //初始化事件 event_set(client->ev_read,client_fd,EV_READ|EV_PERSIST,worker_read,client); event_base_set(p_work_info->base,client->ev_read); event_set(client->ev_write,client_fd,EV_WRITE,worker_write,client); event_base_set(p_work_info->base,client->ev_write); //工作者注册读事件，写事件在需要时再注册 event_add(client->ev_read,NULL); return; } /*初始化本地socket连接用于通知本地工作者注册事件*/ int socketpair_init(){ for(int i=0;i<FORK_NUM;i++){ struct worker_info temp_worker; temp_worker.worker_id = i; //记录本地工作进程编号 //socket_pairs int fd_1[2]; int status_1 = socketpair(AF_UNIX, SOCK_STREAM, 0, fd_1); if (status_1 < 0) { printf("Call socketpair error, errno is %d\n", errno); goto fail; } temp_worker.socket_pairs.connecter=fd_1[0]; temp_worker.socket_pairs.accepter= fd_1[1]; // if(setnonblock(temp_worker.socket_pairs.accepter)<0){ warn("failed to set client socket non-blocking"); } if(setnonblock(temp_worker.socket_pairs.connecter)<0){ warn("failed to set client socket non-blocking"); } v_global_workers.push_back(temp_worker); } return 0; fail: warn("socket pair init failed"); while(v_global_workers.empty()){ worker_info temp = v_global_workers.back(); v_global_workers.pop_back(); if(temp.socket_pairs.connecter >0){ close(temp.socket_pairs.connecter); } if(temp.socket_pairs.accepter >0){ close(temp.socket_pairs.accepter); } } return -1; } int setnonblock(int fd) { int flags; flags = fcntl(fd, F_GETFL); if (flags < 0) return flags; flags |= O_NONBLOCK; if (fcntl(fd, F_SETFL, flags) < 0) return -1; return 0; } void socketpair_manager_read(int fd, short ev, void *arg){ int buf_size = 1024; char buf[buf_size]; memset(buf,0,buf_size); uint16_t data_len = 0; int recv_len = 0; while(1){ int size =read(fd,buf+recv_len,buf_size-recv_len);//recv(fd,buf+recv_len,buf_size-recv_len,0); if(0 == size){ //worker与管理者断开连接 cout<<"worker disconnected"<<endl; worker_info *worker = (worker_info*)arg; vector<worker_info>::iterator it=v_global_workers.begin(); for(;it!=v_global_workers.end();it++){ if(fd == (*it).socket_pairs.connecter){ break; } } event_del(&worker->socket_pair_read_event); event_del(&worker->socket_pair_write_event); delete[] worker->write_buf; close(worker->socket_pairs.connecter); v_global_workers.erase(it); return; } else if(size < 0){ // cout<<"socekt failure,disconnecting worker:"<<strerror(errno)<<endl; worker_info *worker = (worker_info*)arg; vector<worker_info>::iterator it=v_global_workers.begin(); for(;it!=v_global_workers.end();it++){ if(fd == (*it).socket_pairs.connecter){ break; } } event_del(&worker->socket_pair_read_event); event_del(&worker->socket_pair_write_event); delete[] worker->write_buf; close(worker->socket_pairs.connecter); v_global_workers.erase(it); return; } else{ recv_len += size; } while(recv_len >2){ if(0x0b != buf[0] || 0x0d != buf[1]){ recv_len--; memmove(buf,buf+1,recv_len); continue; } if(recv_len > 4){ data_len = *(uint16_t*)(buf+2); if(recv_len < data_len +4){ //一条信息未读取完全，继续读取 break; } proc_socketpair_manager_read((worker_info*)arg,fd, buf+4, data_len); recv_len = recv_len - 4 - data_len; memmove(buf,buf+4+data_len,recv_len); } else{ break; } } //退出 if(0 == recv_len){ break; } } } void proc_socketpair_manager_read(worker_info *worker,int fd, char* data, int len){ char protocol = *data; string log_str=""; switch(protocol){ default: break; } } void socketpair_manager_write(int fd, short ev, void *arg){ struct worker_info* client = (worker_info*)arg; int len =0; len = write(fd,client->write_buf+client->offset,client->len-client->offset); if(len == -1){ //写操作被信号打断或不能写入，重新注册写事件 if(errno == EINTR || errno == EAGAIN){ event_add(&(client->socket_pair_write_event),NULL); return; } else{ err(1,"write error"); } } else if((client->offset + len) <client->len){ //数据没有完全写入 int offset = client->offset +len; memmove(client->write_buf,client->write_buf+offset,client->len-offset); client->offset = 0; client->len -= offset; event_add(&(client->socket_pair_write_event),NULL); return; } else{ //写入完成 client->offset = client->len = 0; } } void* local_worker(int worker_id,pid_t pid){ //重启子进程时退出父进程的事件循环 if(NULL != base){ event_base_loopexit(base,NULL); } vector<worker_info>::iterator it=v_global_workers.begin(); if(pid>0){ //子进程退出，重新fork for(;it!=v_global_workers.end();it++){ if((*it).pid == pid){ (*it).pid = getpid(); break; } } } else{ it = v_global_workers.begin()+worker_id; } //多进程本地工作者删除其他工作者的信息，只保留本工作者的信息 struct worker_info temp = *it; close(temp.socket_pairs.connecter); temp.socket_pairs.connecter = -1; v_global_workers.erase(it); for(int i=0;i<v_global_workers.size();i++){ close(v_global_workers[i].socket_pairs.connecter); close(v_global_workers[i].socket_pairs.accepter); } v_global_workers.clear(); v_global_workers.push_back(temp); struct worker_info *p_work_info = &v_global_workers[0]; //为写事件分配缓冲内存 p_work_info->write_buf = new char[WRITE_BUF_LEN]; // struct event_config *cfg = event_config_new(); // event_config_avoid_method(cfg,"epoll"); // struct event_base * base = event_base_new_with_config(cfg); // event_config_free(cfg); //局部event_base* base区别与全局base，避免父进程读写事件污染子进程事件 struct event_base * base = p_work_info->base = event_base_new(); /*初始化监听客户端连接事件*/ event_set(&ev_accept,listen_fd,EV_READ|EV_PERSIST,on_accept,NULL); event_base_set(base,&ev_accept); event_add(&ev_accept,NULL); event_set(&p_work_info->socket_pair_read_event,p_work_info->socket_pairs.accepter,EV_READ|EV_PERSIST,socketpair_worker_read,NULL); event_base_set(base,&p_work_info->socket_pair_read_event); event_add(&p_work_info->socket_pair_read_event,NULL); event_set(&p_work_info->socket_pair_write_event,p_work_info->socket_pairs.accepter,EV_WRITE,socketpair_worker_write,NULL); event_base_set(base,&p_work_info->socket_pair_write_event); //event_add(&p_work_info->socket_pair_write_event,NULL); event_base_dispatch(base); } void socketpair_worker_write(int fd, short ev, void *arg){ struct worker_info* client = &v_global_workers[0]; int len =0; len = write(fd,client->write_buf+client->offset,client->len-client->offset); if(len == -1){ //写操作被信号打断或不能写入，重新注册写事件 if(errno == EINTR || errno == EAGAIN){ event_add(&(client->socket_pair_write_event),NULL); return; } else{ err(1,"write error"); } } else if((client->offset + len) <client->len){ //数据没有完全写入 int offset = client->offset +len; memmove(client->write_buf,client->write_buf+offset,client->len-offset); client->offset = 0; client->len -= offset; event_add(&(client->socket_pair_write_event),NULL); return; } else{ //写入完成 client->offset = client->len = 0; } } void socketpair_worker_read(int fd, short ev, void *arg){ int buf_size = 1024; char buf[buf_size]; memset(buf,0,buf_size); uint16_t data_len = 0; int recv_len = 0; while(1){ int size =read(fd,buf+recv_len,buf_size-recv_len);//recv(fd,buf+recv_len,buf_size-recv_len,0); if(0 == size){ //管理者与工作者断开连接 cout<<"manager disconnection"<<endl; worker_info *worker = &v_global_workers[0]; event_del(&worker->socket_pair_read_event); event_del(&worker->socket_pair_write_event); delete[] worker->write_buf; close(worker->socket_pairs.accepter); //TODO因与管理者断开联系，结束该本地工作者进程 exit(1); return; } else if(size < 0){ // cout<<"socekt failure,disconnecting manager:"<<strerror(errno)<<endl; worker_info *worker = &v_global_workers[0]; event_del(&worker->socket_pair_read_event); event_del(&worker->socket_pair_write_event); delete[] worker->write_buf; close(worker->socket_pairs.accepter); //TODO因与管理者断开联系，结束该本地工作者进程 exit(1); return; } else{ recv_len += size; } while(recv_len >2){ if(0x0b != buf[0] || 0x0d != buf[1]){ recv_len--; memmove(buf,buf+1,recv_len); continue; } if(recv_len > 4){ data_len = *(uint16_t*)(buf+2); if(recv_len < data_len +4){ //一条信息未读取完全，继续读取 break; } proc_socketpair_worker_read(fd, buf+4, data_len); recv_len = recv_len - 4 - data_len; memmove(buf,buf+4+data_len,recv_len); } else{ break; } } //退出 if(0 == recv_len){ break; } } } void proc_socketpair_worker_read(int fd, char* data, int len){ char protocol = *data; switch(protocol){ default: break; } } void worker_read(int fd, short ev, void *arg){ struct event_accepted *client = (event_accepted*)arg; char* buf = new char[BUF_LEN]; if(buf == NULL){ err(1,"malloc failed"); } int recv_len = read(fd,buf,BUF_LEN); if(recv_len == 0){ //客户端断开连接 cout<<endl; cout<<"client disconnected"<<endl; event_del(client->ev_read); event_del(client->ev_write); delete client->ev_write; delete client->ev_read; delete client->write_buf; close(fd); free(client); delete[] buf; return; } else if(recv_len < 0){ cout<<endl; cout<<"socekt failure,disconnecting client:"<<strerror(errno)<<endl; event_del(client->ev_read); event_del(client->ev_write); delete client->ev_write; delete client->ev_read; delete client->write_buf; close(fd); free(client); delete[] buf; return; } //TODO 1.需要处理数组越界问题；2.这里只是一问一答的情况，如果需要主动推送消息需要增加设计； //TODO 1.在同一个文件描述符上如果多次注册同一事件会发生什么？这里改用链表记录需要发送的数据可能会好一点， memcpy(client->write_buf+client->len,buf,recv_len); client->len += recv_len; event_add(client->ev_write,NULL); delete[] buf; } void worker_write(int fd, short ev, void *arg){ struct event_accepted* client = (event_accepted*)arg; int len =0; len = write(fd,client->write_buf+client->offset,client->len-client->offset); if(len == -1){ //写操作被信号打断或不能写入，重新注册写事件 if(errno == EINTR || errno == EAGAIN){ event_add(client->ev_write,NULL); return; } else{ err(1,"write error"); event_del(client->ev_read); event_del(client->ev_write); delete client->ev_write; delete client->ev_read; delete client->write_buf; close(fd); free(client); } } else if((client->offset + len) <client->len){ //数据没有完全写入 int offset = client->offset +len; memmove(client->write_buf,client->write_buf+offset,client->len-offset); client->offset = 0; client->len -= offset; event_add(client->ev_write,NULL); return; } else{ //写入完成 client->offset = client->len = 0; } } void signal_child_cb(int fd,short ev,void *arg){ int stat; pid_t pid = waitpid(-1,&stat,WNOHANG); if(pid > 0){ pid_t pid_1 = fork(); if(0 == pid_1){ local_worker(-1,pid); exit(0); } } }

回答动态