进程池和线程池

1, 进程池

假设: 我们结合学过的文件操作、网络通信、以及进程和线程的知识, 实现一个基本的文件下载服务器模型, 我们需要做那些准备工作, 或者说我们怎么设计整个数据通信逻辑。

设计一个服务器中, 在大多数情况下, 都会有很多连接的频繁接入和断开, 如果在通信模型中, 我们让一个进程既处理连接接入,又处理业务逻辑, 这样设计在当下应用领域无疑是很低效的。(它既无法做到有效解耦, 增加代码书写的麻烦, 增加并行逻辑设计困难, 又无法有效利用多核性能, 伴随性能瓶颈)

xxxxxxxxxx
2
1
// 当然, 如果是单核心系统, 或者基于单个进程或者进程设计的基于事件驱动(基于select或者epoll管控文件对象, 监听事件, 分发业务逻辑)的服务器模型也是有其非常重要的独特之处的。(书写代码麻烦, 内存利用性能有一定优势, 在并发量比较低低的服务器中调度少性能高)
2
// 当然现在更好的设计是: 我们下面的池化+事件驱动的设计, 让主进程/线程只负责基于事件驱动分发任务, 池中进程/线程执行具体的业务代码。

一个良好的架构需要满足许多侧面的要求，其中最基本的要求是可维护性和性能:

• 可维护性: 是指应用程序对开发者应该足够友好，开发和维护的程序员可以很快速的就能理解程序架构并进行后续开发。为了提供可维护性，项目各个部分的功能应当彼此分离。

• 性能: 是指应用程序应当充分利用操作系统资源，并且减少不要资源消耗。在多进程程序中， 一种非常浪费资源的操作就是创建和销毁进程。

所以我们无论从代码开发难易上还是性能提升上, 都有必要利用多进程或者多线程来实现业务逻辑和任务管控逻辑分离。

我么可以维护一个主进程只负责接收用户请求, 并把接收到的用户请求分配到不同进程上来处理。但是如果我们对任务进程不做任何限制和管理, 随着任务的到达开始一个进程处理任务, 任务处理结束之后结束这个进程, 这样处理是极其不好的, 因为进程的频繁创建和销毁是一份很大的软硬件开销。

而借用池化思想可以有效的避免这个问题(很多地方都有池化设计) , 我们维护一个进程的池子, 包含多个进程, 当有任务到来, 把任务交给空闲的进程执行, 当任务执行完毕, 并且没有任务可执行, 让进程休眠。这种池化的思想可以显著减少创建和销毁进程的软硬件开销，进而提高程序的执行效率。

• 主进程负责接收用户请求, 获得连接的文件描述符, 并且维护所有进程池中进程的状态, 以方便把文件描述符对象交给空闲的池中进程来和客户端直接交互。

• 进程池中进程, 在被主进程唤醒, 接收到对应的连接文件描述符对象, 按照需求读取磁盘文件, 并把读取的磁盘文件发送给客户端。

• 客户端建立连接, 接收返回的文件。

进程池和线程池

以上述服务模型为例, 我们把进程池换成线程池, 逻辑上也是相同的, 那么在我们设计的时候, 怎么确定使用进程池设计还是线程池设计。

进程池设计:

• 每个进程有独立的内存空间, 增加进程间的隔离性。一个进程崩溃不会影响到其它进程。

• 进程间存在隔离性, 这种解耦促使业务逻辑方便书写。

• 进程的创建和销毁比线程开销大, 占用的内存空间也比线程大。

• 上下文的调度切换时间长。

• 适合并发量低(IO请求数少), 业务复杂(CPU密集), 任务执行事件长的系统设计。

线程池设计:

• 线程之间共享资源, 隔离新差, 一个线程极容易影响到另一个线程(数据同步和一致性)。

• 但是隔离性差, 使得线程间通信比进程间通信要更方便。

• 线程较轻量，创建和销毁的开销较小。

• 适合并发量高(I/O密集型),内存使用要求高, 业务简单, 可以大量快速、轻量级任务处理的场景。

1.1 第一版

1.1.1 设计逻辑

• head.h

  • 头文件的引入

  • 函数定义

  • 结构体和类型定义

• main.c

  • 调用pool初始化进程池; 并让pool启动线程池中线程worker

  • 调用tcpInit初始化socket对IP&端口监听

  • 调用epoll监听客户端连接进来, 当连接进来, 调用pool把连接交给进程池中进程worker处理, 修改进程状态为忙

  • 调用epoll监听和进程池中线程, 等待进程池中线程处理完客户端请求, 通知main, 把其状态改为空闲

• pool.c

  • 根据main的要求, 启动对应数量的进程worker

  • 把启动的进程及其状态记录到数组中, 以供main监控worker结束任务后, 向main发起的通信(由忙置为闲的通信)

  • 当main有客户端连接过来, 通知pool, pool从空闲的进程中选取一个进程执行和客户端通信

• worker.c

  • 被pool初始化, 并启动

  • 等待接收main获取的客户端连接

  • 接收到main让pool通知过来的客户端连接, 拿到连接, 与指定的客户端通信, 通信完毕关闭客户端连接

  • 通信结束, 向main发信息, 告诉其自己应该由忙状态更改为闲状态

  • 等待main分发过来新的客户端

• tcpInit.c

  • 初始化ip和端口的监控

  • 避免大量的tcp连接代码写在main中, 为了解耦而存在

• epoll.c

  • epoll的操作相关

  • 避免大量epoll添加文件描述符监听代码在main中, 为了解耦而存在

