进程池

进程池的实现

父子进程创建

35
1
//head.h
2
enum workerStatus {
3
    FREE,
4
    BUSY
5
};
6
typedef struct{
7
    pid_t pid; //工作进程的pid
8
    int status;//工作进程的状态
9
} processData_t;
10
//main.c
11
int main(int argc, char *argv[]){
12
    //./main 192.168.135.132 5678 10
13
    ARGS_CHECK(argc,4);
14
    int workerNum = atoi(argv[3]);
15
    processData_t *workerList = (processData_t *)calloc(sizeof(processData_t),workerNum);
16
    //workerList记录了所有工作进程的状态
17
    makeChild(workerList,workerNum);
18
    while(1);
19
}
20
//worker.c
21
int makeChild(processData_t *pProcssData, int processNum){
22
    pid_t pid;
23
    for(int i = 0;i < processNum; ++i){
24
        pid = fork();
25
        if(pid == 0){
26
            handleEvent();
27
        }
28
        pProcssData[i].pid = pid;
29
        pProcssData[i].status = FREE;
30
    }
31
    return 0;
32
}
33
void handleEvent(){//工作进程目前的工作是死循环
34
    while(1);
35
}

父进程处理网络连接

16
1
int tcpInit(char *ip, char *port,int *pSockFd){
2
    *pSockFd = socket(AF_INET,SOCK_STREAM,0);
3
    struct sockaddr_in addr;
4
    bzero(&addr,sizeof(struct sockaddr_in));
5
    addr.sin_family = AF_INET;
6
    addr.sin_addr.s_addr = inet_addr(ip);
7
    addr.sin_port = htons(atoi(port));
8
    int reuse = 1;
9
    int ret;
10
    ret = setsockopt(*pSockFd,SOL_SOCKET,SO_REUSEADDR,&reuse,sizeof(reuse));
11
    ERROR_CHECK(ret,-1,"setsockopt");
12
    ret = bind(*pSockFd,(struct sockaddr*)&addr,sizeof(struct sockaddr_in));
13
    ERROR_CHECK(ret,-1,"bind");
14
    listen(*pSockFd,10);
15
    return 0;
16
}

本地套接字

1
1
int socketpair(int domain, int type, int protocol, int sv[2]);

在这里domain必须填写AF_LOCAL，type可以选择流式数据还是消息数据，protocol一般填0表示不需要任何额外的协议，sv这个参数和pipe的参数一样，是一个长度为2的整型数据，用来存储管道两端的文件描述符（值得注意的是，sv[0]和sv[1]没有任何的区别）。一般socketpair之后会配合fork函数一起使用，从而实现父子进程之间的通信。从数据传递使用上面来看，本地套接字和网络套接字是完全一致的，但是本地套接字的效率更高，因为它在拷贝数据的时候不需要处理协议相关内容。

父子进程共享文件描述符

父进程会监听特定某个IP:PORT，如果有某个客户端连接之后，子进程需要能够连上accept得到的已连接套接字的文件描述符，这样子进程才能和客户端进行通信。这种文件描述符的传递不是简单地传输一个整型数字就行了，而是需要让父子进程共享一个套接字文件对象。

但是这里会遇到麻烦，因为accept调用是在fork之后的，所以父子进程之间并不是天然地共享文件对象。倘若想要在父子进程之间共享acccept调用返回的已连接套接字，需要采用一些特别的手段：一方面，父子进程之间需要使用本地套接字来通信数据。另一方面需要使用sendmsg和recvmsg函数来传递数据。

20
1
ssize_t sendmsg(int sockfd, const struct msghdr *msg, int flags);
2
ssize_t recvmsg(int sockfd, struct msghdr *msg, int flags);
3
struct iovec
4
{                   /* Scatter/gather array items */
5
    void *iov_base; /* Starting address */
6
    size_t iov_len; /* Number of bytes to transfer */
7
};
8

9
struct msghdr
10
{
11
    void *msg_name;        /* optional address */
12
    socklen_t msg_namelen; /* size of address */
13
    struct iovec *msg_iov; /* scatter/gather array */
14
    size_t msg_iovlen;     /* # elements in msg_iov */
15
    void *msg_control;     /* ancillary data, see below */
16
    size_t msg_controllen; /* ancillary data buffer len */
17
    int msg_flags;         /* flags on received message */
18
};
19
ssize_t readv(int fd, const struct iovec *iov, int iovcnt);
20
ssize_t writev(int fd, const struct iovec *iov, int iovcnt);

使用sendmsg和recvmsg的时候附加一个消息头部，即一个struct msghdr类型的结构体。

首先，需要将要传递的内容存储入msg_iov当中，在这里需要注意的是，元素类型为struct iovec的数组可以存储一组离散的消息，只需要将每个消息的起始地址和本消息的长度存入数组元素中即可。（使用writev和readv可以直接读写一组离散的消息）

接下来，需要将文件描述符的信息存入控制字段msg_control中，这个我们需要存储一个地址值，该地址指向了一个struct cmsghdr类型的控制信息。如果存在多个控制信息，会构成一个控制信息序列，规范要求使用者绝不能直接操作控制信息序列，而是需要用一系列的cmsg宏来间接操作。CMSG_FIRSTHDR用来获取序列中的第一个控制信息（CMSG_NXTHDR获取下一个），CMSG_DATA宏用来设置控制信息的具体数据的地址；CMSG_LEN宏用来设置具体数据占据内存空间的大小。

