简易版的进程池模型学习

1、进程池流程

父进程流程

第一步：

make_child 初始化子进程

循环创建子进程，并初始化父进程的子进程管理结构体数组 child，通过 socket_pair 将 socket描述符一端放入数组

子进程流程

recv_fd 等待父进程发送任务

send_file 发送文件数据

Write 向父进程发送完成任务

第二步：

父进程 epoll 监控 fd_listen 描述符。

父进程 epoll 监控 parr 结构体数组的 socket 描述符

第三步：

While 1 启动 epoll_wait，等待是否有客户端连接

有客户端连接后， accept 获得描述符，循环找到非忙碌的子进程，并发送给子进程，标记对

应子进程忙碌。

当子进程完成任务后，父进程一旦监控 socket 描述符可读，代表子进程非忙碌，然后标记子

进程非忙碌。

2、头文件及相关数据结构

#include<sys/stat.h>
#include <sys/stat.h>
#include<errno.h>
#include <fcntl.h>
#include <signal.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <netdb.h>
#include <string.h>
#include <sys/types.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include<sys/wait.h>#define FILENAME "file"typedef struct{                //维护子进程的数据结构pid_t pid;int tfds;short busy;
}child,*pchild;     typedef struct{               //网络数据发送的数据结构int len;char buf[1000];
}train,*ptrain;void send_fd(int fdw,int fd);
void recv_fd(int fdr,int* fd);
void make_child(pchild p,int len);
void child_handle(int fdr);
void send_data(int );
int send_n(int ,char *,int);
int recv_n(int,char *,int);

3、父进程源代码

#include"func.h"int exitfds[2];       //退出拉起管道
int exit_num=0;       //用于退出时判断机制void set_status(int fd)
{int status=fcntl(fd,F_GETFL);status=status|O_NONBLOCK;fcntl(fd,F_SETFL,status);
}void sighandle(int signum)
{write(exitfds[1],"over",4);
}int main(int argc,char **argv)
{   pipe(exitfds);signal(SIGINT,sighandle);if(4!=argc){perror("error argcs!\nplease enter IP  SOCKET  proccess_num\n");return -1;}int child_len=atoi(argv[3]);//首先为每个子进程动态申请空间，用指针实现数组效果pchild p=(pchild)calloc(child_len,sizeof(child)); //创建子进程，并初始化数据结构，//封装成一个Init函数make_child(p,child_len);int sfd;sfd=socket(AF_INET,SOCK_STREAM,0);if(sfd==-1){perror("socket");return -1;}struct sockaddr_in ser;ser.sin_addr.s_addr=inet_addr(argv[1]);ser.sin_port=htons(atoi(argv[2]));ser.sin_family=AF_INET;int ret;ret=bind(sfd,(struct sockaddr *)&ser,sizeof(ser));if(-1==ret){perror("bind");return -1;}int epfd=epoll_create(1);     //创造一个epfd的句柄struct epoll_event event,*evs;evs=(struct epoll_event *)calloc(child_len+1,sizeof(struct epoll_event));memset(&event,0,sizeof(event));event.data.fd=sfd;event.events=EPOLLIN;epoll_ctl(epfd,EPOLL_CTL_ADD,sfd,&event);int i;for(i=0;i<child_len;i++)     //监管注册每个子进程的描述符管道{event.data.fd=p[i].tfds;event.events=EPOLLIN;epoll_ctl(epfd,EPOLL_CTL_ADD,p[i].tfds,&event);}listen(sfd,child_len);int j,ret1,newfd;char flag[6];static int once=0;set_status(exitfds[0]);while(1){ret=read(exitfds[0],flag,sizeof(flag));if(ret>0){if(0==exit_num){printf("exerting quit action--please wait····\n");event.data.fd=sfd;       //不在接受连接event.events=EPOLLIN;epoll_ctl(epfd,EPOLL_CTL_DEL,sfd,&event);for(i=0;i<child_len;i++){printf("正在处理子进程p[i].pid=%d  ",p[i].pid);printf("\n");kill(p[i].pid,SIGKILL);}wait(NULL);printf("you have quited\n");return(0);}else{printf("请等待，有客户正在交易···\n"); }}ret1=epoll_wait(epfd,evs,child_len+1,-1);for(i=0;i<ret1;i++){                   //妈的，又他妈犯这种错误了if(evs[i].data.fd==sfd)      //网络请求到达{newfd=accept(sfd,NULL,NULL);//找到一个空闲的子进程，并且将newfd发送给它for(j=0;j<child_len;j++){if(p[j].busy==0){//通过tcp管道发送newfd给子进程send_fd(p[j].tfds,newfd); exit_num++;p[j].busy=1;printf("find a not busy proccess,send newfd success!\n");break;}}close(newfd);       //关闭父进程对newfd的引用计数}//监听子进程的tcp管道的发送完成信息for(j=0;j<child_len;j++)              {if(evs[i].data.fd==p[j].tfds){//read(p[j].tfds,&flag,sizeof(flag));char buf[15]={0};read(p[j].tfds,buf,sizeof(buf));printf("%s\n",buf);//read(p[j].tfds,&flag,sizeof(flag));p[j].busy=0;exit_num--;printf("child proccess %d is not busy!\n",j);if(0==exit_num){printf("Are you stall want to finish this deal,please push Ctrl+C\n");}}}}}
}