• localSocket.c

  • 当main监听到一个客户端连接, 需要把这个客户端连接对象交给工作的worker, 这涉及到进程间通信

  • 这个进程间通信不是一个简单的问题, 因为这个进程间通信不是传递字符串,数字等简单的东西,而是要传递一个文件描述符, 而这个传送的文件描述符还要具有共享文件对象的能力

  • 这个地方就要用到一个特殊的功能强大的本地通信socket, 用于main和worker间进行一个可以最终共享文件对象的文件描述符的传递

1.1.2 socketpair

一般我们使用它的目的: 为了在两个进程之间, 传输一个文件对象的描述信息(可以让两个进程共享一个文件对象), 而不是单单只传输一个文件描述符数组的下标.

socketpair()函数用于创建一对互相连接的全双工通信socket。相比较普通的用于网络间不同主机通信的socket; socketpair函数创建的socket主要用于在同一台机器上的进程间通信(我们可以称其为本地socket)。 (man socketpair)

xxxxxxxxxx
10
1
#include <sys/types.h> 
2
#include <sys/socket.h>
3
// create a pair of connected sockets
4
int socketpair(
5
   int domain, // 指定socket使用的协议族, 本地通信我们使用: AF_LOCAL
6
   int type, // 指定socket的类型: SOCK_STREAM(TCP), SOCK_DGRAM(UDP)
7
   int protocol, // 指定协议, 默认设置0即可
8
   int sv[2] // 用于返回两个连接的socket描述符(等价与socket的fd), 父子进程可以通过这个文件描述符进行通信(任何发送到sv[0]的数据都可以从sv[1]读取，反之任何发送到sv[1]的数据都可以从sv[0]读取。)
9
);
10
// 返回值: 成功返回0, 失败返回-1

• 全双工通信: (对于socketpair函数构建int sv[2]文件描述符数组)

xxxxxxxxxx
2
1
// 不同与可以进程间通信的匿名管道, 数据都必须是从pipe[1]写入管道，然后从pipe[0]读出来。
2
// socketpair的创建的通信端点, 可以让一个进程从fd[1]写,另一进程从fd[0]读; 也可以让一个进程从fd[0]写,另一进程从fd[1]读。

如果你想通过socketpair函数实现两个本地进程间的文件对象描述符的传输，除了需要socketpair函数创建通信的端点, 还需要借助sendmsg函数和recvmsg函数来实现具体的数据传输。 (man sendmsg) (man recvmsg)

xxxxxxxxxx
9
1
#include <sys/types.h>
2
#include <sys/socket.h>
3
// send a message on a socket
4
ssize_t sendmsg(
5
   int sockfd, // socket文件描述符(对应上面socketpair中创建的文件描述符sv[]数组)
6
   const struct msghdr *msg, // 要发送的详细信息
7
   int flags // 发送行为, 默认0
8
);
9
// 返回值: 成功返回一个ssize_t类型的值，表示发送的字节数。失败-1。

xxxxxxxxxx
9
1
#include <sys/types.h>
2
#include <sys/socket.h>
3
// receive a message from a socket
4
ssize_t recvmsg(
5
   int sockfd, // socket文件描述符(对应上面socketpair中创建的文件描述符sv[]数组)
6
   struct msghdr *msg,  // 要接收的详细信息
7
   int flags // 接收行为, 默认0
8
);
9
// 返回值: 成功返回一个ssize_t类型的值，表示接收的字节数。失败-1。

xxxxxxxxxx
9
1
struct msghdr {
2
   void         *msg_name;//记录消息地址, 填充NULL交给系统处理
3
   socklen_t     msg_namelen;//地址长度, msg_name为NULL时系统自动填充
4
   struct iovec *msg_iov;//iovec类型的数组, 每个iovec类型都是一份数据 (即:这个数组可以携带多份数据)(比较灵活)
5
   size_t        msg_iovlen;//上个参数iovec数组中的长度
6
   void         *msg_control;//(本质是一个cmsghdr类型指针) 用于包含控制信息(传输文件对象就要用到它)
7
   size_t        msg_controllen;//控制信息缓冲区的长度 (比较复杂)
8
   int           msg_flags;//消息的标志, 系统填充
9
};

xxxxxxxxxx
4
1
struct iovec{
2
   void *iov_base;//一个数据的指针
3
   size_t iov_len;//数据的长度(字节)
4
}

xxxxxxxxxx
16
1
struct cmsghdr {
2
   socklen_t cmsg_len;//数据长度(字节): 头部信息(cmsg_len/cmsg_level/cmsg_type) + cmsg_data长度: 需要计算
3
   int       cmsg_level;//协议类型: (socket使用)SOL_SOCKET
4
   int       cmsg_type;//消息的类型: (传输文件描述符) SCM_RIGHTS 
5
   unsigned char cmsg_data[]//存放具体数据
6
};
7
ps1: cmsg_data数组的长度用于存放额外的数据,可以根据需求的变化而变化(比如现在我们使用它, 准备存放文件描述符)
8
ps2: 我们可以通过CMSG_LEN这个宏用于计算包括cmsg_data在内的完整cmsghdr结构的长度。而使用CMSG_LEN的时候, 其参数只需要传入cmsg_data的长度
9
// 比如: cmsg_data存储文件描述符(文件描述符int类型), 通过CMSG_LEN(sizeof(int))得到的长度就是整个cmsghdr结构体的长度(cmsg_len + cmsg_level + cmsg_type + cmsg_data)
10
ps3: CMSG_DATA这个宏用于获取指向cmsghdr结构中数据部分cmsg_data的指针
11
// 比如: 如果我们准备用cmsghdr传输文件描述符, 就要先通过CMSG_DATA获取指向cmsg_data部分起始位置的指针
12
//      文件描述符指针: *netfd;
13
//      struct cmsghdr *pcms = (struct cmsghdr *) malloc(  CMSG_LEN(sizeof(int))  )
14
//      void *addr = CMSG_DATA(pcms);
15
//      int * p_fd = (int *)addr;
16
//      *p_fd = *netfd