15
1
//man cmsg
2
struct cmsghdr{
3
    size_t cmsg_len; /* Data byte count, including header
4
                        (type is socklen_t in POSIX) */
5
    int cmsg_level;  /* Originating protocol */
6
    int cmsg_type;   /* Protocol-specific type */
7
    /* followed by
8
        unsigned char cmsg_data[]; */
9
};
10
struct cmsghdr *CMSG_FIRSTHDR(struct msghdr *msgh);
11
struct cmsghdr *CMSG_NXTHDR(struct msghdr *msgh, struct cmsghdr *cmsg);
12
size_t CMSG_ALIGN(size_t length);
13
size_t CMSG_SPACE(size_t length);
14
size_t CMSG_LEN(size_t length);
15
unsigned char *CMSG_DATA(struct cmsghdr *cmsg);

为了传递文件描述符，需要将结构体中的cmsg_level字段设置为SOL_SOCKET，而 cmsg_type字段需要设置为SCM_RIGHTS，再将数据部分设置为文件描述符。这样，该文件描述符所指的文件对象就可以传递到另一个进程了。

42
1
int sendFd(int pipeFd, int fdToSend){
2
    struct msghdr hdr;
3
    bzero(&hdr,sizeof(struct msghdr));//这一步绝对不能少
4
    struct iovec iov[1];
5
    char buf[] = "Hello";
6
    iov[0].iov_base = buf;
7
    iov[0].iov_len = 5;
8
    hdr.msg_iov = iov;
9
    hdr.msg_iovlen = 1;
10
    struct cmsghdr *pcmsghdr = (struct cmsghdr *)calloc(1,CMSG_LEN(sizeof(int)));
11
    pcmsghdr->cmsg_len = CMSG_LEN(sizeof(int));
12
    //控制信息的数据部分只有int类型的文件描述符
13
    pcmsghdr->cmsg_level = SOL_SOCKET;
14
    pcmsghdr->cmsg_type = SCM_RIGHTS;//SCM->socket-level control message
15
    //表示在socket层传递的是访问权力，这样接受进程就可以访问对应文件对象了
16
    *(int *)CMSG_DATA(pcmsghdr) = fdToSend;
17
    //数据部分是文件描述符
18
    hdr.msg_control = pcmsghdr;
19
    hdr.msg_controllen = CMSG_LEN(sizeof(int));
20
    int ret = sendmsg(pipeFd,&hdr,0);
21
    ERROR_CHECK(ret,-1,"sendmsg");
22
}
23
int recvFd(int pipeFd, int *pFd){
24
    struct msghdr hdr;
25
    bzero(&hdr,sizeof(struct msghdr));
26
    struct iovec iov[1];
27
    char buf[6] = {0};//除了数据内容以外，其他和sendmsg是一致的
28
    iov[0].iov_base = buf;
29
    iov[0].iov_len = 5;//这里一定不能填0
30
    hdr.msg_iov = iov;
31
    hdr.msg_iovlen = 1;
32
    struct cmsghdr *pcmsghdr = (struct cmsghdr *)calloc(1,CMSG_LEN(sizeof(int)));
33
    pcmsghdr->cmsg_len = CMSG_LEN(sizeof(int));
34
    pcmsghdr->cmsg_level = SOL_SOCKET;
35
    pcmsghdr->cmsg_type = SCM_RIGHTS;//SCM->socket-level control message
36
    hdr.msg_control = pcmsghdr;
37
    hdr.msg_controllen = CMSG_LEN(sizeof(int));
38
    int ret = recvmsg(pipeFd,&hdr,0);
39
    ERROR_CHECK(ret,-1,"recvmsg");
40
    *pFd = *(int *)CMSG_DATA(pcmsghdr);
41
    return 0;
42
}

要特别注意的是，传递的文件描述符在数值上完全可能是不相等的，但是它们对应的文件对象确实是同一个，自然文件读写偏移量也是共享的，和之前使用dup或者是先打开文件再fork的情况是一致的。

至此，我们就可以实现一个进程池的服务端了：

• 启动父进程

• makeChild:父进程在创建每个子进程时，先调用socketpair

• handleEvent:子进程被创建之后，执行进程工作函数

• recvFd:子进程等待一个文件描述符，在父进程未发送的时候，子进程处于阻塞状态

• tcpInit:父进程初始化一个网络socket

• epollFunc:父进程使用epoll等IO多路复用机制监听网络socket和每个子进程的本地socket的一端。

• 如果有客户端通过网络连接父进程，那么父进程会accept得到一个已连接socket。

• sendFd:选择一个空闲的子进程，将已连接socket发送给子进程，之后父进程就不再和客户端直接网络通信，而是由子进程和客户端通信。

• 当某个子进程完成了任务之后，子进程可以通过本地socket通知父进程，并且重新将自己设为空闲。

142
1
//客户端
2
int main(int argc, char *argv[])
3
{
4
    ARGS_CHECK(argc, 3);
5
    int sockFd = socket(AF_INET, SOCK_STREAM, 0);
6
    struct sockaddr_in addr;
7
    bzero(&addr, sizeof(struct sockaddr_in));
8
    addr.sin_family = AF_INET;
9
    addr.sin_addr.s_addr = inet_addr(argv[1]);
10
    addr.sin_port = htons(atoi(argv[2]));
11
    int ret = connect(sockFd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in));
12
    ERROR_CHECK(ret, -1, "connect");
13
    char buf[1024] = {0};
14
    read(STDIN_FILENO,buf,sizeof(buf));
