UNIX环境编程-高级IO（12）

视频教程

代码链接

一、有限状态机

1.概念

有限状态机（Finite State Machine, FSM）是用于表示和控制执行序列的数学模型。它可以处于有限数量的状态之一，而且在任意时刻只能处于其中的一个状态。FSM的状态、状态转移和可能的输入动作都是预定义的。

FSM通常分为两类：

  • 确定性有限自动机（DFA）：对于任何给定的输入和当前状态，其转移是唯一确定的。
  • 非确定性有限自动机（NFA）：对于给定的输入和当前状态，可能有多个可能的下一个状态

有限状态机可以使用以下几个元素来描述：

  • 状态 (States)：FSM可以处于的所有可能情况。
  • 初始状态 (Start State)：FSM开始操作时的状态。
  • 输入 (Inputs)：可以影响状态转移的外部决策。
  • 转移函数 (Transition Function)：定义了基于当前状态和输入如何从一个状态转移到另一个状态。
  • 结束状态 (End States or Accept States)：在某些类型的FSM中定义了特定的结束或接受状态

2.实例（数据中继）

（1）任务要求

image-20230821204332514

（2）补充知识
fcntl()

通过这个函数，你可以更改已经打开的文件的属性

1
2
3
#include <fcntl.h>

int fcntl(int fd, int cmd, ... /* arg */ );

其中：

  • fd 是要操作的文件描述符。
  • cmd 是一个指定操作的命令。
  • arg 是可选参数，其需求取决于 cmd。

常用的 cmd 命令和对应的操作包括：

  • F_DUPFD: 复制文件描述符。
  • F_GETFD: 获取文件描述符标志。
  • F_SETFD: 设置文件描述符标志。
  • F_GETFL: 获取文件状态标志。
  • F_SETFL: 设置文件状态标志。
  • F_GETLK, F_SETLK, F_SETLKW: 用于文件锁定
EAGAIN

EAGAIN (或在某些系统上是 EWOULDBLOCK) 是一个在POSIX兼容的操作系统中定义的错误码，通常与非阻塞I/O操作相关

当你对一个非阻塞的文件描述符执行I/O操作（如read()或write()），而该操作不能立即完成时，系统通常不会将调用线程阻塞，而是立即返回这个错误码(假错误)

这里有一些常见的例子，说明什么时候可能会遇到 EAGAIN：

  • 非阻塞读操作：如果你尝试从一个空的非阻塞套接字或其他设备读取数据，而没有数据可用，read()将返回-1，并设置errno为EAGAIN。
  • 非阻塞写操作：如果你尝试将数据写入一个非阻塞的套接字，但该套接字的发送缓冲区已满，那么write()会返回-1，并设置errno为EAGAIN。

当你收到 EAGAIN 错误时，通常的做法是稍后重试I/O操作，或使用像select()、poll()或epoll()这样的机制来等待文件描述符变得可用

（3）实现

在目录下创建1.txt与2.txt,其内容分别如下：

image-20230821203554949

relay.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>

// #define TTY1 "/dev/tty11"
// #define TTY2 "/dev/tty12"

#define TTY1 "./1.txt"
#define TTY2 "./2.txt"
#define BUFSIZE     1024

// 状态机的状态
enum
{
	STATE_R = 1, // read
    	STATE_W,     // write
    	STATE_Ex,    // error
    	STATE_T      // over
};


// 状态机结构体
struct fsm_st
{
	int state; // 状态
	int sfd;   // 源文件
	int dfd;   // 目标文件
	char buf[BUFSIZE];
	int len;
	int pos;
	char *errstr;
};

// 状态转移函数
static void fsm_driver(struct fsm_st *fsm)
{
	int ret;
	switch(fsm->state)
	{
		// 读
		case STATE_R:
			fsm->len = read(fsm->sfd, fsm->buf, BUFSIZE);
			// 读完
			if(fsm->len == 0)
				// 终止结束
				fsm->state = STATE_T;
			else if(fsm->len < 0)
			{
				// 假错
				if(errno == EAGAIN)
                    			fsm->state = STATE_R;
                    		else
                    			// 真错，报错退出
                    			// 读出错
                    			fsm->errstr = "read()";
                    			fsm->state = STATE_Ex;
			}
			else
			{
				// 状态转移到写态，pos置0
				fsm->pos = 0;
				fsm->state = STATE_W;
			}
			break;
		
		
		// 写
		case STATE_W:
			// 从buf 中后移动pos的位置取得内容写入dfd中
			ret = write(fsm->dfd, fsm->buf + fsm->pos, fsm->len);
			if(ret < 0)
			{
				// 假 错
				if(errno == EAGAIN)
					fsm->state = STATE_W;
				else
					// 真错，报错退出
					// 写出错
					fsm->errstr = "write()";
                    			fsm->state = STATE_Ex;
			}
			else
			{
				// 既然buf中ret 的字节已经写入dfd中，那么buf中位置指针向后移动
				fsm->pos += ret;
				// 坚持写够len个字节，将buf中读取的字节数量写完
				fsm->len -= ret;
				if(fsm->len == 0)
				{
					// 写完-->继续去读
					fsm->state = STATE_R;
				}
				else
					// 继续写
					fsm->state = STATE_W;
			}
			break;
		
		// 异常处理态
		case STATE_Ex:
			perror(fsm->errstr);
			break;
			
		case STATE_T:
			break;
			
		default:
			abort();
			break;
	}
}