代码示例

xxxxxxxxxx
73
1
#include <55header.h>
2

3
int main(int argc,char*argv[])
4
{
5
int fd[2];
6
socketpair(AF_LOCAL, SOCK_STREAM, 0, fd);
7

8
if(fork() == 0){
9
   int fd_txt = open("1.txt", O_WRONLY);
10
   close(fd[0]);
11
   int socket_fd = fd[1];
12

13
   struct msghdr hdr;
14
   bzero(&hdr, sizeof(hdr));
15

16
   char *str = "hello";
17
   struct iovec vec[1];
18
   vec[0].iov_base = str;
19
   vec[0].iov_len = strlen(str);
20

21
   hdr.msg_iov = vec;
22
   hdr.msg_iovlen = 1;
23

24
   struct cmsghdr *pcms = (struct cmsghdr *)malloc(CMSG_LEN(sizeof(int)));
25
   pcms->cmsg_len = CMSG_LEN(sizeof(int));
26
   pcms->cmsg_level = SOL_SOCKET;
27
   pcms->cmsg_type = SCM_RIGHTS;
28
   void *addr = CMSG_DATA(pcms);
29
   int *p_fd = (int *)addr;
30
   *p_fd = fd_txt;
31

32
   hdr.msg_control = pcms;
33
   hdr.msg_controllen = CMSG_LEN(sizeof(int));
34

35
   sendmsg(socket_fd,&hdr, 0);
36

37
   printf("son 1.txt fd: %d \n", fd_txt);
38
}else{
39
   close(fd[1]);
40
   int socket_fd = fd[0];
41

42
struct msghdr hdr;
43
   bzero(&hdr, sizeof(hdr));
44

45
char buf[60] = {0};
46
   struct iovec vec[1];
47
   vec[0].iov_base = buf;
48
   vec[0].iov_len = sizeof(buf);
49

50
hdr.msg_iov = vec;
51
   hdr.msg_iovlen = 1;
52

53
struct cmsghdr *pcms = (struct cmsghdr *)malloc(CMSG_LEN(sizeof(int)));
54
   pcms->cmsg_len = CMSG_LEN(sizeof(int));
55
   pcms->cmsg_level = SOL_SOCKET;
56
   pcms->cmsg_type = SCM_RIGHTS;
57

58
hdr.msg_control = pcms;
59
   hdr.msg_controllen = CMSG_LEN(sizeof(int));
60

61
recvmsg(socket_fd,&hdr, 0);
62

63
void *addr = CMSG_DATA(pcms);
64
   int *fd = (int *)addr;
65

66
   wait(NULL);
67
   printf("main 1.txt fd: %d \n", *fd);
68
   printf("main str :%s \n", buf);
69

70
write(*fd, "world", 5);
71
}
72
return 0;
73
}

1.2.3 Makefile

在一个复杂项目中, 一般情况下我们需要通过多个.c文件, 相互协作, 共同编译出一个可执行文件运行, 那么就需要稍稍修改makefile的书写规则

xxxxxxxxxx
19
1
# makefile
2
# 定义一个srcs变量, 来代指: 使用wildcard函数获取当前目录下的所有.c文件
3
srcs:=$(wildcard *.c)
4
# 定义一个objs变量, 来代指: 使用patsubst函数讲srcs中所有的.c文件拓展名替换成.o文件
5
objs:=$(patsubst %.c,%.o, $(srcs))
6

7
# 编译 ' -c $^ '(源依赖项.c文件)  输出到 ' -o $@ '(目标.o)文件
8
%.o:%.c
9
    gcc -c $^ -o $@ -g
10
# mian文件依赖于所有的objs文件
11
# 把所有依赖项($^) 指定输出到当前目标(-o $@)(即:mian)
12
main:$(objs)
13
    gcc $^ -o $@ -lpthread
14

15
# 清理objs文件 清理mian文件
16
clean:
17
    $(RM) $(objs) main
18

19
rebuild: clean main

1.1.4 CODE

head

xxxxxxxxxx
54
1
#include <testfun.h>
2
// 定义:一个枚举类型, 分别代表进程`忙`和`空闲`
3
enum {
4
BUSY,
5
FREE
6
};
7
// 定义:用来追踪进程池中, 进程的信息和状态
8
typedef  struct worker_status_s{
9
int pid; // 进程ID
10
int status; // 进程的状态: 忙 or 空闲
11
int socket_local_fd; // socket本地通信文件描述符: 用于main和子进程通信
12
} worker_status_t;
13

14
// 根据指定数量初始化进程池
15
int initWorkerArr(
16
        worker_status_t *p, // 记录进程池中进程状态的数组(传入传出参数) 
17
        int num   // 进程池中:要创建的进程个数
18
        );
19

20
// 根据端口和IP构建服务端的网络监听
21
int initTcpSocket(
22
        int * socketfd,// socket的文件描述符(传入传出参数) 
23
        char *ip, // ip地址
24
        char *port // 端口
25
        );
26
// 给epoll添加要监听的文件描述符
27
int epoll_addfd(
28
      int epollfd, // epoll的文件描述符
29
      int filefd // 要监听的文件描述符
30
     );
31
// 把监听到的一个连接交给进程池中某个空闲进程
32
int toProcessPoolDealNet(
33
           int netfd, // accept获取的连接对象的文件描述符id
34
           worker_status_t *p, // 进程池连接数组
35
           int num // 进程池中进程个数
36
          );