4、子进程流程代码

#include"func.h"/*首先，要创建len个子进程，就会用到fork( )函数，于是，必须把父进程和子进程分别开来，一个父进程，len个子进程。
其次，初始化是针对定义的数据结构初始化，有：
1、每个子进程在fork创建之后需要根据返回值保存自己的pid.
2、对于全双工的TCP管道，需要用socketpair( )函数初始化
3、刚开始时，将其每个子进程的busy位置为0，表示空闲
对于子进程，必须让它在循环里面持续发送，从父进程接收任务，父进程将到发送数据
，再到发送完，通知父进程，可以通过写一个flag，然后父进程将其busy改为0，
表示空闲。
*/
void child_handle(int fdr)
{int newfd;char flag[6]={0};strcpy(flag,"over");while(1){recv_fd(fdr,&newfd);send_data(newfd);printf("I am child,send success!\n");write(fdr,&flag,sizeof(flag));//write(fdr,"hello",5);}
}void make_child(pchild p,int len)
{int fds[2],i,ret;pid_t pid;for(i=0;i<len;i++){ret=socketpair(AF_LOCAL,SOCK_STREAM,0,fds);pid=fork();if(0==pid){close(fds[1]);child_handle(fds[0]);//p[i].tfds=fds[0]; 将fds[0]端留给子进程}   //让其子进程在里面循环，不出来close(fds[0]);p[i].pid=pid;       //获取子进程的pidp[i].busy=0;        //置为空闲p[i].tfds=fds[1];   //将管道的一端传递给主进程}
}

5、父进程向子进程发送套接字文件描述符（内核信息）

#include "func.h"void send_fd(int fdw,int fd)
{struct msghdr msg;memset(&msg,0,sizeof(msg));struct cmsghdr *cmsg;int len=CMSG_LEN(sizeof(int));cmsg=(struct cmsghdr *)calloc(1,len);cmsg->cmsg_len=len;cmsg->cmsg_level = SOL_SOCKET;cmsg->cmsg_type = SCM_RIGHTS;*(int*)CMSG_DATA(cmsg)=fd;msg.msg_control=cmsg;msg.msg_controllen=len;char buf1[10]="hello";char buf2[10]="world";struct iovec iov[2];iov[0].iov_base=buf1;iov[0].iov_len=5;iov[1].iov_base=buf2;iov[1].iov_len=5;msg.msg_iov=iov;msg.msg_iovlen=2;int ret=sendmsg(fdw,&msg,0);if(-1==ret){perror("sendmsg");return;}
}
void recv_fd(int fdr,int* fd)
{struct msghdr msg;memset(&msg,0,sizeof(msg));struct cmsghdr *cmsg;int len=CMSG_LEN(sizeof(int));cmsg=(struct cmsghdr *)calloc(1,len);cmsg->cmsg_len=len;cmsg->cmsg_level = SOL_SOCKET;cmsg->cmsg_type = SCM_RIGHTS;msg.msg_control=cmsg;msg.msg_controllen=len;char buf1[10]="hello";char buf2[10]="world";struct iovec iov[2];iov[0].iov_base=buf1;iov[0].iov_len=5;iov[1].iov_base=buf2;iov[1].iov_len=5;msg.msg_iov=iov;msg.msg_iovlen=2;int ret=recvmsg(fdr,&msg,0);if(-1==ret){perror("sendmsg");return;}*fd=*(int*)CMSG_DATA(cmsg);
}

6、子进程的主要逻辑业务----发送文件

#include"func.h"void send_data(int newfd)
{train t;memset(&t,0,sizeof(t));strcpy(t.buf,FILENAME);t.len=strlen(FILENAME);printf("the strcpy t.buf is %s, len =%d\n",t.buf,t.len);//先发文件名int ret;ret=send_n(newfd,(char *)&t,t.len+4);if(ret==-1){perror("send");return;}int fd=open(FILENAME,O_RDONLY);if(-1==fd){perror("open");return;}struct stat filestat;fstat(fd,&filestat);t.len=sizeof(long);memcpy(t.buf,&filestat.st_size,sizeof(filestat.st_size));send_n(newfd,(char*)&t,4+t.len);while(memset(t.buf,0,sizeof(t.buf)),(t.len=read(fd,t.buf,sizeof(t.buf)))>0){send_n(newfd,(char *)&t,t.len+4);}t.len=0;send_n(newfd,(char *)&t,4+t.len);close(newfd);close(fd);
}

7、为了匹配网络两端的发送速度，以及发送大文件时，可能的缓冲区大小的瓶颈，造成

的数据丢失，故需要精确控制要发送和接受的字节数，需要循环发送。

#include"func.h"int send_n(int sfd,char *p,int len)
{int total=0,ret;while(total<len){ret=send(sfd,p+total,len-total,0);//取出每一次发送了多少个total=total+ret;}return 0;
}int recv_n(int sfd,char *p,int len )
{int total=0,ret;while(total<len){ret=recv(sfd,p+total,len-total,0);total=total+ret;}return 0;
}

8、客户器端源代码

#include<time.h>
#include <sys/stat.h>
#include<errno.h>
#include <fcntl.h>
#include <signal.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <netdb.h>
#include <string.h>
#include <sys/types.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/epoll.h>
#include<sys/wait.h>typedef struct{pid_t pid;int tfds;short busy;
}child,*pchild;void send_fd(int fdw,int fd);
void recv_fd(int fdr,int* fd);
void make_child(pchild p,int len);
void child_handle(int fdr);
int recv_n(int,char *,int);int send_n(int sfd,char *p,int len)
{int total=0,ret;while(total<len){ret=send(sfd,p+total,len-total,0);//取出每一次发送了多少个total=total+ret;}return 0;
}int recv_n(int sfd,char *p,int len )
{int total=0,ret;while(total<len){ret=recv(sfd,p+total,len-total,0);total=total+ret;}return 0;
}int main(int argc,char **argv)
{if(3!=argc){printf("error argcs!\n");return -1;}int sfd;sfd=socket(AF_INET,SOCK_STREAM,0);if(-1==sfd){perror("socket");return -1;}struct sockaddr_in ser;memset(&ser,0,sizeof(ser));int ret,ret1;ser.sin_family=AF_INET;ser.sin_port=htons(atoi(argv[2]));ser.sin_addr.s_addr=inet_addr(argv[1]);ret=connect(sfd,(struct sockaddr*)&ser,sizeof(ser));int len;char buf[1000]={0};recv_n(sfd,(char *)&len,sizeof(len));     //先接受文件名recv_n(sfd,buf,len);int fd=open(buf,O_WRONLY|O_CREAT,0666);if(-1==fd){perror("open");return -1;}recv_n(sfd,(char *)&len,sizeof(len));     //在发送文件大小long file_len;recv_n(sfd,(char*)&file_len,len);long l=0;time_t  now,check;time(&now);check=now;while(1){recv_n(sfd,(char *)&len,sizeof(len));if(len>0){l=l+len;memset(buf,0,sizeof(buf));recv_n(sfd,buf,len);//接受完后就要写到袭击者边新建的文件里write(fd,buf,len);//接受多少就写多少time(&now);if(now>check+1){printf("Now Data is %5.2f%s\r",100*(double)l/file_len,"%");fflush(stdout);check=now;}}else{printf("Down load successful!\n");break;}}close(fd);close(sfd);return 0;
}