// 数据中继：需要保证文件均为非阻塞的状态
static void relay(int fd1,int fd2)
{
	int fd1_save,fd2_save;
	// fsm12 读左写右  fsm21 读右写左
	struct fsm_st fsm12,fsm21;
	
	// F_GETFL: 获取文件状态标志
	fd1_save = fcntl(fd1,F_GETFL);
	// F_SETFL: 设置文件状态标志
	fcntl(fd1,F_SETFL,fd1_save|O_NONBLOCK);
	
	fd2_save = fcntl(fd2,F_GETFL);
	fcntl(fd2,F_SETFL,fd2_save|O_NONBLOCK);
	
	
	// fsm12的初始状态
	fsm12.state = STATE_R;
	fsm12.sfd = fd1;
	fsm12.dfd = fd2;
	
	// fsm21的初始状态
	fsm21.state = STATE_R;
	fsm21.sfd = fd2;
	fsm21.dfd = fd1;
	
	// 当两个状态机均非终止态
	while(fsm12.state != STATE_T || fsm21.state != STATE_T)
    	{
    		// 驱动状态机(调用状态转移函数)
        	fsm_driver(&fsm12);
        	fsm_driver(&fsm21);
    	}
	
	// 恢复文件描述符的初始状态
	fcntl(fd1,F_SETFL,fd1_save);
	fcntl(fd2,F_SETFL,fd2_save);
}

int main()
{
	int fd1,fd2;
	
	fd1 = open(TTY1,O_RDWR);
	if(fd1 < 0)
	{
		perror("open()");
		exit(1);
	}
	
	// O_NONBLOCK 非阻塞I/O操作
	fd2 = open(TTY2,O_RDWR|O_NONBLOCK);
	if(fd2 < 0)
	{
		perror("open()");
		exit(1);
	}
	
	relay(fd1, fd2);
	
	close(fd1);
	close(fd2);
	
	
	exit(0);
}

编译运行之后,查看两个文件

image-20230821203720692

二、中继引擎实例

视频教程

main.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include<errno.h>
#include<string.h>
#include"relayer.h"

// 创建两个设备互相传递信息
#define TTY1 "../12.txt"
#define TTY2 "../21.txt"

#define TTY3 "../34.txt"
#define TTY4 "../43.txt"


int main()
{
	int fd1,fd2,fd3,fd4;
	int job1,job2;

	fd1 = open(TTY1,O_RDWR); // 先以阻塞形式打开
	if(fd1 < 0)
	{
		perror("open()");
		exit(1);
	}
	write(fd1, "TTY1\n", 5);

	fd2 = open(TTY2,O_RDWR|O_NONBLOCK); // 非阻塞
	if(fd2 < 0)
	{
		perror("open()");
              	exit(1);
	}
	write(fd2, "TTY2\n", 5);

	// 添加任务
	job1 = rel_addjob(fd1,fd2);
	if(job1 < 0)
	{
		fprintf(stderr,"rel_addjob():%s\n",strerror(-job1));
		exit(1);
	}

		
	fd3 = open(TTY3,O_RDWR); // 先以阻塞形式打开
	if(fd3 < 0)
	{
		perror("open()");
		exit(1);
	}
	write(fd3, "TTY3\n", 5);

	fd4 = open(TTY4,O_RDWR|O_NONBLOCK); // 非阻塞
	if(fd4 < 0)
	{
		perror("open()");
              	exit(1);
	}
	write(fd4, "TTY4\n", 5);

	// 添加任务
	job2 = rel_addjob(fd3,fd4);
	if(job2 < 0)
	{
		fprintf(stderr,"rel_addjob():%s\n",strerror(-job2));
		exit(1);
	}
	
	// 使进程休眠，让任务持续执行 
	while(1)
		pause();
	

	close(fd1);
	close(fd2);
	close(fd3);	
	close(fd4);

	exit(0);
}

relayer.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
#include<stdio.h>
#include<stdlib.h>
#include<errno.h>
#include<pthread.h>
#include <unistd.h>
#include<string.h>
#include<fcntl.h>
#include"relayer.h"

#define BUFSIZE 1024

// 任务数组
static struct rel_job_st *rel_job[REL_JOBMAX];
// 互斥量(静态初始化)
static pthread_mutex_t mut_rel_job = PTHREAD_MUTEX_INITIALIZER;
static pthread_once_t init_once = PTHREAD_ONCE_INIT;


// 状态机状态
enum
{
	STATE_R = 1, // read
    	STATE_W,     // write
    	STATE_Ex,    // error
    	STATE_T      // over
};



// 任务状态
enum{
	STATE_RUNNING = 1,  // 运行
	STATE_CANCELED,     // 取消
	STATE_OVER          // 结束
};


// 状态机结构体
struct rel_fsm_st{
	int state;  // 状态机状态
	int sfd;    // 源文件
	int dfd;    // 目的文件
	char buf[BUFSIZE]; // 中间缓冲区
	int len; // 读到的长度
	int pos; // 写的过程如果一次没有写完，记录上次写的位置
	char *err; // 错误信息
	int64_t count; // 输出的字符数量
};


// 任务结构体
struct rel_job_st{
	// 两个终端设备
	int fd1;
	int fd2;
	// 任务的状态
	int job_state;
	// 两个终端设备(文件)状态机结构体
	struct rel_fsm_st fsm12,fsm21;
	// 用来退出复原设备（文件）原有的状态
	int fd1_save,fd2_save;
};