37
// 启动一个工作进程
38
int startWorker(
39
      int socket_local_fd // 用以和父进程通信的本地socket文件描述符
40
     );
41
// 工作进程,从本地socket中读取main进程发过来的客户端连接对象
42
int read_net_fd(
43
      int socket_local_fd, //用以和main进程和工作进程通信的本地socket文件描述符
44
      int *netfd // 客户端连接对象的文件描述符指针
45
     );
46
// main进程accept获得的客户端连接对象发给工作进程
47
int write_net_fd(
48
       int socket_local_fd, // mian进程用来和工作进程通信的本地socket文件描述符
49
       int *netfd // main进程accept的客户端连接对象的文件描述符
50
      );
51
// 根据客户端连接文件对象和客户端通信
52
int netToClient(
53
      int netfd // 客户端连接对象的文件描述符
54
     );

Main

xxxxxxxxxx
54
1
#include "head.h"
2
int main()
3
{
4
// 初始化进程池
5
worker_status_t  workerArr[3];
6
// 进程池中的进程个数: 一倍~二倍cpu (一倍: cpu密集,  二倍: IO密集, 可以实际调整)
7
initWorkerArr(workerArr, 3);
8

9
// 初始化网络监听
10
int socketfd;
11
initTcpSocket(&socketfd, "192.168.106.129", "8080");
12

13
// 初始化epoll获得epoll文件描述符
14
int epollfd = epoll_create(1);
15
// 添加连接监听
16
epoll_addfd(epollfd, socketfd);
17

18
// 子进程完成任务， 要通知进程池:把忙状态改为闲状态
19
// 所以要监听， 子进程向父进程通信的本地socket， 如果本地socket就绪， 说明状态要变为闲
20
for(int i=0; i<3; i++){
21
   epoll_addfd(epollfd,workerArr[i].socket_local_fd);
22
}
23
// 循环获取网络连接
24
while(1){
25
   // 就绪集合
26
   struct epoll_event ready_set[1024];
27
   // 监听就绪
28
   int ready_num = epoll_wait(epollfd, ready_set, 1024, -1);
29

30
   // 遍历就绪集合
31
   for(int i=0; i<ready_num; i++){
32
       if(ready_set[i].data.fd == socketfd){
33
           // socket监听读就绪， 有连接进来
34
           int netfd = accept(socketfd, NULL, NULL);
35
           // 把监听到的连接交给进程池
36
           toProcessPoolDealNet(netfd, workerArr, 3);
37
           // 主进程不需要再维护连接（已经交给子进程了）
38
           // 关闭连接
39
           close(netfd);
40
       }else {
41
           // 某个字进程完成任务，子进程通过本地socket发回信息
42
           // 导致epoll监听对应本地socket就绪
43
           for(int j=0; j<3; j++){
44
               if(workerArr[j].socket_local_fd = ready_set[i].data.fd){
45
                   // workerArr[j]的子进程发来通信
46
                   // 由忙状态-》 闲状态
47
                   workerArr[j].status = FREE;
48
               }
49
           }
50
       }
51
   }
52
}
53
return 0;
54
}

pool

xxxxxxxxxx
52
1
#include "head.h"
2

3
// 根据指定数量初始化进程池
4
// 参数1: 追踪进程池中进程状态的数组(传入传出参数)
5
// 参数2: 进程池中进程个数
6
int initWorkerArr(worker_status_t *p, int num){
7

8
   for(int i=0; i<num; i++){
9
       // 使用socketpair初始化本地socket通信两端
10
       // 创建一个用于和子进程通信, 可以传输文件对象的本地socket
11
       int socket_fd[2];
12
       socketpair(AF_LOCAL, SOCK_STREAM, 0, socket_fd);
13

14
       int pid = fork();
15
       if(pid == 0){
16
           // 子进程
17
           // 关闭socket_fd[1], 保留socket_fd[0], 用于和主进程通信
18
           close(socket_fd[1]);
19
           // 启动当前子进程
20
           startWorker(socket_fd[0]);
21
       }else{
22
           // 父进程: 保存子进程pid,子进程状态, 和子进程通信的本地socket文件描述符
23
           p[i].pid = pid; 
24
           p[i].status = FREE;
25
           p[i].socket_local_fd = socket_fd[1];
26
           // 保留socket_fd[1], 关闭socket_fd[0]
27
           // 也可以:关闭socket_fd[1], 保留socket_fd[0]
28
           // 和子进程相对应的凑成一对即可
29
           close(socket_fd[0]);
30
       }
31
   }
32
   return 0;
33
}
34

35
// 把监听到的一个连接交给进程池中某个空闲进程
36
// 参数1：accept获取的网络连接对象的文件描述符
37
// 参数2：进程池数组
38
// 参数3：进程池大小
39
int toProcessPoolDealNet( int netfd,  worker_status_t *p, int num){
40
// 遍历进程池， 寻找空闲进程
41
for(int i=0; i<num; i++){
42
   if(p[i].status == FREE){
43
       // 把main进程accpet获得的客户端连接文件描述符对象
44
       // 通过本地socket 发送给工作进程
45
       write_net_fd(p[i].socket_local_fd, &netfd);
46

47
       // 把当前进程置为忙状态
48
       p[i].status = BUSY;
49
       break;
50
   }
51
}
52
}

worker

xxxxxxxxxx
36
1
#include "head.h"
2