15
    send(sockFd,buf,strlen(buf)-1,0);
16
    bzero(buf,sizeof(buf));
17
    recv(sockFd,buf,sizeof(buf),0);
18
    puts(buf);
19
    close(sockFd);
20
    return 0;
21
}
22
//服务端主进程
23
int main(int argc, char *argv[]){
24
    //./main 192.168.135.132 5678 10
25
    ARGS_CHECK(argc,4);
26
    int workerNum = atoi(argv[3]);
27
    processData_t *workerList = (processData_t *)calloc(sizeof(processData_t),workerNum);
28
    makeChild(workerList,workerNum);
29
    int sockFd;
30
    tcpInit(argv[1],argv[2],&sockFd);
31
    int epfd = epollCtor();
32
    epollAdd(sockFd,epfd);
33
    for(int i = 0;i < workerNum; ++i){
34
        epollAdd(workerList[i].pipeFd,epfd);
35
    }
36
    int listenSize = workerNum+1;//socket+每个进程pipe的读端
37
    struct epoll_event * readylist = (struct epoll_event *)calloc(listenSize,sizeof(struct epoll_event));
38
    while(1){
39
        int readynum = epoll_wait(epfd,readylist,listenSize,-1);
40
        for(int i = 0;i < readynum; ++i){
41
            if(readylist[i].data.fd == sockFd){
42
                puts("accept ready");
43
                int netFd = accept(sockFd,NULL,NULL);
44
                for(int j = 0;j < workerNum; ++j){
45
                    if(workerList[j].status == FREE){
46
                        printf("No. %d worker gets his job, pid = %d\n", j, workerList[j].pid);
47
                        sendFd(workerList[j].pipeFd, netFd);
48
                        workerList[j].status = BUSY;
49
                        break;
50
                    }
51
                }
52
                close(netFd);//父进程交给子进程一定要关闭
53
            }
54
            else{
55
                puts("One worker finish his task!");
56
                int j;
57
                for(j = 0;j < workerNum;++j){
58
                    if(workerList[j].pipeFd == readylist[i].data.fd){
59
                        pid_t pid;
60
                        int ret = recv(workerList[j].pipeFd,&pid,sizeof(pid_t),0);
61
                        printf("No. %d worker finish, pid = %d\n",j,pid);
62
                        workerList[j].status = FREE;
63
                        break;
64
                    }
65
                }
66
            }
67
        }
68
    }
69
}
70
//服务端子进程
71
int makeChild(processData_t *pProcssData, int processNum)
72
{
73
    pid_t pid;
74
    for (int i = 0; i < processNum; ++i)
75
    {
76
        int pipeFd[2];
77
        socketpair(AF_LOCAL,SOCK_STREAM,0,pipeFd);
78
        pid = fork();
79
        if (pid == 0)
80
        {
81
            close(pipeFd[0]);
82
            handleEvent(pipeFd[1]);
83
        }
84
        close(pipeFd[1]);
85
        printf("pid = %d, pipefd[0] = %d\n", pid, pipeFd[0]);
86
        pProcssData[i].pid = pid;
87
        pProcssData[i].status = FREE;
88
        pProcssData[i].pipeFd = pipeFd[0];
89
    }
90
    return 0;
91
}
92
void handleEvent(int pipeFd)
93
{
94
    int netFd;
95
    while(1){
96
        recvFd(pipeFd,&netFd);
97
        char buf[1024] = {0};
98
        recv(netFd,buf,sizeof(buf),0);
99
        puts(buf);
100
        send(netFd,"Echo",4,0);
101
        close(netFd);
102
        pid_t pid = getpid();
103
        send(pipeFd,&pid,sizeof(pid_t),0);        
104
    }
105
}
106
//TCP初始化相关代码
107
int tcpInit(char *ip, char *port, int *pSockFd)
108
{
109
    *pSockFd = socket(AF_INET, SOCK_STREAM, 0);
110
    struct sockaddr_in addr;
111
    bzero(&addr, sizeof(struct sockaddr_in));
112
    addr.sin_family = AF_INET;
113
    addr.sin_addr.s_addr = inet_addr(ip);
114
    addr.sin_port = htons(atoi(port));
115
    int reuse = 1;
116
    int ret;
117
    ret = setsockopt(*pSockFd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
118
    ERROR_CHECK(ret, -1, "setsockopt");
119
    ret = bind(*pSockFd, (struct sockaddr *)&addr, sizeof(struct sockaddr_in));
120
    ERROR_CHECK(ret, -1, "bind");
121
    listen(*pSockFd, 10);
122
    return 0;
123
}
124
//epoll相关代码
125
int epollCtor(){
126
    int epfd = epoll_create(1);
127
    ERROR_CHECK(epfd,-1,"epoll_create");
128
    return epfd;
129
}
130
int epollAdd(int fd,int epfd){
131
    struct epoll_event event;
132
    event.events = EPOLLIN;
133
    event.data.fd = fd;
134
    int ret = epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&event);
135
    ERROR_CHECK(ret,-1,"epoll_ctl add");
136
    return 0;
137
}
138
int epollDel(int fd, int epfd){
139
    int ret = epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL);
140
    ERROR_CHECK(ret,-1,"epoll_ctl del");
141
    return 0;
142
}

文件的传输

14
1
//客户端
2
//...
3
    send(sockFd,filename,strlen(filename),0);
4
    ret = read(fd,buf,sizeof(buf))
5
    send(sockFd,buf,ret,0);
6
//...
7