// 状态转移函数
static void fsm_driver(struct rel_fsm_st *fsm)
{
	int ret;
	switch(fsm->state)
	{	
		// 读态
		case STATE_R:
			fsm->len = read(fsm->sfd,fsm->buf,BUFSIZE);
			if(fsm->len == 0)
			{
				// 为0 表示文件读到末尾处，没有内容了，退出到终止态
				fsm->state = STATE_T;
			}
			// 读取失败，这是系统调用需要判断是真错还是假错
			else if(fsm->len < 0)
			{	
				// 假错
				if(errno == EAGAIN) 
					// 继续读
					fsm->state = STATE_R;
				// 真错
				else
					// 存储报错信息
					fsm->err = "read()";
					// 错误态
					fsm->state = STATE_Ex;
			}
			// 读取到内容
			else
			{	
				// 写入文件的位置置0
				fsm->pos = 0;
				// 切换到写态
				fsm->state = STATE_W;
			}

			break;

		
		// 写态
		case STATE_W:
			ret = write(fsm->dfd,fsm->buf +fsm->pos,fsm->len);
			// 判断真假错
			if(ret < 0)
			{
				if(errno == EAGAIN)
					// 继续写
					fsm->state = STATE_W;
				else
					// 真错,报错退出
					fsm->err = "wirte()";
					// 错误态
					fsm->state = STATE_Ex;
			}
			// 成功写入内容
			else
			{
				// 防止读取的内容没有一次性全部写入
				fsm->pos += ret;
				fsm->len -= ret;
				// 内容全部写入，继续读
				if(fsm->len == 0)
					// 切换到读态
					fsm->state = STATE_R;
				else
					// 没有写完继续写
					fsm->state = STATE_W;
			}

			break;

		// 错误态
		case STATE_Ex:
			// 将错误信息打印
			perror(fsm->err);
			// 切换到终止态
			fsm->state = STATE_T;
		
		// 终止态
		case STATE_T:
			break;

		default:
			// 异常终止
			abort();
			break;
	}
}


static void *thr_relayer(void *p)
{
	int i;
	while(1)
	{
		// 加锁
		pthread_mutex_lock(&mut_rel_job);
		for(i = 0;i<REL_JOBMAX;i++)
		{
			if(rel_job[i] != NULL)
			{
				if (rel_job[i]->job_state == STATE_RUNNING)
				{
					// 当前任务为运行态 驱动状态转移函数
					fsm_driver(&rel_job[i]->fsm12);
					fsm_driver(&rel_job[i]->fsm21);
					// 若状态机均为终止态，任务终止
					if(rel_job[i]->fsm12.state == STATE_T && rel_job[i]->fsm21.state == STATE_T)
						// 任务终止
						rel_job[i]->job_state = STATE_OVER;
				}
			}
		}
		// 解锁
		pthread_mutex_unlock(&mut_rel_job);
	}
}


static void module_load(void)
{
	int err;
	pthread_t tid_relayer;

	err = pthread_create(&tid_relayer,NULL,thr_relayer,NULL);
	if(err)
	{
		fprintf(stderr,"pthread_create():%s\n",strerror(err));
		exit(1);
	}
}



// 找到任务数组为NULL的数组下标
static int get_free_pos_unlocked()
{
	int i;
	for(i = 0;i<REL_JOBMAX;i++)
	{
		if(rel_job[i] == NULL)
			return i;
	}
	return -1;
}


// 添加任务
int rel_addjob(int fd1,int fd2)
{
	struct rel_job_st *me;
	int pos;
	

	// 确保module_load在多线程环境中只执行一次的机制
	pthread_once(&init_once,module_load);
	me = malloc(sizeof(*me));
	if (me == NULL)
	{
		return -ENOMEM;
	}
	me->fd1 = fd1;
	me->fd2 = fd2;
	me->job_state = STATE_RUNNING; // 对设备设置正在运行
	

	// 读取设备1的状态
	me->fd1_save = fcntl(me->fd1, F_GETFL); 
	// 设置设备1的状态为非阻塞
	fcntl(me->fd1, F_SETFL, me->fd1_save | O_NONBLOCK);  //非阻塞	打开

	// 读取设备2的状态
        me->fd2_save = fcntl(me->fd2, F_GETFL);
   	// 设置设备2的状态为非阻塞
   	fcntl(me->fd2, F_SETFL, me->fd2_save | O_NONBLOCK);  //非阻塞   打开

	me->fsm12.sfd = me->fd1;
	me->fsm12.dfd = me->fd2;
	me->fsm12.state = STATE_R;

	me->fsm21.sfd = me->fd2;
	me->fsm21.dfd = me->fd1;
	me->fsm21.state = STATE_R;
	
	pthread_mutex_lock(&mut_rel_job);
	pos = get_free_pos_unlocked();
	if (pos < 0) {
		pthread_mutex_unlock(&mut_rel_job);
		fcntl(me->fd1, F_SETFL, me->fd1_save);
		fcntl(me->fd2, F_SETFL, me->fd2_save);
		free(me);
		return -ENOSPC;
	}
	rel_job[pos] = me;
	pthread_mutex_unlock(&mut_rel_job);
	return pos;

}



int rel_canceljob(int id)
{

}
/*
	return == 0			成功，指定任务成功取消
		   == -EINVAL	失败，参数非法
		   == -EBUSY    失败，任务早已被取消
*/



int rel_waitjob(int id, struct rel_stat_st *rel)
{}




int rel_statjob(int id, struct rel_stat_st *rel)
{}

relayer.h

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
#ifndef RELAYER_H_
#define RELAYER_H_

#include <stdio.h>

#define REL_JOBMAX 1000