3
int startWorker(int socket_local_fd){
4
while(1){
5
   // 读取主进程通过socketpair传过来的客户端连接对象的文件描述符
6
   // 客户端连接的对象的(主进程accept的获取)
7
   // 通过封装的read_net_fd()读取
8
   int netfd;
9
   read_net_fd(socket_local_fd, &netfd);
10
   printf("worker netfd: %d \n", netfd);
11

12
   // 根据获得的客户端连接对象的文件描述符和客户端通信
13
   netToClient(netfd);
14

15
   // 关闭和客户端连接
16
   close(netfd);
17

18
   // 进程完成任务，通过本地socket,发信息送给主进程（随便发点什么），
19
   // 表示连接通信完成，促使主进程修改忙状态位闲状态
20
   pid_t pid = getpid();
21
   printf("server pid : %d \n", pid);
22
   send(socket_local_fd, &pid, sizeof(pid), 0);
23
}
24
return 0;
25
}
26

27
// 通过客户端的连接的文件描述符和客户端通信
28
// 参数一: 和客户端连接的文件描述符
29
int netToClient(int netfd){
30

31
char *str = "hello";
32
send(netfd, str, strlen(str), 0);
33
sleep(20);
34

35
return 0;
36
}

localSocket

xxxxxxxxxx
73
1
#include "head.h"
2

3
// 工作线程用来读取main线程accept的客户端连接对象
4
// 参数一: 工进程程用来和main进程通信的特制本地socket
5
// 参数二: 用来存储从本地socket中读到的客户端连接对象的文件描述符
6
int read_net_fd(int socket_local_fd, int *netfd){
7

8
struct msghdr hdr;
9
bzero(&hdr, sizeof(hdr));
10

11
struct iovec vec[1];
12
int flag = 1;
13
vec[0].iov_base = &flag;
14
vec[0].iov_len = sizeof(int);
15

16
hdr.msg_iov = vec;
17
hdr.msg_iovlen = 1;
18

19
struct cmsghdr *pcmsg = 
20
   (struct cmsghdr *)malloc(
21
                   CMSG_LEN(sizeof(int)));
22
pcmsg->cmsg_len = CMSG_LEN(sizeof(int));
23
pcmsg->cmsg_level = SOL_SOCKET;
24
pcmsg->cmsg_type = SCM_RIGHTS;
25

26
hdr.msg_control = pcmsg;
27
hdr.msg_controllen = CMSG_LEN(sizeof(int));
28

29
int ret = recvmsg(socket_local_fd, &hdr, 0);
30
ERROR_CHECK(ret, -1, "recvmsg");
31

32
void *addr = CMSG_DATA(pcmsg);
33
int *p_fd = (int *)addr;
34
*netfd = *p_fd;
35

36
return 0;
37
}
38

39
// main进程accept获得的客户端连接对象发送给工作进程
40
// 参数一: main进程用来和工作进程通信的本地socket
41
// 参数二: main进程accept获得到的客户端连接文件对象的文件描述符
42
int write_net_fd(int socket_local_fd, int* netfd){
43

44
struct msghdr hdr;
45
bzero(&hdr, sizeof(hdr));
46

47
struct iovec vec[1];
48
int flag = 1;
49
vec[0].iov_base = &flag;
50
vec[0].iov_len = sizeof(int);
51

52
hdr.msg_iov = vec;
53
hdr.msg_iovlen = 1;
54

55
struct cmsghdr *pcmsg = 
56
   (struct cmsghdr *)malloc(
57
                   CMSG_LEN(sizeof(int)));
58
pcmsg->cmsg_len = CMSG_LEN(sizeof(int));
59
pcmsg->cmsg_level = SOL_SOCKET;
60
pcmsg->cmsg_type = SCM_RIGHTS;
61

62
void *addr = CMSG_DATA(pcmsg);
63
int *p_int = (int *)addr;
64
*p_int = *netfd;
65

66
hdr.msg_control = pcmsg;
67
hdr.msg_controllen = CMSG_LEN(sizeof(int));
68

69
int ret = sendmsg(socket_local_fd, &hdr, 0);
70
ERROR_CHECK(ret, -1, "sendmsg");
71

72
return 0;
73
}

tcpInit

