好久沒寫網(wǎng)絡(luò)聊天室了，去年暑假可以說寫了一暑假，最近復習這些，又因為我一直偏向于多線程，就用多進程復習一下。
下面給出昨天寫的基于多進程、共享內(nèi)存的網(wǎng)絡(luò)聊天室代碼。每個進程負責一個連接，多個進程之間僅共享讀，不共享寫，因此無需信號量來同步。分配的一段內(nèi)存中，以數(shù)組的方式，分配給每個client一段buffer，每個clilent對應的buffer的索引就是connfd。當一個子進程收到客戶端數(shù)據(jù)后，通過每客戶端管道發(fā)送自己的pid給主進程，主進程通知除了該子進程的其他進程將該片內(nèi)存寫好的數(shù)據(jù)轉(zhuǎn)發(fā)給其他客戶端（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;//封裝每個客戶端連接數(shù)據(jù)struct client_data { sockaddr_in address; int connfd; pid_t pid; //負責該客戶端子進程的pid int pipefd[2]; //每個子進程pipe};static const char* shm_name = "/my_shm"; //共享內(nèi)存的名字int sig_pipefd[2]; //用來統(tǒng)一事件源int epollfd;int listenfd;int shmfd; char* share_mem = NULL; //共享內(nèi)存起始地址//客戶端連接數(shù)組，進程用客戶連接的編號來索引這個數(shù)組，即可取得相關(guān)的客戶連接數(shù)據(jù)client_data* users = NULL; //子進程和客戶連接的關(guān)系映射表，用子進程的pid來索引這個數(shù)組，即可取得該進程處理的客戶連接的編號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);}//子進程的信號處理函數(shù)，停止一個子進程void child_term_handler(int sig){ stop_child = true;}//子進程運行的函數(shù)，參數(shù)inx指出該子進程處理的客戶連接的編號，users是保存所有客戶連接數(shù)據(jù)的數(shù)組，參數(shù)share_mem指出共享內(nèi)存的起始地址int run_child(int idx, client_data* users, char* share_mem){ epoll_event events[MAX_EVENT_NUMBER]; //每個子進程使用I/O服用同時監(jiān)聽客戶連接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; //子進程需要設(shè)置自己的信號處理函數(shù)，因為fork會繼承父進程信號處理函數(shù) 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; //本子進程負責的客戶鏈接有數(shù)據(jù)到達 if(sockfd == connfd && (events[i].events & EPOLLIN)){ //清零該客戶對應的緩沖區(qū) memset(share_mem+idx*BUFFER_SIZE, '/0', BUFFER_SIZE); //將客戶數(shù)據(jù)讀取到對應的讀緩存中，該讀緩存是共享內(nèi)存的一段，它開始于idx*BUFFER_SIZE處，長度為BUFFER_SIZE字節(jié)，因此每個客戶連接是共享的 ret = recv(connfd, share_mem+idx*BUFFER_SIZE, BUFFER_SIZE-1, 0); //留一個字節(jié)為'/0'間隔 if(ret < 0){ if(errno != EAGAIN) stop_child = true; } else if(ret == 0) stop_child = true; else //成功讀取客戶數(shù)據(jù)后就通知主進程，讓主進程吩咐其他進程轉(zhuǎn)發(fā) send(pipefd, (char*)&idx, sizeof(idx), 0); } //主進程通過管道通知本進程需要轉(zhuǎn)發(fā)第client個客戶端的數(shù)據(jù)到本進程負責的客戶 else if(sockfd == pipefd && (events[i].events & EPOLLIN)){ int client = 0; //接受主進程發(fā)來的數(shù)據(jù)，即客戶的編號，用來索引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 //轉(zhuǎn)發(fā)給自己的客戶 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;/////////////////////////////////////////////////////////////// //創(chuàng)建共享內(nèi)存，作為所有客戶連接的讀緩存 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映射共享內(nèi)存 //這是共享內(nèi)存的一種方式，另外一種使用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個客戶連接的數(shù)據(jù) users[user_count].address = client_address; users[user_count].connfd = connfd; //在子進程和父進程間建立管道，以傳遞必要的數(shù)據(jù) 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]); //子進程關(guān)掉一端，子進程給父進程發(fā)數(shù)據(jù)使用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]); //記錄新的客戶連接在數(shù)組users中的索引值，建立進程pid和索引值的映射關(guān)系 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: //子進程退出，表示有客戶端關(guān)閉了連接 { pid_t pid; int stat; while((pid = waitpid(-1, &stat, WNOHANG)) > 0){ //用子進程的pid取得被關(guān)閉客戶連接的編號 int del_user = sub_process[pid]; sub_process[pid] = -1; if(del_user < 0 || del_user > USER_LIMIT) continue; //清除數(shù)據(jù) 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: //結(jié)束服務器程序 { 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; } } } } //某個子進程收到數(shù)據(jù)，向父進程通知 else if(events[i].events & EPOLLIN){ int child = 0; //讀取管道數(shù)據(jù)，收到的數(shù)據(jù)時child變量記錄了哪個客戶連接有數(shù)據(jù)到達 ret =recv(sockfd, (char*)&child, sizeof(child), 0); if(ret == -1) continue; else if(ret == 0) continue; else{ //向除負責第child個客戶的子進程之外的子進程發(fā)送消息，通知他們有客戶數(shù)據(jù)要寫 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;}新聞熱點
疑難解答