// 任务状态结构体
struct rel_stat_st{
	int state;
	int fd1;
	int fd2;
	int64_t count12,count21; // 1向2发送了多少字节  2向1发送了多少字节
};


int rel_addjob(int fd1,int fd2);
/*	func:添加任务
 *	return >= 0			成功，返回当前任务ID
 *	       == -EINVAL	失败，参数非法
 *	       == -ENOSPC   失败，任务数组满
 *	       == -ENOMEM   失败，内存分配有误
 * */

int rel_canceljob(int id);
/*	func: 取消任务
 *	return == 0			成功，指定任务成功取消
 *		   == -EINVAL	失败，参数非法
 *		   == -EBUSY    失败，任务早已被取消
* */


int rel_waitjob(int id,struct rel_stat_st *);
/*	func: 等待任务
 *	return == 0			成功，指定任务已终止并返回状态
 *             == -EINVAL	失败，参数非法
 * */


int rel_statjob(int id,struct rel_stat_st *);
/*	func: 任务状态
 *	return == 0			成功，指定任务状态返回
 *             == -EINVAL	失败，参数非法
 * */

#endif
1
2
3
4
5
6
7
8
cmake_minimum_required(VERSION 3.16)
project(CALC)    # 工程名字可以与可执行程序名不一样
add_executable(main main.c relayer.c relayer.h)

# 查找pthread库
find_package(Threads REQUIRED)
# 链接pthread库到你的目标
target_link_libraries(main Threads::Threads)

三、高级IO

1.IO多路转接（多路复用）

监视多个文件描述符

IO多路转接允许单个进程监视多个文件描述符以检查它们是否已准备好进行读取或写入操作。这是非阻塞IO操作的一种形式，常用于提高对多个连接或数据流的同时处理能力

IO多路转接的主要应用是在网络编程中，尤其是在服务器中，它可以处理大量的并发连接。例如，一个Web服务器可能需要同时处理数千个连接，每个连接都可能在不同的时间请求数据。通过使用IO多路转接，服务器可以有效地管理这些连接，而无需为每个连接启动一个单独的线程或进程

让我们考虑一个简单的场景：假设你是一个服务器，你有多个客户端连接到你。每个连接都是一个文件描述符。客户端可能会在任何时间发送数据，所以你需要知道哪个客户端已经发送了数据，以便你可以读取它。使用阻塞IO，你只能一次处理一个连接，这是低效的。但使用IO多路转接，你可以监视所有连接，并当数据可用时立即处理它。

（1）select()

是最古老的IO多路转接解决方案，它允许应用程序监视多个文件描述符的状态

该函数若不设置超时时间，这个函数就会一直死等（阻塞）

1
2
3
#include <sys/select.h>

int select(int nfds, fd_set *readfds, fd_set *writefds,fd_set *exceptfds, struct timeval *timeout);
1
2
3
4
struct timeval{
  long tv_sec;    // 秒
  long tv_usec;   // 微妙
};

参数：

  • nfds: 最大的文件描述符编号加1。比如，如果你要监视的文件描述符是3、4和5，那么nfds应该是6
  • readfds: 一个文件描述符集合，用于检查哪些描述符准备好进行读取
  • writefds: 一个文件描述符集合，用于检查哪些描述符准备好进行写入
  • exceptfds: 一个文件描述符集合，用于检查哪些描述符有异常条件发生
  • timeout: 指定select()等待的最大时间。如果为NULL，那么select()会无限期等待（不进行超时设置，函数会进行阻塞）

fd_set是一个特殊的数据类型，用于表示文件描述符集合。你可以使用以下宏来操作它：

  • FD_ZERO(fd_set *set): 清除一个文件描述符集合。
  • FD_SET(int fd, fd_set *set): 将一个文件描述符添加到集合中。
  • FD_CLR(int fd, fd_set *set): 从集合中删除一个文件描述符。
  • FD_ISSET(int fd, fd_set *set): 检查一个文件描述符是否在集合中。

返回值：

  • −1: 出错。
  • 0: 超时，指定的时间内没有任何文件描述符准备好。
  • >0: 准备好的文件描述符的数量。

（2）poll()

与 select 类似，但提供了一个更灵活的接口

poll 函数是一个用于监控一组文件描述符的状态变化的系统调用。它是网络编程和系统编程中常用的一个功能。这个函数允许程序监控多个文件描述符，以查看是否有一个或者多个文件描述符准备好进行输入输出操作

在使用 poll 函数时，检查其返回值是非常重要的，因为它告诉你函数调用的结果以及是否有文件描述符准备好进行IO操作。如果 poll 返回一个正数，那么你需要遍历文件描述符集合，检查哪些描述符的 revents 字段指示了实际发生的事件，然后据此执行相应的IO操作。

函数原型

1
2
3
#include <poll.h>

int poll(struct pollfd *fds, nfds_t nfds, int timeout);
  • fds: 指向一个 pollfd 结构体数组的指针，每个结构体用于指定一组文件描述符和需要监控的事件。
  • nfds: 指定数组 fds 中结构体的数量。
  • timeout: 等待的毫秒数，如果设置为负数，则表示无限等待，直到某个文件描述符准备好。timeout写-1表示阻塞死等，写0表示阻塞

返回值

  • >0：表示成功，这个正数描述的是有多少文件描述符实际上满足了调用中请求的条件。换句话说，它告诉你有多少个文件描述符准备好了进行IO操作
  • 0：如果 poll 函数返回0，这意味着没有文件描述符在指定的超时时间内准备好进行IO操作。这通常表示操作超时，需要根据应用程序的逻辑来处理这种情况
  • -1：当 poll 函数返回-1时，这通常表示发生了错误。在这种情况下，可以检查 errno 来获取错误的具体原因