8
//服务端
9
//...
10
    recv(netFd,filename,sizeof(filename),0);
11
    int fd = open(filename,O_RDONLY|O_CREAT,0666);
12
    ret = recv(netFd,buf,sizeof(buf),0);
13
    write(fd,buf,ret);
14
//...

这种写法会引入一个非常严重的问题，服务端在接收文件名，实际上并不知道有多长，所以它会试图把网络缓冲区的所有内容都读取出来，但是send底层基于的协议是TCP协议——这是一种流式协议。这样的情况下，服务端没办法区分到底是哪些部分是文件名而哪些部分是文件内容。完全可能会出现服务端把文件名和文件内容混杂在一起的情况，这种就是所谓的"粘包"问题。

所以我们要做的事情是在应用层上构建一个私有协议，这个协议的目的是规定TCP发送和接收的实际长度从而确定单个消息的边界。

进程池1.0

客户端

58
1
//client.c
2
typedef struct train_s
3
{
4
    int dataLength;
5
    char buf[1000];
6
} train_t;
7
int recvFile(int netFd);
8
int recvn(int netFd,void* pstart,int len);
9
int main(int argc, char *argv[])
10
{
11
    ARGS_CHECK(argc, 3);
12
    int sockFd = socket(AF_INET, SOCK_STREAM, 0);
13
    struct sockaddr_in addr;
14
    bzero(&addr, sizeof(struct sockaddr_in));
15
    addr.sin_family = AF_INET;
16
    addr.sin_addr.s_addr = inet_addr(argv[1]);
17
    addr.sin_port = htons(atoi(argv[2]));
18
    int ret = connect(sockFd, (struct sockaddr *)&addr, sizeof(struct sockaddr_in));
19
    ERROR_CHECK(ret, -1, "connect");
20
    recvFile(sockFd);
21
    close(sockFd);
22
    return 0;
23
}
24
int recvFile(int netFd)
25
{
26
    train_t t;
27
    bzero(&t,sizeof(t));
28
    //先接收文件名长度
29
    recvn(netFd, &t.dataLength, sizeof(int));
30
    //再接收文件名
31
    recvn(netFd, t.buf, t.dataLength);
32
    //接收方创建一个同名文件
33
    int fd = open(t.buf, O_WRONLY | O_CREAT, 0666);
34
    ERROR_CHECK(fd, -1, "open");
35
    while (1)
36
    {
37
        bzero(&t,sizeof(t));
38
        recvn(netFd,&t.dataLength,sizeof(int));
39
        if(0 == t.dataLength){
40
            break;
41
        }
42
        recvn(netFd,t.buf,t.dataLength);
43
        write(fd,t.buf,t.dataLength);
44
    }
45
    close(fd);
46
}
47
int recvn(int netFd,void* pstart,int len)
48
{
49
    int total=0;
50
    int ret;
51
    char *p=(char*)pstart;
52
    while(total<len)
53
    {
54
        ret=recv(netFd,p+total,len-total,0);
55
        total+=ret;//每次接收到的字节数加到total上
56
    }
57
    return 0;
58
}

服务端

189
1
//main.c
2
int main(int argc, char *argv[]){
3
    //./server 192.168.135.132 5678 10
4
    ARGS_CHECK(argc,4);
5
    int workerNum = atoi(argv[3]);
6
    processData_t *workerList = (processData_t *)calloc(sizeof(processData_t),workerNum);
7
    makeChild(workerList,workerNum);
8
    int sockFd;
9
    tcpInit(argv[1],argv[2],&sockFd);
10
    int epfd = epollCtor();
11
    epollAdd(sockFd,epfd);
12
    for(int i = 0;i < workerNum; ++i){
13
        epollAdd(workerList[i].pipeFd,epfd);
14
    }
15
    int listenSize = workerNum+1;//socket+每个进程pipe的读端
16
    struct epoll_event * readylist = (struct epoll_event *)calloc(listenSize,sizeof(struct epoll_event));
17
    while(1){
18
        int readynum = epoll_wait(epfd,readylist,listenSize,-1);
19
        for(int i = 0;i < readynum; ++i){
20
            if(readylist[i].data.fd == sockFd){
21
                puts("accept ready");
22
                int netFd = accept(sockFd,NULL,NULL);
23
                for(int j = 0;j < workerNum; ++j){
24
                    if(workerList[j].status == FREE){
25
                        printf("No. %d worker gets his job, pid = %d\n", j, workerList[j].pid);
26
                        sendFd(workerList[j].pipeFd, netFd);
27
                        workerList[j].status = BUSY;
28
                        break;
29
                    }
30
                }
31
                close(netFd);
32
            }
33
            else{
34
                puts("One worker finish his task!");
35
                int j;
36
                for(j = 0;j < workerNum;++j){
37
                    if(workerList[j].pipeFd == readylist[i].data.fd){
38
                        pid_t pid;
39
                        int ret = recv(workerList[j].pipeFd,&pid,sizeof(pid_t),0);
40
                        printf("No. %d worker finish, pid = %d\n",j,pid);
41
                        workerList[j].status = FREE;
42
                        break;
43
                    }
44
                }
45
            }
46
        }
47
    }
48
}
49
//worker.c
50
int makeChild(processData_t *pProcssData, int processNum)
51
{
52
    pid_t pid;
53
    for (int i = 0; i < processNum; ++i)
54
    {
55
        int pipeFd[2];
56
        socketpair(AF_LOCAL,SOCK_STREAM,0,pipeFd);
57
        pid = fork();
58
        if (pid == 0)
59
        {
60
            close(pipeFd[0]);
61
            handleEvent(pipeFd[1]);
62
        }
63
        close(pipeFd[1]);
64
        printf("pid = %d, pipefd[0] = %d\n", pid, pipeFd[0]);
65
        pProcssData[i].pid = pid;
66
        pProcssData[i].status = FREE;
67
        pProcssData[i].pipeFd = pipeFd[0];
68
    }
69
    return 0;
70
}
71
void handleEvent(int pipeFd)
72
{
73
    int netFd;
74
    while(1){
75
        recvFd(pipeFd,&netFd);
76
        transFile(netFd);
77
        close(netFd);
78
        pid_t pid = getpid();
79
        send(pipeFd,&pid,sizeof(pid_t),0);        
80
    }
81
}
82
//tcpInit.c
83
int tcpInit(char *ip, char *port, int *pSockFd)
84
{
85
    *pSockFd = socket(AF_INET, SOCK_STREAM, 0);
86
    struct sockaddr_in addr;
87
    bzero(&addr, sizeof(struct sockaddr_in));
88
    addr.sin_family = AF_INET;
89
    addr.sin_addr.s_addr = inet_addr(ip);
90
    addr.sin_port = htons(atoi(port));
91
    int reuse = 1;
92
    int ret;
93
    ret = setsockopt(*pSockFd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
94
    ERROR_CHECK(ret, -1, "setsockopt");
95
    ret = bind(*pSockFd, (struct sockaddr *)&addr, sizeof(struct sockaddr_in));
96
    ERROR_CHECK(ret, -1, "bind");
97
    listen(*pSockFd, 10);
98
    return 0;
99
}
100
//epollFunc.c
101
int epollCtor(){
102
    int epfd = epoll_create(1);
103
    ERROR_CHECK(epfd,-1,"epoll_create");
104
    return epfd;
105
}
106
int epollAdd(int fd,int epfd){
107
    struct epoll_event event;
108
    event.events = EPOLLIN;
109
    event.data.fd = fd;
110
    int ret = epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&event);
111
    ERROR_CHECK(ret,-1,"epoll_ctl add");
112
    return 0;
113
}
114
int epollDel(int fd, int epfd){
115
    int ret = epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL);