xxxxxxxxxx
26
1
#include "head.h"
2
// 根据端口和IP构建服务端的网络监听
3
// 参数1: socket的文件描述符(传入传出)
4
// 参数2: ip地址
5
// 参数3: 端口
6
int initTcpSocket(int * socketfd, char *ip, char *port){
7

8
// 创建socket文件对象
9
*socketfd = socket(AF_INET, SOCK_STREAM, 0);
10

11
// 解除TIME_WAIT等待时:导致端口占用问题
12
int reuse = 1;
13
setsockopt(*socketfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
14

15
// 构建sockadd_in
16
struct sockaddr_in addr;
17
addr.sin_family = AF_INET;
18
addr.sin_addr.s_addr = inet_addr(ip);
19
addr.sin_port = htons(atoi(port));
20

21
// 绑定端口
22
bind(*socketfd, (struct sockaddr *) &addr, sizeof(addr));
23
// 开始监听
24
listen(*socketfd, 10);
25
return 0;
26
}

epoll

xxxxxxxxxx
13
1
#include "head.h"
2

3
// 添加epoll监听
4
// 参数1: epoll的文件描述符id
5
// 参数2: 要添加到epoll监控的文件的文件描述符
6
int epoll_addfd(int epollfd,  int filefd){
7
struct epoll_event event;
8
event.data.fd = filefd;
9
event.events = EPOLLIN;
10
epoll_ctl(epollfd,EPOLL_CTL_ADD, filefd, &event);
11

12
return 0;
13
}

client

xxxxxxxxxx
24
1
#include <55header.h>
2

3
int main(int argc,char*argv[])
4
{
5
   char *ip = "192.168.106.129";
6
   char *port = "8080";
7

8
   int socketfd = socket(AF_INET, SOCK_STREAM, 0);
9

10
   struct sockaddr_in addr;
11
   addr.sin_family = AF_INET;
12
   addr.sin_addr.s_addr = inet_addr(ip);
13
   addr.sin_port = htons(atoi(port));
14

15
   connect(socketfd, (struct sockaddr *)&addr, sizeof(addr));
16

17
   char buf[60] = {0};
18
   int res = recv(socketfd, buf, sizeof(buf), 0);
19
   ERROR_CHECK(res, -1, "recv");
20
   printf("buf: %s \n", buf);
21

22
   close(socketfd);
23
   return 0;
24
}

1.2 第二版

在上面过程中, 我们解决了进程间共享文件对象的问题, 并且实现了服务器的主进程接收客户端连接请求, 并把客户端连接交给进程池中进程具体和客户端进行交互的功能。

假设在上一版本的基础上, 如果客户端的请求是想获得一份在服务端的文件, 我们该把前面向客户端发送简单的字符信息, 变为向客户端传输文件那? 客户端又怎么解决接收文件的问题?

1.2.1 文件的传输

客户端下载文件/服务器向客户端传输文件: 以一个小文件(eg:1000字节)为例

EgCode: 服务器 (修改worker: netToClient -> sendFile)

xxxxxxxxxx
19
1
// 向客户端发送文件
2
// 参数一: 客户端连接的文件对象
3
int sendFile(int netFd){
4
// 发送文件名
5
char *file_name = "file.txt";
6
send(netFd, file_name, strlen(file_name), 0);
7

8
// 打开文件
9
int file_fd = open(file_name, O_RDONLY);
10

11
// 读取文件数据
12
char buf[1000] = {0};
13
ssize_t sret = read(file_fd, buf, sizeof(buf));
14

15
// 发送给客户端
16
send(netFd, buf, sret, 0);
17

18
return 0;
19
}

EgCode: 客户端( client调用downloadFile )

xxxxxxxxxx
18
1
// client: 接收文件
2
//下载文件
3
int downloadFile(int netFd){
4
// 读取文件的名字
5
char buf_name[60] = {0};
6
recv(netFd, buf_name, sizeof(buf_name), 0);
7

8
// 创建文件
9
int file_fd = open(buf_name, O_RDWR|O_CREAT, 0666);
10

11
// 读取文件内容
12
char buf[1000] = {0};
13
ssize_t res = recv(netFd,buf, sizeof(buf),0);
14
// 写入文件
15
write(file_fd, buf, res);
16

17
return 0;
18
}

1.2.2 粘包问题

在上面示例中, 当我们先调用send向客户端发送文件名, 之后又调用send向客户端发送从文件读取的数据, 连续两次发送, 而我们发送的数据是没有设定的数据边界的(因为TCP的接收缓冲区将多个发送的数据序列视为连续的字节流), 所以当客户端读取的时候, 有可能一次recv读取了两次发送的内容. 并且把文件名+文件内容作为文件名创建一个文件. 这就是说为的粘包问题. (UDP没有粘包问题, 因为UDP发送数据, 并不会像TCP协议那样在TCP层/传输层对数据进行拆分和重组, UDP的拆分重组行为是IP层/网络层进行的, UDP层/传输层无感知, UDP层/传输层只会觉得每一个UDP报文段都是完整的.)

所以我们需要做的事情就是厘定数据传输的边界: 参考我们前面的管道传输文件的示例, 我们可以进行如下改造.

EgCode : worker

xxxxxxxxxx
35
1
// 发送信息, 指明长度
2
typedef struct train_s {
3
int len;
4
char buf[1000];
5
}train_t;
6

7
// 向客户端发送文件
8
// 参数一: 客户端连接的文件对象
9
int sendFile(int netFd){
10
// 文件名
11
char *file_name = "file.txt";
12
// 初始化文件名信息
13
train_t train;
14
bzero(&train, 0);
15
train.len = strlen(file_name);
16
memcpy(train.buf, file_name, train.len);
17

18
// 发送文件名长度
19
send(netFd, &train.len, sizeof(int), 0);
20
// 根据指定长度发送文件名
21
send(netFd, train.buf, train.len, 0);
22

23
// 打开文件
24
int file_fd = open(file_name, O_RDONLY);
25

26
bzero(&train, 0);
27
// 读取文件数据
28
ssize_t sret = read(file_fd, train.buf, sizeof(train.buf));
29
train.len = sret;
30

31
// 发送给客户端
32
send(netFd, train.buf, train.len, 0);
33

34
return 0;
35
}

EgCode: client

xxxxxxxxxx
22
1
//下载文件
2
int downloadFile(int netFd){
3
// 读取文件名字的长度
4
int file_name_len;
5
recv(netFd, &file_name_len, sizeof(int), 0);
6
printf("name_len: %d \n", file_name_len);
7

8
// 根据文件名字的长度:读取文件的名字
9
char buf_name[60] = {0};
10
recv(netFd, buf_name, file_name_len, 0);
11

12
// 创建文件
13
int file_fd = open(buf_name, O_RDWR|O_CREAT, 0666);
14

15
// 读取文件内容
16
char buf[1000] = {0};
17
ssize_t res = recv(netFd,buf, sizeof(buf),0);
18
// 写入文件
19
write(file_fd, buf, res);
20

21
return 0;
22
}

1.2.3 发送大文件

假设我们发送一个大文件: 代码改造如下

EgCode: worker

xxxxxxxxxx
42
1
typedef struct train_s {
2
int len;
3
char buf[1000];
4
}train_t;
5

6
// 向客户端发送文件
7
// 参数一: 客户端连接的文件对象
8
int sendFile(int netFd){
9
// 文件名
10
char *file_name = "file.txt";
11
// 初始化文件名信息
12
train_t train;
13
bzero(&train, 0);
14
train.len = strlen(file_name);
15
memcpy(train.buf, file_name, train.len);
16

17
// 发送文件名长度
18
send(netFd, &train.len, sizeof(int), 0);
19
// 根据指定长度发送文件名
20
send(netFd, train.buf, train.len, 0);
21

22
// 打开文件
23
int file_fd = open(file_name, O_RDONLY);
24

25
while(1){
26
  bzero(&train, sizeof(train));
27
  // 读取文件数据
28
  ssize_t sret = read(file_fd, train.buf, sizeof(train.buf));
29
  train.len = sret;
30
  if(sret == 0){
31
      // 文件读完
32
      break;
33
  }
34
  // 发送给客户端   -----------------------> 出问题, 半包问题 (会发送半包)
35
  int net_res = send(netFd, &train, train.len+sizeof(train.len), 0);
36
  if(net_res == -1){
37
      // 发送失败, 对端断开
38
      break;
39
  }
40
}
41
return 0;
42
}

EgCode: client

xxxxxxxxxx
33
1
//下载文件
2
int downloadFile(int netFd){
3
// 读取文件名字的长度
4
int file_name_len;
5
recv(netFd, &file_name_len, sizeof(int), 0);
6
printf("name_len: %d \n", file_name_len);
7

8
// 根据文件名字的长度:读取文件的名字
9
char buf_name[60] = {0};
10
recv(netFd, buf_name, file_name_len, 0);
11

12
// 创建文件
13
int file_fd = open(buf_name, O_RDWR|O_CREAT, 0666);
14

15
// 循环读取
16
while(1){
17
  // 读取长度
18
  int len = 0;
19
  recv(netFd, &len, sizeof(int), 0);
20
  // len值出错会:    -----------------------> 出问题, 半包问题
21
      if(len == 0){
22
          // 对方发送完毕
23
      break;
24
  }
25
  // 读取文件内容
26
  char buf[1000] = {0};
27
  ssize_t res = recv(netFd,buf, len,0);
28
  // 写入文件
29
      write(file_fd, buf, res);
30
   }
31
   return 0;
32
}
33

问题:

在上面发送的过程中, 我们会发现, 如果一个文件比较大, 偶尔会在客户端接收文件的时候, 产生接收错误. 这是因为, 我们的信息发送行为并不是由send函数控制(send函数本身只是把要发送的数据, 交给操作系统), 具体什么时候真正发送数据, 是由操作系统决定的. 操作系统, 有可能在发送数据的时候, 某个train只发送了一半, 然后被客户端读取, 在一段时之后, 操作系统发送了另一半train给客户端, 客户端先读取数据长度的时候, 出现错误, 导致最终数据读取错误. 这种不可控的行为我们称之为半包问题.

所以在recv函数中, 提供了关于接收行为的标志位中提供了MSG_WAITALL字段, 用于控制recv读取指定的len长度的数据才返回. 进而解决半包问题.

xxxxxxxxxx
9
1
#include <sys/types.h>
2
#include <sys/socket.h>
3
ssize_t recv(
4
  int sockfd,
5
  void *buf,
6
  size_t len,
7
  int flags     // 定接收行为的标志位:MSG_WAITALL(等待所有请求的数据才返回)...大多数情况下，flags设置为0。
8
);
9
// 返回值: 成功时返回实际读取的字节数。如果连接已经关闭返回0(对方close: 四次挥手)。读取失败返回-1

EgCode: client

xxxxxxxxxx
25
1
//下载文件
2
int downloadFile(int netFd){
3
int file_name_len;
4
recv(netFd, &file_name_len, sizeof(int), MSG_WAITALL);
5

6
char buf_name[60] = {0};
7
recv(netFd, buf_name, file_name_len, MSG_WAITALL);
8

9
int file_fd = open(buf_name, O_RDWR|O_CREAT, 0666);
10

11
while(1){
12
int len = 0;
13
 recv(netFd, &len, sizeof(int), MSG_WAITALL);
14
 if(len == 0){
15
 // 对方发送完毕
16
     break;
17
 }
18
  // 读取文件内容
19
  char buf[1000] = {0};
20
  ssize_t res = recv(netFd,buf, len,MSG_WAITALL);
21
  // 写入文件
22
  write(file_fd, buf, res);
23
}
24
return 0;
25
}

问题:

在发送大文件的时候, 客户端有可能在发送的时候提前终止, 这会导致发送端/写端(send)因为抛出SIGPIPE导致进程终止.

EgCode: worker

xxxxxxxxxx
39
1
typedef struct train_s {
2
int len;
3
char buf[1000];
4
}train_t;
5

6
void fun(int num){
7
printf("sigpeipe \n");
8
}
9
int sendFile(int netFd){
10
// TODO: 注册信号
11
signal(SIGPIPE, fun);
12
char *file_name = "file.txt";
13
int file_fd = open(file_name, O_RDONLY);
14
train_t train;
15
bzero(&train, 0);
16

17
// 发送文件长度和名字
18
train.len = strlen(file_name);
19
memcpy(train.buf, file_name, train.len);
20
send(netFd, &train,sizeof(int)+train.len, 0);
21

22
while(1){
23
   bzero(&train, 0);
24
   ssize_t sret = read(file_fd, train.buf, sizeof(train.buf));
25
   train.len = sret;
26
   if(sret == 0){
27
       // 文件读完
28
       break;
29
   }
30
   int net_res = send(netFd, &train, train.len+sizeof(train.len), 0);
31
   sleep(1); // TODO:睡1秒发一次
32
   printf("net_res :%d  \n", net_res);
33
   if(net_res == -1){
34
       // 发送失败, 对端断开
35
       break;
36
   }
37
}
38
return 0;
39
}

改进

xxxxxxxxxx
9
1
#include <sys/types.h>
2
#include <sys/socket.h>
3
ssize_t send(
4
int sockfd, 
5
const void *buf, 
6
size_t len, 
7
int flags// 额外选项:MSG_NOSIGNAL防止发送时由于连接断开而引发的SIGPIPE信号 ...大多数情况下，flags参数设置为0。
8
);
9
// 返回值: 成功返回实际发送的字节数。失败返回-1

xxxxxxxxxx
32
1
typedef struct train_s {
2
int len;
3
char buf[1000];
4
}train_t;
5

6
int sendFile(int netFd){
7
char *file_name = "file.txt";
8
int file_fd = open(file_name, O_RDONLY);
9
train_t train;
10
bzero(&train, 0);
11

12
// 发送文件长度和名字
13
train.len = strlen(file_name);
14
memcpy(train.buf, file_name, train.len);
15
send(netFd, &train,sizeof(int)+train.len, MSG_NOSIGNAL); // 设置:MSG_NOSIGNAL
16

17
while(1){
18
   bzero(&train, 0);
19
   ssize_t sret = read(file_fd, train.buf, sizeof(train.buf));
20
   train.len = sret;
21
   if(sret == 0){
22
       // 文件读完
23
       break;
24
   }
25
   int net_res = send(netFd, &train, train.len+sizeof(train.len), MSG_NOSIGNAL);// 设置:MSG_NOSIGNAL
26
   if(net_res == -1){
27
       // 发送失败, 对端断开
28
       break;
29
   }
30
}
31
return 0;
32
}

1.2.4 进度条

如果我们想模仿日常下载文件的时候, 进度条显示的效果. 我们可以在文件传输之前, 先传输文件大小给客户端, 在客户端不断接收文件的时候, 根据已经接收的文件的大小/总文件的大小, 显示进度条.

我们需要用到fstat函数获得一个文件的状态信息 (man fstat)

xxxxxxxxxx
9
1
#include <sys/types.h>
2
#include <sys/stat.h>
3
#include <unistd.h>
4
//get file status
5
int fstat(
6
   int fd, // 文件描述符
7
   struct stat *statbuf // 存储文件状态的指针
8
);
9
// 返回值:成功返回0, 失败-1

xxxxxxxxxx
4
1
struct stat{
2
   off_t     st_size;  //文件的大小，以字节为单位
3
   // .....很多参数(文件所属用户/组,文件块数, 文件修改时间.....)
4
}

EgCode: worker

xxxxxxxxxx
39
1
typedef struct train_s {
2
int len;
3
char buf[1000];
4
}train_t;
5

6
int sendFile(int netFd){
7
char *file_name = "file.txt";
8
int file_fd = open(file_name, O_RDONLY);
9
train_t train;
10
bzero(&train, 0);
11

12
// 获得文件信息
13
struct stat stat_file;
14
fstat(file_fd, &stat_file);
15
// 发送文件长度
16
send(netFd, &stat_file.st_size, sizeof(off_t), MSG_NOSIGNAL);
17

18
// 发送文件长度和名字
19
bzero(&train, 0);
20
train.len = strlen(file_name);
21
memcpy(train.buf, file_name, train.len);
22
send(netFd, &train,sizeof(int)+train.len, MSG_NOSIGNAL);
23

24
while(1){
25
   bzero(&train, 0);
26
   ssize_t sret = read(file_fd, train.buf, sizeof(train.buf));
27
   train.len = sret;
28
   if(sret == 0){
29
       // 文件读完
30
       break;
31
   }
32
   int net_res = send(netFd, &train, train.len+sizeof(train.len), MSG_NOSIGNAL);
33
   if(net_res == -1){
34
       // 发送失败, 对端断开
35
       break;
36
   }
37
}
38
return 0;
39
}

EgCode: client

xxxxxxxxxx
42
1
int downloadFile(int netFd){
2

3
// 读取文件长度
4
off_t file_size = 0;
5
recv(netFd, &file_size, sizeof(off_t), MSG_WAITALL);
6
printf("file_size : %ld \n", file_size);
7

8
int file_name_len;
9
recv(netFd, &file_name_len, sizeof(int), MSG_WAITALL);
10
char buf_name[60] = {0};
11
recv(netFd, buf_name, file_name_len, MSG_WAITALL);
12
    // 创建文件
13
int file_fd = open(buf_name, O_RDWR|O_CREAT, 0666);
14

15
// 数据传输标记:记录接收数据量
16
off_t cursize = 0; // 当前接收
17
off_t last_update_size = 0; // 每更新一次百分比增加
18

19
while(1){
20
   int len = 0;
21
   recv(netFd, &len, sizeof(int), MSG_WAITALL);
22
   if(len == 0){
23
       // 对方发送完毕
24
       break;
25
   }
26
   char buf[1000] = {0};
27
   ssize_t res = recv(netFd,buf, len,MSG_WAITALL);
28
   write(file_fd, buf, res);
29

30
   // 记录当前接收的文件大小
31
   cursize += len;
32
   // 计算相比上一次打印, 增加的百分比
33
   double num = (double)cursize*100/file_size - (double)last_update_size*100/file_size;
34
   if(num > 1){
35
       // 进度增加了百分之一
36
       last_update_size = cursize;
37
       // 打印进度条
38
       printf("now: %.2f \n", (double)last_update_size*100/file_size);
39
   }
40
}
41
return 0;
42
}