1
2
3
4
5
struct pollfd {
    int   fd;       // 文件描述符
    short events;   // 需要监控的事件
    short revents;  // 实际发生的事件
};

其中，events 字段用于指定需要监控的事件类型，比如：

  • POLLIN: 表示数据可读。
  • POLLOUT: 表示数据可写。
  • POLLERR: 表示错误。

如何使用：

创建一个struct pollfd形式的数组，大小为n（表示监视了n个文件描述符），并且用struct pollfd *fds类型的指针指向。数组中每一个成员均是一个struct pollfd类型的结构体，里面包含文件描述符，与用户关系关心的事件（需要监控的事件），以及已发生的事件。

（3）epoll()

是 Linux 特定的（方言），提供了一个更高效的方式来处理大量的文件描述符

若编写的程序考虑移植性时，则不能使用epoll。应该使用poll或者select

在C语言中，epoll 是Linux系统提供的一种高效的多路复用IO接口，它是专门设计用来处理大量并发连接的。epoll 比传统的 select 和 poll 接口更为高效，主要是因为它在处理大量文件描述符时不会有显著的性能下降。epoll 在网络编程，尤其是在高性能服务器的设计中非常重要

epoll 的基本使用包括以下几个步骤：

创建 epoll 实例：

通过调用 epoll_create 函数来创建一个 epoll 的实例，打开一个epoll文件描述符

size 参数在较新的Linux内核中已经不再使用，但必须提供一个大于0的数值,（size并不表示，监视文件描述符的个数，只需要对函数传入一个正数即可）

1
2
3
#include <sys/epoll.h>

int epoll_create(int size);

返回值

  • >0：如果 epoll_create 函数返回一个正数，这个数值是新创建的 epoll 实例的文件描述符。这个描述符后续用于所有针对该 epoll 实例的操作，比如在调用 epoll_ctl 和 epoll_wait 时
  • <0：当 epoll_create 返回-1时，表示创建 epoll 实例失败。这种情况通常是由于某些错误，比如资源限制、无效的参数或内部错误。在这种情况下，可以检查 errno 来获取错误的具体原因
控制 epoll 监听的文件描述符：

通过调用 epoll_ctl 函数来添加、修改或删除要监控的文件描述符

1
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

相当于对poll函数封装了一层，不让用户直接操作poll函数中的struct pollfd形式的数组，而是给用户提供方法告诉系统，用户的需求，系统帮用户进行管理

  • epfd：由 epoll_create 返回的 epoll 实例的文件描述符。
  • op：要进行的操作，比如 EPOLL_CTL_ADD（添加描述符）、EPOLL_CTL_MOD（修改描述符）、EPOLL_CTL_DEL（删除描述符）。
  • fd：要操作的文件描述符。
  • event：指向 epoll_event 结构的指针，该结构指定了要监控的事件和关联的数据
等待事件的发生：

通过调用 epoll_wait 函数来等待一个或多个文件描述符上的事件

1
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);
  • epfd：epoll 实例的文件描述符。
  • events：用于从内核获取事件的 epoll_event 结构数组。
  • maxevents：告诉内核这个 events 数组可以接受多少个 epoll_event。
  • timeout：等待事件的最大毫秒数，如果设置为-1，则表示无限等待。

epoll_event 结构通常定义如下：

1
2
3
4
struct epoll_event {
    uint32_t events;  // Epoll events
    epoll_data_t data; // User data variable
};
  • events 字段用于指定感兴趣的事件和发生的事件。
    • EPOLLIN: 表示对应的文件描述符可读（包括普通文件描述符、管道和套接字）。如果有新的连接请求，也会触发这个事件
    • EPOLLOUT: 表示对应的文件描述符可写
    • EPOLLPRI: 表示对应的文件描述符有紧急的数据可读（这通常用于TCP套接字上的带外数据）
    • EPOLLERR: 表示对应的文件描述符发生错误。epoll_wait 会始终等待此事件，即使没有明确指定
    • EPOLLHUP: 表示对应的文件描述符被挂断。同样，epoll_wait 也总是等待此事件
    • EPOLLET: 设置边缘触发行为，这意味着事件只在状态发生变化时通知一次
  • data 字段通常用于存储用户定义的数据，例如文件描述符、指针等。

epoll 相比 select 和 poll 的优势在于它不需要每次调用时都重新传入整个监听列表，而且在活动连接数很多的情况下，epoll 的效率要高得多。这使得 epoll 成为处理大规模并发连接的首选方法，特别是在构建高性能网络服务器时。

四、内存映射

1.readv与writev

在C语言中，readv 和 writev 函数属于散布/聚集I/O操作，用于在单个系统调用中从文件描述符读取多个非连续缓冲区（readv）或将多个非连续缓冲区写入文件描述符（writev）。这些函数定义在 POSIX 标准中，常用于网络编程和文件操作，提高了处理多个缓冲区的效率

多个碎片地址，并非连续地址空间，可以使用readv一次将文件描述符中的内容读取至碎片地址中；或者使用writev将碎片地址的内容写入文件描述符中

（1）readv函数

readv 函数从文件描述符读取数据到多个缓冲区。其函数原型如下：

1
2
3
#include <sys/uio.h>

ssize_t readv(int fd, const struct iovec *iov, int iovcnt);
  • fd: 文件描述符，指向要读取的文件或套接字。
  • iov: 指向一个或多个 iovec 结构的指针，这些结构定义了缓冲区的位置和大小。
  • iovcnt: iov 数组中的元素数量。