116
    ERROR_CHECK(ret,-1,"epoll_ctl del");
117
    return 0;
118
}
119
//sendFd.c
120
int sendFd(int pipeFd, int fdToSend){
121
    struct msghdr hdr;
122
    bzero(&hdr,sizeof(struct msghdr));
123
    struct iovec iov[1];
124
    char buf[] = "Hello";
125
    iov[0].iov_base = buf;
126
    iov[0].iov_len = 5;
127
    hdr.msg_iov = iov;
128
    hdr.msg_iovlen = 1;
129
    struct cmsghdr *pcmsghdr = (struct cmsghdr *)calloc(1,sizeof(CMSG_LEN(sizeof(int))));
130
    pcmsghdr->cmsg_len = CMSG_LEN(sizeof(int));
131
    //控制信息的数据部分只有int类型的文件描述符
132
    pcmsghdr->cmsg_level = SOL_SOCKET;
133
    pcmsghdr->cmsg_type = SCM_RIGHTS;//SCM->socket-level control message
134
    //表示在socket层传递的是访问权力，这样接受进程就可以访问对应文件对象了
135
    *(int *)CMSG_DATA(pcmsghdr) = fdToSend;
136
    //数据部分是文件描述符
137
    hdr.msg_control = pcmsghdr;
138
    hdr.msg_controllen = CMSG_LEN(sizeof(int));
139
    int ret = sendmsg(pipeFd,&hdr,0);
140
    ERROR_CHECK(ret,-1,"sendmsg");
141
}
142
int recvFd(int pipeFd, int *pFd){
143
    struct msghdr hdr;
144
    bzero(&hdr,sizeof(struct msghdr));
145
    struct iovec iov[1];
146
    char buf[6] = {0};
147
    iov[0].iov_base = buf;
148
    iov[0].iov_len = 5;
149
    hdr.msg_iov = iov;
150
    hdr.msg_iovlen = 1;
151
    struct cmsghdr *pcmsghdr = (struct cmsghdr *)calloc(1,sizeof(struct cmsghdr));
152
    pcmsghdr->cmsg_len = CMSG_LEN(sizeof(int));
153
    //控制信息的数据部分只有int类型的文件描述符
154
    pcmsghdr->cmsg_level = SOL_SOCKET;
155
    pcmsghdr->cmsg_type = SCM_RIGHTS;//SCM->socket-level control message
156
    hdr.msg_control = pcmsghdr;
157
    hdr.msg_controllen = CMSG_LEN(sizeof(int));
158
    int ret = recvmsg(pipeFd,&hdr,0);
159
    ERROR_CHECK(ret,-1,"recvmsg");
160
    *pFd = *(int *)CMSG_DATA(pcmsghdr);
161
    return 0;
162
}
163
//transFile.c
164
int transFile(int netFd){
165
    train_t t = {5,"file2"};
166
    send(netFd,&t,4+5,MSG_NOSIGNAL);
167
    int fd = open(t.buf,O_RDONLY);
168
    ERROR_CHECK(fd,-1,"open");
169
    bzero(&t,sizeof(t));
170
    while(1){
171
        t.dataLength = read(fd,t.buf,sizeof(t.buf));
172
        ERROR_CHECK(t.dataLength,-1,"read");
173
        if(t.dataLength != sizeof(t.buf)){
174
            printf("t.dataLength = %d\n",t.dataLength);
175
        }
176
        if(t.dataLength == 0){
177
            bzero(&t,sizeof(t));
178
            send(netFd,&t,4,MSG_NOSIGNAL);
179
            break;
180
        }
181
        int ret = send(netFd,&t,sizeof(int)+t.dataLength,MSG_NOSIGNAL);
182
        if(ret == -1){
183
            perror("send");
184
            break;
185
        }
186
    }
187
    close(fd);
188
    return 0;
189
}

