好久沒寫網絡聊天室了，去年暑假可以說寫了一暑假，最近復習這些，又因為我一直偏向于多線程，就用多進程復習一下。
下面給出昨天寫的基于多進程、共享內存的網絡聊天室代碼。每個進程負責一個連接，多個進程之間僅共享讀，不共享寫，因此無需信號量來同步。分配的一段內存中，以數組的方式，分配給每個client一段buffer，每個clilent對應的buffer的索引就是connfd。當一個子進程收到客戶端數據后，通過每客戶端管道發送自己的pid給主進程，主進程通知除了該子進程的其他進程將該片內存寫好的數據轉發給其他客戶端（sub_PRoess[pid]=connd）。
代碼如下：
#include <sys/socket.h>#include <netinet/in.h>#include <arpa/inet.h>#include <assert.h>#include <stdio.h>#include <unistd.h>#include <errno.h>#include <string.h>#include <stdlib.h>#include <sys/epoll.h>#include <signal.h>#include <sys/wait.h>#include <sys/mman.h>#include <sys/stat.h>#include <fcntl.h>const int USER_LIMIT = 3;const int BUFFER_SIZE = 1024;const int FD_LIMIT = 65545;const int MAX_EVENT_NUMBER = 1024;const int PROCESS_LIMIT = 65536;//封裝每個客戶端連接數據struct client_data { sockaddr_in address; int connfd; pid_t pid; //負責該客戶端子進程的pid int pipefd[2]; //每個子進程pipe};static const char* shm_name = "/my_shm"; //共享內存的名字int sig_pipefd[2]; //用來統一事件源int epollfd;int listenfd;int shmfd; char* share_mem = NULL; //共享內存起始地址//客戶端連接數組，進程用客戶連接的編號來索引這個數組，即可取得相關的客戶連接數據client_data* users = NULL; //子進程和客戶連接的關系映射表，用子進程的pid來索引這個數組，即可取得該進程處理的客戶連接的編號int* sub_process = 0;int user_count = 0; //客戶連接下標，這個名字有點誤導，總之user_count>=USER_LIMIT即連接過多bool stop_child = false; //停止一個子進程，這個是全部變量，每個子進程都有自己拷貝的一份int setnonblocking(int fd){ int old_option = fcntl(fd, F_GETFL); int new_option = old_option | O_NONBLOCK; fcntl(fd, F_SETFL, new_option); return old_option;}void addfd(int epfd, int fd){ epoll_event event; event.data.fd = fd; event.events = EPOLLIN | EPOLLET; epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event); setnonblocking(fd);}void sig_handler(int sig){ int save_errno = errno; int msg = sig; send(sig_pipefd[1], (char*)&msg, 1, 0); errno = save_errno;}void addsig(int sig, void(*handler)(int), bool restart = true){ struct sigaction sa; memset(&sa, '/0', sizeof(sa)); sa.sa_handler = handler; if(restart) sa.sa_flags |= SA_RESTART; sigfillset(&sa.sa_mask); assert(sigaction(sig, &sa, NULL) != -1);}void del_resource(){ close(sig_pipefd[0]); close(sig_pipefd[1]); close(listenfd); close(epollfd); shm_unlink(shm_name);}//子進程的信號處理函數，停止一個子進程void child_term_handler(int sig){ stop_child = true;}//子進程運行的函數，參數inx指出該子進程處理的客戶連接的編號，users是保存所有客戶連接數據的數組，參數share_mem指出共享內存的起始地址int run_child(int idx, client_data* users, char* share_mem){ epoll_event events[MAX_EVENT_NUMBER]; //每個子進程使用I/O服用同時監聽客戶連接socket和與父進程通信的pipe描述符 int child_epollfd = epoll_create(5); assert(child_epollfd != -1); int connfd = users[idx].connfd; addfd(child_epollfd, connfd); int pipefd = users[idx].pipefd[1]; addfd(child_epollfd, pipefd); int ret; //子進程需要設置自己的信號處理函數，因為fork會繼承父進程信號處理函數 addsig(SIGTERM, child_term_handler, false); while(!stop_child){ int number = epoll_wait(child_epollfd, events, MAX_EVENT_NUMBER, -1); if(number < 0 && errno != EINTR){ printf("epoll failure/n"); break; } for(int i=0; i<number; ++i){ int sockfd = events[i].data.fd; //本子進程負責的客戶鏈接有數據到達 if(sockfd == connfd && (events[i].events & EPOLLIN)){ //清零該客戶對應的緩沖區 memset(share_mem+idx*BUFFER_SIZE, '/0', BUFFER_SIZE); //將客戶數據讀取到對應的讀緩存中，該讀緩存是共享內存的一段，它開始于idx*BUFFER_SIZE處，長度為BUFFER_SIZE字節，因此每個客戶連接是共享的 ret = recv(connfd, share_mem+idx*BUFFER_SIZE, BUFFER_SIZE-1, 0); //留一個字節為'/0'間隔 if(ret < 0){ if(errno != EAGAIN) stop_child = true; } else if(ret == 0) stop_child = true; else //成功讀取客戶數據后就通知主進程，讓主進程吩咐其他進程轉發 send(pipefd, (char*)&idx, sizeof(idx), 0); } //主進程通過管道通知本進程需要轉發第client個客戶端的數據到本進程負責的客戶 else if(sockfd == pipefd && (events[i].events & EPOLLIN)){ int client = 0; //接受主進程發來的數據，即客戶的編號，用來索引buffer ret = recv(sockfd, (char *)&client, sizeof(client), 0); if(ret < 0){ if(errno != EAGAIN) stop_child = true; } else if(ret == 0) stop_child = true; else //轉發給自己的客戶 send(connfd, share_mem+client*BUFFER_SIZE, BUFFER_SIZE, 0); } else continue; } } close(connfd); close(pipefd); close(child_epollfd); return 0;}int main(int argc, char** argv){ if( argc <= 2 ) { printf( "usage: %s ip_address port_number/n", basename( argv[0] ) ); return 1; } const char* ip = argv[1]; int port = atoi( argv[2] ); int ret = 0; struct sockaddr_in address; bzero(&address, sizeof(address)); address.sin_family = AF_INET; inet_pton(AF_INET, ip, &address.sin_addr); address.sin_port = htons(port); int listenfd = socket(PF_INET, SOCK_STREAM, 0); assert(listenfd >= 0); int on = 1; ret = setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)); assert(ret != -1); ret = bind(listenfd, (struct sockaddr*)&address, sizeof(address)); assert(ret != -1); ret = listen(listenfd, 5); assert(ret != -1);///////////////////////////////////////////////////////////// user_count = 0; users = new client_data[USER_LIMIT]; sub_process = new int [PROCESS_LIMIT]; for(int i=0; i<PROCESS_LIMIT; ++i) sub_process[i] = -1;//////////////////////////////////////////////////////////// epoll_event events[MAX_EVENT_NUMBER]; epollfd = epoll_create(5); assert(epollfd != -1); addfd(epollfd, listenfd); //socketpair是全雙工的，所以父子進程通信無需向pipe一樣需要兩個pipe[2] //fork完畢socketpair可以雙向通信 ret = socketpair(PF_UNIX, SOCK_STREAM, 0, sig_pipefd); assert(ret != -1); setnonblocking( sig_pipefd[1] ); addfd(epollfd, sig_pipefd[0]); // add all the interesting signals here addsig(SIGCHLD, sig_handler); addsig(SIGTERM, sig_handler); addsig(SIGINT, sig_handler); addsig(SIGPIPE, SIG_IGN); bool stop_server = false; bool terminate = false;/////////////////////////////////////////////////////////////// //創建共享內存，作為所有客戶連接的讀緩存 shmfd = shm_open(shm_name, O_CREAT | O_RDWR, 0666); assert(shmfd != -1); //清空且resize文件大小為USER_LIMIT*BUFFER_SIZE ret = ftruncate(shmfd, USER_LIMIT*BUFFER_SIZE); assert(ret != -1); //通過上面生成的一定大小的文件來使用mmap映射共享內存 //這是共享內存的一種方式，另外一種使用SystemV的shmat share_mem = (char *)mmap(NULL, USER_LIMIT*BUFFER_SIZE, PROT_WRITE | PROT_READ, MAP_SHARED, shmfd, 0); assert(share_mem != MAP_FAILED); close(shmfd); //close shmfd is ok////////////////////////////////////////////////////////////// while(!stop_server){ int number = epoll_wait(epollfd, events, MAX_EVENT_NUMBER, -1); if(number < 0 && errno != EINTR){ printf("epoll failure/n"); break; } for(int i=0; i<number; ++i){ int sockfd = events[i].data.fd; if(sockfd == listenfd){ struct sockaddr_in client_address; socklen_t len = sizeof(client_address); int connfd = accept(listenfd, (struct sockaddr*)&client_address, &len); if(connfd < 0){ printf("errno is: %d/n", errno); continue; } if(user_count >= USER_LIMIT){ //limit const char* info = "too many users/n"; printf("%s", info); send(connfd, info, strlen(info), 0); close(connfd); continue; } //保存第user_count個客戶連接的數據 users[user_count].address = client_address; users[user_count].connfd = connfd; //在子進程和父進程間建立管道，以傳遞必要的數據 ret = socketpair(PF_UNIX, SOCK_STREAM, 0, users[user_count].pipefd); assert(ret != -1); pid_t pid = fork(); if(pid < 0){ close(connfd); continue; //!!!!!!! } else if(pid == 0){ //in child close(epollfd); close(listenfd); close(users[user_count].pipefd[0]); //子進程關掉一端，子進程給父進程發數據使用pipefd[1] close(sig_pipefd[0]); close(sig_pipefd[1]); run_child(user_count, users, share_mem); munmap((void*)share_mem, USER_LIMIT*BUFFER_SIZE); exit(0); } else{ close(connfd); close(users[user_count].pipefd[1]); //同理 addfd(epollfd, users[user_count].pipefd[0]); //記錄新的客戶連接在數組users中的索引值，建立進程pid和索引值的映射關系 users[user_count].pid = pid; sub_process[pid] = user_count; user_count++; } } //handle signal else if(sockfd == sig_pipefd[0] && (events[i].events & EPOLLIN)){ int sig; char signals[1024]; ret = recv(sig_pipefd[0], signals, sizeof(signals), 0); if(ret == -1) continue; else if(ret == 0) continue; else{ for(int i=0; i<ret; ++i){ switch(signals[i]){ case SIGCHLD: //子進程退出，表示有客戶端關閉了連接 { pid_t pid; int stat; while((pid = waitpid(-1, &stat, WNOHANG)) > 0){ //用子進程的pid取得被關閉客戶連接的編號 int del_user = sub_process[pid]; sub_process[pid] = -1; if(del_user < 0 || del_user > USER_LIMIT) continue; //清除數據 epoll_ctl(epollfd, EPOLL_CTL_DEL, users[del_user].pipefd[0], 0); close(users[del_user].pipefd[0]); //用最后一個user替換該位置 users[del_user] = users[--user_count]; sub_process[users[del_user].pid] = del_user; //修正sub_process對應的值，也就是修正最后一個客戶端pid對應的客戶編號 } if(terminate && user_count == 0) stop_server = true; break; } case SIGTERM: case SIGINT: //結束服務器程序 { printf("kill all the child new/n"); if(user_count == 0){ stop_server = true; break; } for(int i=0; i<user_count; ++i){ int pid = users[i].pid; kill(pid, SIGTERM); //kill每個子進程 } terminate = true; break; } default: break; } } } } //某個子進程收到數據，向父進程通知 else if(events[i].events & EPOLLIN){ int child = 0; //讀取管道數據，收到的數據時child變量記錄了哪個客戶連接有數據到達 ret =recv(sockfd, (char*)&child, sizeof(child), 0); if(ret == -1) continue; else if(ret == 0) continue; else{ //向除負責第child個客戶的子進程之外的子進程發送消息，通知他們有客戶數據要寫 for(int j=0; j<user_count; ++j){ if(users[j].pipefd[0] != sockfd){ printf("send data to child accross pipe/n"); send(users[j].pipefd[0], (char*)&child, sizeof(child), 0); } } } } } } del_resource(); return 0;}新聞熱點
疑難解答