iovec 结构体定义如下：

1
2
3
4
struct iovec {
    void  *iov_base; // 缓冲区的起始地址
    size_t iov_len;  // 缓冲区的长度
};
（2）writev函数

writev 函数将多个缓冲区的数据写入文件描述符。其函数原型如下：

1
2
3
#include <sys/uio.h>

ssize_t writev(int fd, const struct iovec *iov, int iovcnt);
  • fd: 文件描述符，指向要写入的文件或套接字。
  • iov: 指向一个或多个 iovec 结构的指针，这些结构定义了要写入的数据的位置和大小。
  • iovcnt: iov 数组中的元素数量

返回值

对于 readv 和 writev 函数，返回值是读取或写入的字节数。如果出现错误，函数返回-1，并设置 errno 以指示错误类型。如果在读取过程中到达文件末尾，readv 可能返回小于请求的字节数。

使用场景

这些函数的优势在于能够在单个系统调用中处理多个缓冲区，这减少了系统调用的开销并提高了效率。在网络编程中，尤其是在需要读取或发送格式化的数据（如多个字段的协议消息）时，readv 和 writev 非常有用。

2.内存映射IO

内存映射I/O（Memory-mapped I/O）是一种在程序中处理文件的高效方式，它将文件的内容直接映射到进程的地址空间。这种方法通过内存操作代替传统的文件I/O调用（如read和write），提高了文件操作的效率，并简化了文件内容的处理。

工作原理

内存映射I/O通过将一个文件或者其他对象（比如设备）映射到进程的虚拟地址空间来工作。这意味着文件的一部分或全部直接映射到内存地址，应用程序可以像访问普通内存那样访问这些文件数据。

关键函数

（1）mmap函数

在Unix和类Unix系统（如Linux）中，内存映射I/O主要通过mmap函数实现：

1
2
3
#include <sys/mman.h>

void *mmap(void *addr, size_t length, int prot, int flags, int fd, off_t offset);
  • addr：建议的映射起始地址（通常设为NULL，由系统选择地址）。
  • length：映射区域的长度。
  • prot：映射区域的保护方式，比如PROT_READ（可读）、PROT_WRITE（可写）。
  • flags：指示映射类型的标志，比如MAP_SHARED（对文件的修改影响到原文件）或MAP_PRIVATE（对文件的修改不影响原文件）。
  • fd：文件描述符，指向要映射的文件。
  • offset：文件中的偏移量，映射从此处开始。

返回值

  • 成功时：当 mmap 成功时，它返回映射区域的起始地址。这个地址是一个指向 void 的指针，可以被转换为任何类型的指针以便于数据访问。
  • 失败时：如果映射失败，mmap 返回 MAP_FAILED，这在大多数系统中定义为 (void *)-1。当映射失败时，通常需要检查 errno 以确定失败的原因，可能的错误原因包括无效的文件描述符、不允许的内存保护等级、映射区域过大等。
（2）mnumap函数

使用munmap函数来撤销映射：

1
int munmap(void *addr, size_t length);
  • addr: 函数mmap成功返回映射区域的起始地址
  • length: 映射区域的长度

使用mmap与mnumap函数优势：

  • 性能提升：减少了系统调用和内核空间到用户空间的数据拷贝，提高了数据访问的效率。
  • 简化操作：可以直接使用指针操作文件数据，无需繁琐的读写函数调用。
  • 动态加载：只有实际访问的内存页才会被加载，对于大文件处理尤为高效。
  • 共享内存：通过mmap可以实现不同进程之间的内存共享。
（3）实例程序

程序目标：指定任意的文件，对文件内容中的字符a进行计数

目标文件为1.txt，其内容如下：

image-20231227155757661

main.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
/*
 * 随意指定一个文件，输出文件中有多少字符a
 * */

#include<stdlib.h>
#include <stdio.h>
#include <sys/mman.h>
#include <fcntl.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>

int main(int argc,char* argv[])
{
    int fd,count = 0;
    struct stat statres;
    char *str;
    if(argc < 2)
    {
        fprintf(stderr,"USage...\n");
        exit(1);
    }

    // 获得文件的文件描述符
    fd = open(argv[1],O_RDONLY);
    if(fd < 0)
    {
        perror("open()");
        exit(1);
    }

    // 获取文件内容长度
    // fstat 函数可以获取一个已经打开的文件（由文件描述符指定）的状态信息
    // 其中 statres.st_size 表示文件中内容的长度
    if(fstat(fd,&statres) < 0)
    {
        // 失败
        perror("fstat()");
        exit(1);
    }


    // 将文件中内容进行映射
    /*
     * 1.第一个参数，通常为NULL
     * 2.第二个参数，为映射区域的长度，即想从文件中读取的内容长度
     * 3.第三个参数，映射区域的保护方式，该实例代码，只读即可
     * 4.第四个参数，指示映射类型的标志，即将文件内容映射之后，对映射空间内容进行修改是否影响源文件
     * 5.第五个参数，文件描述符，指向要映射的文件
     * 6.第六个参数，文件中的偏移量，映射从此处开始
     * */
    str = mmap(NULL,statres.st_size,PROT_READ,MAP_SHARED,fd,0);
    // 映射失败
    if(str == MAP_FAILED)
    {
        perror("mmap()");
        exit(1);
    }

    close(fd);

    // 对str中a字符进行计数
    for(int i = 0;i<statres.st_size;i++)
    {
        if(str[i]=='a')
        {
            count++;
        }
    }
    printf("文件：%s,字符a个数为：%d\n",argv[1],count);

    // 撤销内存映射
    munmap(str,statres.st_size);

    exit(0);
}