进程池的其他功能

进度条显示

30
1
//服务端略
2
//下面是客户端
3
    //...
4
    off_t fileSize;
5
    bzero(&t,sizeof(t));
6
    recvn(netFd,&t.dataLength, sizeof(int));
7
    recvn(netFd,&fileSize,t.dataLength);
8
    printf("fileSize = %ld\n", fileSize);
9
    off_t doneSize = 0;
10
    off_t lastSize = 0;
11
    off_t slice = fileSize/100;
12
    int percentage = 0;
13
    while (1)
14
    {
15
        bzero(&t,sizeof(t));
16
        recvn(netFd,&t.dataLength,sizeof(int));
17
        if(0 == t.dataLength){
18
            break;
19
        }
20
        doneSize += t.dataLength;
21
        if(doneSize-lastSize >= slice){
22
            printf("%5.2lf%%\r", 100.0*doneSize/fileSize);  
23
            fflush(stdout);
24
            lastSize = doneSize;
25
        } 
26
        recvn(netFd,t.buf,t.dataLength);
27
        write(fd,t.buf,t.dataLength);
28
    }
29
    printf("100.00%%\n");
30
    //...

零拷贝、sendfile和splice

目前我们传输文件的时候是采用read和send来组合完成，这种当中的数据流向是怎么样的呢？首先打开一个普通文件，数据会从磁盘通过DMA设备传输到内存，即文件对象当中的内核缓冲区部分，然后调用read数据会从内核缓冲区拷贝到一个用户态的buf上面（buf是read函数的参数），接下来调用send，就将数据拷贝到了网络发送缓存区，最终实现了文件传输。但是实际上这里涉及了大量的不必要的拷贝操作。

如何减少从内核文件缓冲区到用户态空间的拷贝呢？解决方案就是使用mmap系统调用直接建立文件和用户态空间buf的映射。这样的话数据就减少了一次拷贝。在非常多的场景下都会使用mmap来减少拷贝次数，典型的就是使用图形的应用去操作显卡设备的显存。除此以外，这种传输方式也可以减少由于系统调用导致的CPU用户态和内核态的切换次数。

108
1
//客户端
2
int recvFile(int netFd)
3
{
4
    train_t t;
5
    bzero(&t,sizeof(t));
6
    //先接收文件名长度
7
    recvn(netFd, &t.dataLength, sizeof(int));
8
    //再接收文件名
9
    recvn(netFd, t.buf, t.dataLength);
10
    //接收方创建一个同名文件
11
    int fd = open(t.buf, O_RDWR| O_CREAT, 0666);
12
    ERROR_CHECK(fd, -1, "open");
13
    off_t fileSize;
14
    bzero(&t,sizeof(t));
15
    recvn(netFd,&t.dataLength, sizeof(int));
16
    recvn(netFd,&fileSize,t.dataLength);
17
    printf("fileSize = %ld\n", fileSize);
18
    /* case 1 分批接收
19
    off_t doneSize = 0;
20
    off_t lastSize = 0;
21
    off_t slice = fileSize/100;
22
    int percentage = 0;
23
    while (1)
24
    {
25
        bzero(&t,sizeof(t));
26
        recvn(netFd,&t.dataLength,sizeof(int));
27
        if(0 == t.dataLength){
28
            break;
29
        }
30
        doneSize += t.dataLength;
31
        if(doneSize-lastSize >= slice){
32
            printf("%5.2lf%%\r", 100.0*doneSize/fileSize);  
33
            fflush(stdout);
34
            lastSize = doneSize;
35
        } 
36
        recvn(netFd,t.buf,t.dataLength);
37
        write(fd,t.buf,t.dataLength);
38
    }
39
    */
40
    //case 1一次性接收完 注意此时客户端需要修改
41
    ftruncate(fd,fileSize);
42
    //前面open的权限需要改成O_RDWR
43
    char *p = (char *)mmap(NULL,fileSize,PROT_READ|PROT_WRITE,MAP_SHARED,fd,0);
44
    ERROR_CHECK(p,MAP_FAILED,"mmap");
45
    recvn(netFd,p,fileSize);
46
    printf("100.00%%\n");
47
    munmap(p,fileSize);
48
    close(fd);
49
}
50
//服务端 分批发送
51
int transFile(int netFd){//mmap_multi
52
    train_t t = {5,"file2"};
53
    send(netFd,&t,4+5,MSG_NOSIGNAL);
54
    int fd = open(t.buf,O_RDONLY);
55
    ERROR_CHECK(fd,-1,"open");
56
    struct stat statbuf;
57
    int ret = fstat(fd,&statbuf);
58
    bzero(&t,sizeof(t));
59
    t.dataLength = sizeof(statbuf.st_size);
60
    memcpy(t.buf,&statbuf.st_size,t.dataLength);
61
    send(netFd,&t,sizeof(off_t)+4, MSG_NOSIGNAL);
62
    char *p = (char *)mmap(NULL,statbuf.st_size,PROT_READ,MAP_SHARED,fd,0);
63
    ERROR_CHECK(p,(void *)-1, "mmap");
64
    off_t total = 0;
65
    while(total < statbuf.st_size){
66
        if(statbuf.st_size - total > sizeof(t.buf)){
67
            t.dataLength = sizeof(t.buf);
68
        }
69
        else{
70
            t.dataLength = statbuf.st_size - total;
71
        }
72
        memcpy(t.buf,p+total,t.dataLength);
73
        total += t.dataLength;
74
        int ret = send(netFd,&t,sizeof(int)+t.dataLength,MSG_NOSIGNAL);
75
        if(ret == -1){
76
            perror("send");
77
            break;
78
        }
79
    }
80
    //发送结束标志
81
    t.dataLength = 0;
82
    send(netFd,&t,4,MSG_NOSIGNAL);
83
    munmap(p,statbuf.st_size);
84
    close(fd);
85
    return 0;
86
}
87
//服务端 一次性发送
88
int transFile(int netFd){// mmap_once
89
    train_t t = {5,"file2"};
90
    send(netFd,&t,4+5,MSG_NOSIGNAL);
91
    int fd = open(t.buf,O_RDONLY);
92
    ERROR_CHECK(fd,-1,"open");
93
    struct stat statbuf;
94
    int ret = fstat(fd,&statbuf);
95
    bzero(&t,sizeof(t));
96
    t.dataLength = sizeof(statbuf.st_size);
97
    memcpy(t.buf,&statbuf.st_size,t.dataLength);
98
    send(netFd,&t,sizeof(off_t)+4, MSG_NOSIGNAL);
99
    char *p = (char *)mmap(NULL,statbuf.st_size,PROT_READ,MAP_SHARED,fd,0);
100
    ERROR_CHECK(p,(void *)-1, "mmap");
101
    send(netFd,p,statbuf.st_size,MSG_NOSIGNAL);
102
    //发送结束标志
103
    t.dataLength = 0;
104
    send(netFd,&t,4,MSG_NOSIGNAL);
105
    munmap(p,statbuf.st_size);
106
    close(fd);
107
    return 0;
108
}

使用mmap系统调用只能减少数据从磁盘文件的文件对象到用户态空间的拷贝，但是依然无法避免从用户态到内核已连接套接字的拷贝（因为网络设备文件对象不支持mmap）。sendfile系统调用可以解决这个问题，它可以使数据直接在内核中传递而不需要经过用户态空间，调用sendfile系统调用可以直接将磁盘文件的文件对象的数据直接传递给已连接套接字文件对象，从而直接发送到网卡设备之上（在内核的底层实现中，实际上是让内核磁盘文件缓冲区和网络缓冲区对应同一片物理内存）。

2
1
#include <sys/sendfile.h>
2
ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);

使用sendfile的时候要特别注意，out_fd一般只能填写网络套接字的描述符，表示写入的文件描述符，in_fd一般是一个磁盘文件，表示读取的文件描述符。从上述的需求可以得知，sendfile只能用于发送文件方的零拷贝实现，无法用于接收方，并且发送文件的大小上限通常是2GB。

18
1
int transFile(int netFd){
2
    train_t t = {5,"file2"};
3
    send(netFd,&t,4+5,MSG_NOSIGNAL);
4
    int fd = open(t.buf,O_RDONLY);
5
    ERROR_CHECK(fd,-1,"open");
6
    struct stat statbuf;
7
    int ret = fstat(fd,&statbuf);
8
    bzero(&t,sizeof(t));
9
    t.dataLength = sizeof(statbuf.st_size);
10
    memcpy(t.buf,&statbuf.st_size,t.dataLength);
11
    send(netFd,&t,sizeof(off_t)+4, MSG_NOSIGNAL);
12
    //发送结束标志
13
    sendfile(netFd,fd,NULL,statbuf.st_size);
14
    t.dataLength = 0;
15
    send(netFd,&t,4,MSG_NOSIGNAL);
16
    close(fd);
17
    return 0;
18
}

考虑到sendfile只能将数据从磁盘文件发送到网络设备中，那么接收方如何在避免使用mmap的情况下使用零拷贝技术呢？一种方式就是采用管道配合splice的做法。splice系统调用可以直接将数据从内核管道文件缓冲区发送到另一个内核文件缓冲区，也可以反之，将一个内核文件缓冲区的数据直接发送到内核管道缓冲区中。所以只需要在内核创建一个匿名管道，这个管道用于本进程中，在磁盘文件和网络文件之间无拷贝地传递数据。

2
1
ssize_t splice(int fd_in, loff_t *off_in, int fd_out,
2
                      loff_t *off_out, size_t len, unsigned int flags);

13
1
//...
2
    recvn(netFd,&t.dataLength, sizeof(int));
3
    recvn(netFd,&fileSize,t.dataLength);
4
    printf("fileSize = %ld\n", fileSize);
5
    int pipefds[2];
6
    pipe(pipefds);
7
    int total = 0;
8
    while(total < fileSize){
9
        int ret = splice(netFd,NULL,pipefds[1],NULL,4096,SPLICE_F_MORE);
10
        total += ret;
11
        splice(pipefds[0],NULL,fd,NULL,ret,SPLICE_F_MORE);
12
    }
13
//...

进程池的退出

进程池的简单退出

进程池的简单退出要实现功能很简单，就是让父进程收到信号之后，再给每个子进程发送信号使其终止，这种实现方案只需要让父进程在一个目标信号（通常是10信号SIGUSR1）的过程给目标子进程发送信号即可。