CMakeLists.txt

1
2
3
4
5
6
7
8
9
10
11
cmake_minimum_required(VERSION 3.20)
project(Adv_IO C)

set(CMAKE_C_STANDARD 99)

# 查找pthread库
# find_package(Threads REQUIRED)

add_executable(adv_io main.c)

# target_link_libraries(adv_io Threads::Threads)

运行结果

image-20231227160016481

（4）创建匿名映射

创建匿名映射

要创建一个匿名映射，通常需要在 mmap 调用中设置 MAP_ANONYMOUS（或在某些系统中是 MAP_ANON）标志，并将文件描述符设置为 -1。同时，通常与 MAP_SHARED 或 MAP_PRIVATE 标志结合使用，分别用于创建可共享或仅当前进程可用的内存区域

1
2
3
#include <sys/mman.h>

void *addr = mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_ANONYMOUS | MAP_SHARED, -1, 0);
  • NULL 表示让系统决定映射区域的起始地址。
  • size 是要映射的内存大小。
  • PROT_READ | PROT_WRITE 设置内存区域为可读写。
  • MAP_ANONYMOUS | MAP_SHARED 表示这是一个匿名且仅此进程可用的映射。
  • 文件描述符 -1 和偏移量 0 用于匿名映射。

用途

匿名映射的一些常见用途包括：

  • 替代 malloc 或其他内存分配函数：用于动态分配一大块内存，尤其是当需要的内存大小超过了 malloc 等标准函数的限制时。
  • 进程间通信：通过创建共享的匿名映射，不同的进程可以访问和修改同一块内存区域，这对于共享数据或实现进程间通信非常有用。
  • 缓冲区或临时存储：用于创建一个缓冲区，特别是当涉及到大量数据且需要避免频繁的堆分配时。

注意事项

  • 使用匿名映射时，映射的内存区域通常初始化为零。
  • 与任何通过 mmap 创建的映射一样，使用完毕后应该通过 munmap 来释放映射。
  • 匿名映射是操作系统特定的功能，主要在类Unix系统（如Linux）中可用。
（5）mmap实现父子进程的内存共享

使用mmap实现父子进程的进程通信

思路：使用mmap中的匿名映射，使用 mmap 函数进行匿名映射（Anonymous Mapping）时，映射的内存区域不与任何文件关联。这种映射通常用于分配一块可以被多个进程共享的内存，或者仅在当前进程中作为一种动态分配的、可调整大小的内存区域。在父进程中使用mmap函数在父进程中进行匿名映射，在使用fork函数创建出子进程，子进程会将父进程的资源进行复制，因此从文件中映射的内容也会进行复制，这个时候就可以通过mmap函数进行进程间的通信。

main.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
/*
 * mmap函数匿名映射 实现父子进程的进程通信
 * 父进程进行读，子进程进行写
 * */

#include <stdio.h>
#include <stdlib.h>
#include <sys/mman.h>

#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/wait.h>

#define MEMSIZE 1024

int main()
{
    char *ptr;
    pid_t pid;
    // 匿名映射，不依赖于任何文件 flags参数需要或上MAP_ANONYMOUS fd写为-1
    ptr = mmap(NULL, MEMSIZE, PROT_READ | PROT_WRITE, MAP_ANONYMOUS | MAP_SHARED, -1, 0);
    // 失败
    if(ptr == MAP_FAILED)
    {
        perror("mmap()");
        exit(1);
    }

    // 创建子进程
    pid = fork();
    // 失败
    if(pid < 0)
    {
        perror("fork()");
        // 撤销父进程的内存映射
        munmap(ptr,MEMSIZE);
        exit(1);
    }

    // 子进程返回0
    if(pid == 0)
    {
        // 子进程中的操作
        // 在子进程中进行写入
        // 对字符串指针指向的内存空间进行写入操作
        strcpy(ptr,"HELLO!");
        // 子进程中解除映射
        munmap(ptr,MEMSIZE);
        // 退出子进程
        exit(0);
    }else
    {
        // 对子进程中资源进行回收
        wait(NULL);
        // 父进程中的操作
        // 父进程中进行读取
        puts(ptr);
        // 父进程中解除映射
        munmap(ptr,MEMSIZE);
        // 退出父进程
        exit(0);
    }

    exit(0);
}

运行结果：

image-20231227165039892

3.文件锁

文件锁是一种用于控制对文件或文件一部分的访问的机制，主要用于同步多个进程或线程对同一文件的访问。在多用户或多任务的操作系统中，文件锁非常重要，因为它们可以防止同时运行的进程互相干扰，从而确保数据的一致性和完整性

主要类型

  • 排他锁（Exclusive Locks）：也称为写锁。当一个进程获得了文件的排他锁时，其他任何进程都不能读取或写入该文件。这保证了锁持有者可以安全地读取或修改文件。

  • 共享锁（Shared Locks）：也称为读锁。多个进程可以同时获得同一文件的共享锁。共享锁允许这些进程读取文件，但不允许写入。如果一个进程需要写入文件，它必须等待所有共享锁被释放，并获得一个排他锁。

实现方法

  • POSIX锁（fcntl锁）：通过fcntl函数实现，支持对文件任意区域的锁定。POSIX锁是建议性锁（Advisory Locks），意味着它们不会阻止其他进程操作文件，除非这些进程也使用相同的锁机制。

    1
    2
    3
    4
    5
    6
    struct flock lock;
    lock.l_type = F_WRLCK;    // 设置为写锁
    lock.l_start = 0;         // 锁定文件起始位置
    lock.l_whence = SEEK_SET; // 锁定位置相对于文件开头
    lock.l_len = 0;           // 锁定整个文件
    fcntl(fd, F_SETLK, &lock);

  • 系统V锁（System V Locks）：通过lockf或flock函数实现。这些锁通常是全文件锁，并且是建议性锁

    1
    flock(fd, LOCK_EX); // 获取排他锁
（1）lockf函数

lockf函数

1
2
3
#include <unistd.h>

int lockf(int fd, int cmd, off_t len);
  • fd：文件描述符，指向要锁定或解锁的文件。
  • cmd：操作命令，指定要执行的锁定操作。
  • len：锁定或解锁的字节数。

操作命令

cmd 参数支持以下几种操作：

  • F_LOCK：对指定的文件区域加上排他锁。如果锁已被其他进程持有，lockf 会阻塞，直到能够获取锁。
  • F_TLOCK：尝试对指定文件区域加上排他锁。如果锁已被其他进程持有，函数会立即返回，不会阻塞。
  • F_ULOCK：解锁指定的文件区域。
  • F_TEST：测试指定的文件区域是否被其他进程锁定。

返回值

  • 成功时：返回0。
  • 失败时：返回-1，并设置 errno 来指示错误原因。例如，如果文件描述符无效，errno 可能被设置为 EBADF。
（2）实例程序

多进程并发对文件../out,中的数字进行逐次+1，若最开始文件中数字为1,程序运行结束之后数字应为20

main.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
/*
 * 多进程并发
 * */
#include <stdio.h>
#include <stdlib.h>

#include <string.h>
#include <unistd.h>

#include <sys/types.h>
#include <sys/wait.h>


#define PROCNUM 20
#define FNAME "../out"
#define LINESIZE 1024


// 线程函数
static void func_add(void)
{
    int fd;
    FILE *fp;
    char linebuf[LINESIZE];

    // 打开文件
    // fopen用于将文件流（例如，由 fopen 创建的 FILE* 类型的对象）转换成对应的文件描述符
    fp = fopen(FNAME,"r+");
    if(fp ==NULL)
    {
        perror("fopen()");
        exit(1);
    }

    // fileno函数可以将文件流对应的文件描述符
    fd = fileno(fp);
    if(fd < 0)
    {
        perror("fileno()");
        exit(1);
    }

    // 临界区进行加文件锁的操作 对指定的文件区域加上排他锁
    /*
     * 参数1: 指定的文件区域，文件描述符
     * 参数2: cmd操作命令，排他锁
     * 参数3: 锁定或解锁的字节数,若为0，表示当前长度到未来长度，即所有内容
     * */
    lockf(fd,F_LOCK,0);

    // 读取文件中的一行
    fgets(linebuf,LINESIZE,fp);
    // 定位到文件开始的位置进行覆盖写入
    fseek(fp,0,SEEK_SET);
    sleep(1);
    // 将linfbuf字符串中的转为整形+1 在写入文件流fp中
    fprintf(fp,"%d\n",atoi(linebuf)+1);
    // fptintf为行缓冲模式，而文件是全缓冲模式，需要刷新流
    fflush(fp);
    // 解开文件锁
    lockf(fd,F_ULOCK,0);

    fclose(fp);

    return;
}

int main()
{
    int i;
    pid_t pid;

    for(i = 0;i<PROCNUM;i++)
    {
        pid = fork();
        // 进程创建失败
        if(pid < 0)
        {
            perror("fork()");
            exit(1);
        }
        // 子进程
        if(pid == 0)
        {
            func_add();
            exit(0);
        }
    }


    // 回收线程资源（收尸）
    for(i = 0;i<PROCNUM;i++)
    {
        wait(NULL);
    }

    exit(0);
}

fileno函数

fileno 函数是一个标准库函数，用于将文件流（例如，由 fopen 创建的 FILE* 类型的对象）转换成对应的文件描述符。这个函数主要在需要将高级I/O函数（如 fprintf, fscanf）与低级I/O函数（如 read, write）混合使用的场景中非常有用

fileno 函数的原型定义在 <stdio.h> 头文件中，如下所示：

1
2
3
#include <stdio.h>

int fileno(FILE *stream);

五、管道

管道可以看作队列

管道（Pipe）是一种用于在进程之间传输数据的机制。管道可以被视为一个特殊的文件，其中一个进程可以写入数据，而另一个进程则从中读取数据。管道主要用于实现进程间通信（IPC，Inter-Process Communication），尤其是在类Unix操作系统（如Linux）中。

管道分为两种主要类型：匿名管道（Anonymous Pipes）和命名管道（Named Pipes 或 FIFOs）。

匿名管道（Anonymous Pipes）

  • 用途：仅用于有父子关系的进程间通信。
  • 特点：它们是临时的，只存在于进程间通信的持续期间。
  • 实现：在Unix/Linux系统中，通常通过pipe()系统调用创建。一个进程创建管道后，它会产生两个文件描述符：一个用于读取，另一个用于写入。

命名管道（Named Pipes 或 FIFOs）

  • 用途：可用于任何两个希望通信的进程之间，无需具有父子关系。
  • 特点：以文件形式存在于文件系统中，因此它们在创建后可以被任何进程访问。
  • 实现：在Unix/Linux系统中，通常通过mkfifo命令或mkfifo()系统调用创建。