在实现的过程需要注意的是signal函数和fork函数之间调用顺序，因为父进程会修改默认递送行为，而子进程会执行默认行为，所以fork应该要在signal的之后调用。

19
1
processData_t *workerList;//需要改成全局变量
2
int workerNum;
3
void sigFunc(int signum){
4
    printf("signum = %d\n", signum);
5
    for(int i = 0; i < workerNum; ++i){
6
        kill(workerList[i].pid,SIGUSR1);
7
    }
8
    for(int i = 0; i < workerNum; ++i){
9
        wait(NULL);
10
    }
11
    puts("process pool is over!");
12
    exit(0);
13
}
14
int main(){
15
    //..
16
    makeChild(workerList,workerNum);
17
    signal(SIGUSR1,sigFunc);
18
    //注意fork和signal的顺序
19
}

使用管道通知工作进程终止

采用信号就不可避免要使用全局变量，因为信号处理函数当中只能存储有限的信息，有没有办法避免全局的进程数量和进程数组呢？一种解决方案就是采取“异步拉起同步”的策略：虽然还是需要创建一个管道全局变量，但是该管道只用于处理进程池退出，不涉及其他的进程属性。这个管道的读端需要使用IO多路复用机制管理起来，而当信号产生之后，主进程递送信号的时候会往管道中写入数据，此时可以依靠epoll的就绪事件，在事件处理中来完成退出的逻辑。

24
1
int pipeFd[2];
2
void sigFunc(int signum){
3
    printf("signum = %d\n",signum);
4
    write(pipeFd[1],"1",1);
5
}
6
int main(){
7
    //...
8
    pipe(pipeFd);
9
    epollAdd(pipeFd[0],epfd);
10
    //...
11
    //...epoll就绪事件处理
12
    else if(readylist[i].data.fd == pipeFd[0]){
13
                for(int j = 0; j < workerNum; ++j){
14
                    kill(workerList[j].pid,SIGINT);
15
                    puts("send signal to worker!");
16
                }
17
                for(int j = 0; j < workerNum; ++j){
18
                    wait(NULL);
19
                }
20
                printf("Parent process exit!\n");
21
                exit(0);
22
            }
23
    //...
24
}

进程池的优雅退出

上述的退出机制存在一个问题，就是即使工作进程正在传输文件中，父进程也会通过信号将其终止。如何实现进程池在退出的时候，子进程要完成传输文件的工作之后才能退出呢？

一种典型的方案是使用sigprocmask在文件传输的过程中设置信号屏蔽字，这样可以实现上述的机制。

另一种方案就是调整sendFd的设计，每个工作进程在传输完文件之后总是循环地继续下一个事件，而在每个事件处理的开始，工作进程总是会调用recvFd来使自己处于阻塞状态直到有事件到达。我们可以对进程池的终止作一些调整：用户发送信号给父进程表明将要退出进程池；随后父进程通过sendFd给所有的工作进程发送终止的信息，工作进程在完成了一次工作任务了之后就会recvFd收到进程池终止的信息，然后工作进程就可以主动退出；随着所有的工作进程终止，父进程亦随后终止，整个进程池就终止了。

60
1
int sendFd(int pipeFd, int fdToSend, int exitFlag){
2
    struct msghdr hdr;
3
    bzero(&hdr,sizeof(struct msghdr));
4
    struct iovec iov[1];
5
    iov[0].iov_base = &exitFlag;
6
    iov[0].iov_len = sizeof(int);
7
    hdr.msg_iov = iov;
8
    hdr.msg_iovlen = 1;
9
//...
10
}
11
int recvFd(int pipeFd, int *pFd, int *exitFlag){
12
    struct msghdr hdr;
13
    bzero(&hdr,sizeof(struct msghdr));
14
    struct iovec iov[1];
15
    iov[0].iov_base = exitFlag;
16
    iov[0].iov_len = sizeof(int);
17
    hdr.msg_iov = iov;
18
    hdr.msg_iovlen = 1;
19
//.....
20
}
21
void handleEvent(int pipeFd)
22
{
23
    int netFd;
24
    while(1){
25
        int exitFlag;
26
        recvFd(pipeFd,&netFd,&exitFlag);
27
        if(exitFlag == 1){
28
            puts("I am closing!");
29
            exit(0);
30
        }
31
        //...
32
    }
33
}
34
//... epoll
35
        for(int i = 0;i < readynum; ++i){
36
            if(readylist[i].data.fd == sockFd){
37
                puts("accept ready");
38
                int netFd = accept(sockFd,NULL,NULL);
39
                for(int j = 0;j < workerNum; ++j){
40
                    if(workerList[j].status == FREE){
41
                        printf("No. %d worker gets his job, pid = %d\n", j, workerList[j].pid);
42
                        sendFd(workerList[j].pipeFd, netFd, 0);
43
                        workerList[j].status = BUSY;
44
                        break;
45
                    }
46
                }
47
                close(netFd);
48
            }
49
            else if(readylist[i].data.fd == exitpipeFd[0]){
50
                for(int j = 0; j < workerNum; ++j){
51
                    puts("set exitFlag to worker!");
52
                    sendFd(workerList[j].pipeFd,0,1);
53
                }
54
                for(int j = 0; j < workerNum; ++j){
55
                    wait(NULL);
56
                }
57
                printf("Parent process exit!\n");
58
                exit(0);
59
            }
60
//....

线程池

线程池的实现