AUPE-第14章-高级IO

本章主要讨论的是高级I/O话题，有：非阻塞I/O、记录锁、I/O多路转接、异步I/O、存储映射I/O等。

非阻塞I/O

系统调用可以分为：”低速“系统调用和其他，低速系统调用是指可能会使进程永远阻塞的系统调用，对于像读写磁盘文件的I/O会暂时阻塞调用者，不能称为低速I/O。

非阻塞I/O可以在我们使用open、read、write等I/O操作时，保证这些操作不会阻塞。如果该操作不能完成，调用会立即出错并返回，表示该操作如果继续就会阻塞。

实例

下面是使用非阻塞I/O的实例，它从标准输入中读取500000字节，然后试图将它们写到标准输出上，它将标准输出设置为非阻塞

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
#include "apue.h"
#include <errno.h>
#include <fcntl.h>

char buf[500000];

int
main(void)
{
    int     ntowrite, nwrite;
    char    *ptr;
	//read系统调用：read(int fd, void *buf, size_t nbytes)
    ntowrite = read(STDIN_FILENO, buf, sizeof(buf));
    fprintf(stderr, "read %d bytes\n", ntowrite);
	
    /* 设置SETOUT_FILLENO为非阻塞,
     * 在linux中，将文件描述符STDOUT_FILENO设置为O_NONBLOCK，即可将其设置为非阻塞状态
     */
    /*
    	void set_fl(int fd, int flags)
		{
			int		val;

			if ((val = fcntl(fd, F_GETFL, 0)) < 0)
				err_sys("fcntl F_GETFL error");

			val |= flags;		

			if (fcntl(fd, F_SETFL, val) < 0)
				err_sys("fcntl F_SETFL error");
		}
    */
    set_fl(STDOUT_FILENO, O_NONBLOCK);

    ptr = buf;
    while (ntowrite > 0) {
        errno = 0;
        //write系统调用：write(int fd, const void *buf, size_t nbytes)
        nwrite = write(STDOUT_FILENO, ptr, ntowrite);
        fprintf(stderr, "nwrite = %d, errno = %d\n", nwrite, errno);

        if (nwrite > 0) {
            ptr += nwrite;
            ntowrite -= nwrite;
        }
    }

    /*	void
		clr_fl(int fd, int flags)
		{
			int		val;

			if ((val = fcntl(fd, F_GETFL, 0)) < 0)
				err_sys("fcntl F_GETFL error");

			val &= ~flags;

			if (fcntl(fd, F_SETFL, val) < 0)
				err_sys("fcntl F_SETFL error");
		}
     */
    clr_fl(STDOUT_FILENO, O_NONBLOCK);  //清理SETOUT_FILENO的非阻塞文件状态

    exit(0);
}

这里使用while循环的方式进行调用write函数，该方式称为轮询，若标准输出是终端时（因为终端是行缓冲，超过缓冲上限，缓冲会被冲洗，冲洗时write调用就会失败），会反复调用write系统调用，并且大多数会返回错误，这会浪费CPU的时间，后续会讲到可以使用I/O多路转接，很好的解决这类问题。

记录锁

记录所（record locking，又称字节范围锁）的功能是：当一个进程正在读或写文件的某个部分时，记录锁可以阻止其他进程修改同一文件区。注意，这里是锁住文件区域，可以是一个文件，也可以是一个文件中的一个字节。

在Linux中可以使用fcntl方法设置记录锁。

1
2
3
4
#include <fcntl.h>

int fcntl(int fd, int cmd, ... /* struct flock *flockptr */);
								返回值：若成功，依赖cmd，否则，返回-1

对于记录锁，cmd是F_GETLK、F_SETLK、F_SETLKW，第三个参数是指向flock结构的指针：

1
2
3
4
5
6
7
struct flock {
    short l_type;	/* 锁类型：F_RDLOC—共享读锁、F_WRLCK—独占性写锁、F_UNLCK—解锁一个区域 */
    short l_whence; /* 文件偏移位置：SEEK_SET—文件首位、SEEK_CUR—文件当前位置、SEEK_END—文件末尾 */
    off_t l_start;  /* 加锁或解锁区域的起始字节偏移量 */
    off_t l_len;	/* 加锁区域字节长度，若为0，表示可以写任意数据，不必猜测会有多少数据被写入 */
    pid_t l_pid;	/* 进程ID，指持有锁的进程，仅由F_GETLK返回 */
};

由l_type可知，记录锁的效果和线程中的读写锁效果类似，读锁共享，写锁独占。

若一个进程在同一个文件区两次加锁，则新锁会替换旧锁。

fcntl的cmd参数可以有下面3个（记录锁情况下）：

  • F_GETLK：用于判断是否创建锁，如果相应的位置已经有锁存在，则将现有锁的信息重写flockptr中，4；如果没有锁，则将flockptr中的l_type修改为F_UNLCK，表示该锁可以被获取。
  • F_SETLK：用户尝试向文件建立锁，如果系统阻止我们获取锁，则fcntl立即出错返回，errno设置为EACCES或EAGAIN。
  • F_SETLKW：F_SETLK的阻塞版本（后面的W即wait），如果进程企图加锁的区域被其他进程占有而导致无法获取，则该进程进入阻塞状态，直到锁可用或被信号唤醒。

注意，若想使用F_GETLK测试是否可以获取锁，然后用F_SETLK或F_SETLKW获取锁，这两者之间不是原子操作，不能保证在两个操作之间没有其他进程企图获取相同的锁。

实例：死锁

如果两个进程相互等待对方持有并且不释放锁定的资源时，则两个进程就会处于死锁状态。下面是死锁的例子，子进程对第0字节加锁，父进程对第1字节加锁，并且它们试图向对方加锁的字节加锁。我在原书的基础上添加了一些打印动作，以便直观的看到父进程和子进程的动作。

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
#include "apue.h"
#include <fcntl.h>

static void lockabyte(const char *name, int fd, off_t offset)
{
    if (writew_lock(fd, offset, SEEK_SET, 1) < 0)
        err_sys("%s: writew_lock error", name);
    printf("%s: got the lock, byte %lld\n", name, (long long)offset);
}

int main(void)
{
    int     fd;
    pid_t   pid;

    if ((fd = creat("templock", FILE_MODE)) < 0)
        err_sys("creat error");
    if ((write(fd, "ab", 2)) != 2)
        err_sys("write error");
	/* 该操作为自定义函数，具体作用是：
	 * 创建两个信号SIGUSR1和SIGUSR2，并设置信号处理程序，处理程序的功能是将一个信号标记(sigflag)设置为1
	 * 然后阻塞这两个信号，为了防止在其他操作前收到SIGUSR1或SIGUSR2，从而改变sigflag的值
	 */
    TELL_WAIT();
    if ((pid = fork()) < 0) {
        err_sys("fork error");
    } else if (pid == 0) {
        lockabyte("child", fd, 0);  //子进程锁住第0处的字节
        /* 自定义函数，功能是：向父进程发送SIGUSR2信号 */
        TELL_PARENT(getppid());	
        /* 自定义函数，功能是：子进程先进入休眠状态，并恢复信号屏蔽字，这里目的是释放对SIGUSR1和SIGUSR2的屏蔽，子进程可以收到父进程发送的SIGUSR1信号 */
        WAIT_PARENT();
        printf("子进程尝试获取字节1\n");
        lockabyte("child", fd, 1);	//尝试锁住父进程控制的第1处的字节
    } else {
        lockabyte("parent", fd, 1);	//父进程锁住第1处的字节
        TELL_CHILD(pid);			//与上同理，向子进程发送SIGUSR1信号
        WAIT_CHILD();				//与上同理，释放信号屏蔽字，父进程可以收到子进程发送的SIGUSR2信号
        printf("父进程尝试获取字节0\n");
        lockabyte("parent", fd, 0);	//尝试锁住子进程控制的第0处的字节
    }

    exit(0);
}

下面是上述自定义函数的实现，也添加的一些打印动作：

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
#include "apue.h"

static volatile sig_atomic_t sigflag;
static sigset_t newmask, oldmask, zeromask;

static void sig_usr(int signo)
{
	sigflag = 1;
}
/* 添加两个信号，并将这两个信号阻塞 */
void TELL_WAIT(void)
{
	/* 创建两个信号及其相应的信号处理程序 */
	if (signal(SIGUSR1, sig_usr) == SIG_ERR)
		err_sys("signal(SIGUSR1) error");
	if (signal(SIGUSR2, sig_usr) == SIG_ERR)
		err_sys("signal(SIGUSR2) error");

	sigemptyset(&zeromask);
	sigemptyset(&newmask);

	/* 向newmask中添加要阻塞的信号 */
	sigaddset(&newmask, SIGUSR1);
	sigaddset(&newmask, SIGUSR2);	
	
	/* 通过newmask添加期望阻塞的附加信号，这里是添加SIGUSR1和SIGUSR2 */
	if (sigprocmask(SIG_BLOCK, &newmask, &oldmask) < 0)
		err_sys("SIG_BLOCK error");
}

void TELL_PARENT(pid_t pid)
{
	/* 将信号SIGUSR2发送给父进程 */
	printf("子进程发送信号SIGUSR2\n");
	kill(pid, SIGUSR2);
}

void WAIT_PARENT(void)
{
	while (sigflag == 0) {
		printf("子进程进入休眠\n");
		sigsuspend(&zeromask);
	}
	printf("子进程解除休眠\n");
	sigflag = 0;
	/* 恢复信号屏蔽字 */
	if (sigprocmask(SIG_SETMASK, &oldmask, NULL) < 0)
		err_sys("SIG_SETMASK error");
}

void TELL_CHILD(pid_t pid)
{
	//将信号SIGUSR1发送给子进程
	printf("父进程发送信号SIGUSR1\n");
	kill(pid, SIGUSR1);
}

void WAIT_CHILD(void)
{
	while (sigflag == 0) {
		printf("父进程进入休眠\n");
		sigsuspend(&zeromask);	//此时可以接受所有信号
	}
	printf("父进程解除休眠\n");
	sigflag = 0;
	if (sigprocmask(SIG_SETMASK, &oldmask, NULL) < 0)	//恢复信号屏蔽字
		err_sys("SIG_SETMASK error");
}

在linux中，上述的程序输出为：

parent: got the lock, byte 1
父进程发送信号SIGUSR1
父进程进入休眠
child: got the lock, byte 0
子进程发送信号SIGUSR2
子进程进入休眠
子进程解除休眠
子进程尝试获取字节1
父进程解除休眠
父进程尝试获取字节0
parent: writew_lock error: Resource deadlock avoided
child: got the lock, byte 1

检查到死锁时，内核必须选择一个进程接受出错返回，这里内核决定的是父进程出错返回，子进程成功获取父进程控制的字节。

锁的隐含继承和释放

记录锁的自动继承和释放有3条规则：

  • 锁与进程和文件两者关联，这里有两重含义：（1）当一个进程终止，其建立的锁全部释放；（2）一个文件描述符关闭时，进程通过该文件描述符引用的文件上的锁都会被释放。如下：

    1
    2
    3
    4
    fd1 = open(pathname, ...);
    read_lock(fd1, ...);	//自定义函数，功能是：在fd1上创建一个读锁
    fd2 = dup(fd1);		//复制一个文件描述符
    close(fd2);				//关闭fd2关联的文件

​ 在执行close(fd2)后，通过fd1创建的锁也会被释放，因为fd2和fd1指向同一个文件

  • 由fork产生的子进程不继承父进程锁设置的锁。这是有意义的，因为锁本身的目的就是为了阻止多个进程同时写同一个文件，如果子进程继承了父进程的锁，那么就会导致有两个进程同时写同一个文件。
  • 在执行exec后，新程序可以继承原执行程序的锁。这可以理解为新创建了一个进程，原进程已经终止。注意，如果对文件描述符设置了执行时关闭表示，则exec后，该文件描述符会被关闭，并释放所有锁。

在文件尾端加锁

文件尾端会一直变化，因此在向文件尾端加锁或解锁时需要小心。考虑下面代码：

1
2
3
4
writew_lock(fd, 0, SEEK_END, 0);	//自定义函数，功能是：向文件尾端添加写锁
write(fd, buf, 1);					//向文件尾端写一个字节
un_lock(fd, 0, SEEK_END);			//释放文件尾端的写锁，注意，此时的文件尾端已经变化
write(fd, buf, 1);					//向新的文件尾端写一个字节

在文件尾端添加写锁，后续向文件写的任何数据也会被锁上。上述代码的效果如下：

文件区域锁

如果想要解除包括第一次write所写字节的锁，则在un_lock函数中的第二个参数设置为-1，表示解锁的区域从当前位置（这里是文件末尾）的上一个字节开始，这样就可以释放所有锁了。

I/O多路转接

当从一个文件描述符读，然后写到另一个文件描述符，可以使用下述的阻塞I/O：

1
2
3
while ((n = read(STDIN_FILENO, buf, BUFSIZ)) > 0)
    if (write(STDOUT_FILENO, buf, n) != n)
        err_sys("write error");

但是，如果必须从两个文件描述符读，就不能使用这种阻塞I/O了，因为我们不能在一个描述符上阻塞read，如果此时另一个文件描述符有数据，就无法调用read进行处理，例如：

telnet程序

telnet程序有两个输入，两个输出。因为不知道是哪一个输入会有数据，不能对两个输入的任何一个进行阻塞。

解决这个问题较好的技术是I/O多路转接。先构造一个描述符列表，然后调用一个函数，直到这些描述符中的一个已经准备好I/O时，函数才返回。poll、pselect、select这3个函数可以执行I/O多路转接。

函数select和pselect

通过select参数可以告诉内核：

  • 我们关心的描述符；
  • 关心的描述符的条件（读、写、异常）；
  • 愿意等待的时间（永远、一段时间、不等待）。

select返回后，内核告诉我们：

  • 已准备好的描述符数量
  • 对于读、写、异常三个条件中哪些描述符已经准备好

根据select返回的信息，就可以调用相应的I/O函数，并且保证该函数不会阻塞。

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

int select(int maxfdp1, fd_set *restrict readfds, fd_set *restrict writefds,
           fd_set *restrict exceptfds, struct timeval *restrict tvptr);
								返回值：准备就绪的描述符数目；若超时，返回0；若出错，返回-1

对于参数tvptr有三种情况：

  • tvptr == NULL：永远等待。当指定描述符中一个已准备好或捕捉到一个信号则返回。如果捕捉到信号，则返回-1，errno设置为EINTR。
  • tvptr->tv_sec == 0 && tvptr->tv_usec == 0：根本不等待，测试所有指定描述符后立即返回。用于轮询找到多个描述符状态而不阻塞select的方法；
  • tvptr->tv_sec != 0 || tvptr->tv_usec != 0：等待指定的描述和微秒数。当指定描述符已准备好，或指定时间超时后返回。若超时后没有描述符准备好，则返回0。

中间三个参数readfds、writefds、execptfds是指向描述符集的指针，每个描述符集存储在fd_set结构中，可以认为它是一个很大的数组。

select的读、写、异常描述符集

参数maxfdp1表示”最大文件描述符编号值加1“。通过我们给定最大描述符，则内核只需要在此范围内寻找即可，而不需要在没有使用的位内搜索。

select函数有三种返回值：

  • 返回值-1，表示出错。例如没有描述符准备好时捕捉到一个信号，此时一个描述符集都不修改；
  • 返回值0，表示没有描述符准备好。超时后，一个描述符都没准备好，此时描述符集都置0；
  • 返回值正数，表示准备好的描述符数，3个描述符集已准备好的描述符之和，若同一描述符准备好了读和写，则返回值计数两次。此时，描述符集对应已准备好的描述符置1。

pselect是select的变体，它可以安装信号屏蔽字。

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

int pselect(int maxfdp1, fd_set *restrict readfds, fd_set *restrict writefds,
            fd_set *restrict execptfds, const struct timespec *restrict tsptr, 
            const sigset_t *restrict sigmask);
								返回值：准备好的描述符数目；若超时，返回0；若出错，返回-1

pselect和select有以下不同：

  • select超时值使用timeval，pselect使用timespec。timeval使用秒和微秒，timespec使用秒和纳秒；
  • pselect的超时值声明为const，保证pselect不会修改该值；
  • pselect可以使用信号屏蔽字，pselect保证以原子方式安装屏蔽字，返回后，恢复以前的信号屏蔽字。

函数poll

poll类似select，但接口不同：

1
2
3
4
#include <poll.h>

int poll(struct pollfd fdarry[], nfds_t nfds, int timeout);
								返回值：准备就绪的描述符数目；若超时，返回0；若出错，返回-1

与select不同，poll通过pollfd数组，每个数组元素指定一个描述符编号和对描述符感兴趣的条件。

1
2
3
4
5
struct pollfd {
    int fd;			//文件描述符
    short events;	//fd中感兴趣的事件
    short revents;	//fd中发生的事件
};

fdarry数组中元素个数由nfds指定。pollfd中的events告诉内核我们关心的描述符对应的事件。返回时，内核设置revents，说明对应描述符发生的事件。（注意，poll没有修改events成员）。

poll中的timeout表示我们愿意等待的时间。如同select，有3个情形：

  • timeout == -1：永远等待。当指定描述符中的一个已准备好，或捕捉到一个信号，则返回。如果捕捉到信号，返回-1，并且errno设置为EINTR；
  • timeout == 0：不等待。测试所有描述的状态（从revents获得），并不阻塞poll函数；
  • timeout > 0：等待timeout毫秒，若给定描述符之一已准备好，或超时后，立即返回。若超时后没有描述符准备好，则poll返回0。

与select相同，一个文件描述符阻塞并不影响poll阻塞。

异步I/O

异步I/O使用一个信号（System V中是SIGPOLL，BSD中是SIGIO）通知进程，表示某个描述符关心的时间已经发生。但信号只有一个，当如果有多个描述符使用异步I/O，进程接收到该信号时不知道其对应的是哪一个文件描述。

POSIX异步I/O

异步I/O接口使用AIO控制块来描述I/O操作，aiocb结构描述了AIO控制块，至少包括一下字段：

1
2
3
4
5
6
7
8
9
struct aiocb {
    int 			aio_fileds;		//文件描述符
    off_t 			aio_offset;		//文件偏移量，表示读写操作的起始地址
    volatile void 	*aio_buf;		//I/O的缓冲区，读操作将数据复制进缓冲区，写操作从缓冲区复制出来
    size_t 			aio_nbytes;		//传输的字节数
    int				aio_reqprio;	//I/O的优先级
    struct sigevent aio_sigevent;	//signal信息，表示在I/O完成后，如何通知应用程序
    int				aio_lio_opcode;	//I/O操作列表
};

异步I/O接口的偏移量并不影响操作系统维护的文件偏移量。只要在一个进程中，不将异步I/O函数和传统I/O函数（指read、write）一起使用，就不会出问题。

aio_sigevent字段结构如下：

1
2
3
4
5
6
7
struct sigevent {
    int 			sigev_notify;					//通知类型
    int 			sigev_signo;					//信号number，用于异步I/O完成后指定信号
    union sigval	sigev_value;					//notify argument
    void (*sigev_notify_function)(union sigval);	//通知函数
    pthread_attr_t *sigev_notify_attributes;		//通知属性
}

sigve_notify字段控制通知的类型，取值有3种：

  • SIGEV_NONE：异步I/O完成后，不通知进程；
  • SIGEV_SIGNAL：异步I/O完成后，产生由sigev_signo字段指定的信号；
  • SIGEV_THREAD：异步I/O完成后，由sigev_notify_function字段指定函数被调用，sigev_value作为它的唯一参数。

函数aio_read进行异步读操作，函数aio_write进行异步写操作。

1
2
3
4
5
#include <aio.h>

int aio_read(struct aiocb *aiocb);
int aio_write(struct aiocb *aiocb);
								返回值：若成功，返回0；若出错，返回-1

当函数返回成功时，异步I/O请求被操作系统放入等待队列中。注意，两个函数的返回值与I/O操作无任何关系，在I/O完成之前，AIO控制块和缓冲区不能被复用。

函数aio_fsync可以强制所有等待中的异步操作立即执行写入持久化存储过程，也就是执行数据同步操作。

1
2
3
4
#include <aio.h>

int aio_fsync(int op, struct aiocb *aiocb);
								返回值：若成功，返回0；若出错，返回-1

在异步同步操作完成前，数据不会被持久化。

函数aio_error可以获取异步读、写、同步操作的完成状态。

1
2
3
4
#include <aio.h>

int aio_error(const struct aiocb *aiocb);
								返回值：如下

aio_error有返回值有四种情况：

  • 0：表示异步操作（指读、写、同步等操作）成功完成，此时可以调用aio_return获取异步操作返回值；
  • -1：aio_error调用失败，可以从error获取与原因值；
  • EINPROGRESS：异步写、读、同步操作正在等待中；
  • 其他情况：其他返回值是异步操作（指读、写、同步操作）失败返回的错误码。

异步操作成功后，可以调用aio_return获取异步操作返回值

1
2
3
4
#include <aio.h>

ssize_t aio_return(cosnt struct aiocb *aiocb);
								返回值：如下

aio_return的返回值

  • -1 ： aio_return调用失败，并设置errno；
  • 其他：返回异步操作的结果，即读、写、同步操作的返回结果

注意，在异步操作完成之前，不要调用aio_return，此时操作未定义；并且对一个异步操作只能调用一次aio_return。调用该函数后，操作系统会删除I/O操作的返回值。

执行I/O操作时，不想被阻塞就可以使用异步I/O。当所有事务都完成，还有异步操作没有完成，则可以调用aio_suspend阻塞进程，直到异步操作完成。

1
2
3
4
5
#include <aio.h>

int aio_suspend(const struct aiocb *const list[], int nent,
                const struct timespec *timeout);
								返回值：若成功，返回0；若出错，返回-1

如果调用aio_suspend的阻塞过程中，被信号中断，则它返回-1，并在errno中设置EINTR；

如果没有任何的I/O操作完成，阻塞时间超过timeout参数，则它返回-1，并将errno设置EAGAIN（不想设置时间限制，可以将timeout传入为NULL）；

如果任何I/O操作完成，则它返回0；

如果在调用aio_suspend时，所有异步I/O以完成，则aio_suspend不阻塞直接返回。

参数list表示指向aiocb数组的指针，参数nent表示数组中的条目数量，除了空指针，其他条目必须指向初始化I/O操作的AIO控制块。

函数aio_cancel可以取消等待中的异步I/O操作。

1
2
3
4
#include <aio.h>

int aio_cancel(int fd, struct aiocb *aiocb);
								返回值：如下

aio_cancel返回值有：

  • AIO_ALLDONE：所有操作在尝试取消它们前已完成
  • AIO_CANCELED：所有请求的操作已被取消
  • AIO_NOTCANCELED：至少一个请求的操作没被取消
  • -1：aio_cancel调用失败，并在errno中设置错误码

参数fd指定了执行异步操作的文件描述符，如果aiocb设置为NULL，则系统尝试取消fd指向的文件上的所有异步操作。其他情况下，系统尝试取消单个异步操作。之所以描述为“尝试”，因为操作系统无法保证能成功取消正在进程中的异步操作。

aio_cancel操作成功，对相应的AIO控制块调用aio_error会返回错误ECANCELED。如果操作不成功，AIO控制块无变化。

函数lio_listio可以提交一系列有AIO控制块列表描述的I/O请求。

1
2
3
4
5
#include <aio.h>

int lio_listio(int mode, struct aiocb *restrict const list[restrict],
               int nent, struct sigevent *restrict sigev);
								返回值：若成功，返回0；若出错，返回-1

参数mode有：

  • LIO_WAIT：函数将在列表指定的所有I/O完成后返回；
  • LIO_NOWAIT：函数将I/O操作插入等待队列后立即返回，进程将在对应I/O完成后，由sigev参数决定如何异步通知。如果进程不想被通知，则将sigev设置为NULL。注意，每个AIO对应也有其各自操作完成时的异步通知，sigev参数的异步通知是另加的，并且只会在所有I/O操作完成后发送。

参数list指向AIO控制块列表，指代所有要进行的I/O操作。

参数nent指定数组元素的个数，如果list为NULL，该参数被忽略。

引入POSIX异步操作I/O接口的目的是为了避免在执行I/O操作时阻塞进程。

实例

下面使用异步I/O翻译一个文件

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
#include "apue.h"
#include <ctype.h>
#include <fcntl.h>
#include <aio.h>
#include <errno.h>

#define BSZ 4096
#define NBUF 8

//用于表示异步操作所处的状态
enum rwop
{
    UNUSED = 0,
    READ_PENDING = 1,
    WRITE_PENDING = 2
};

struct buf
{
    enum rwop op;			//异步操作的状态
    int last;				//是否达到文件尾的标志
    struct aiocb aiocb;		//异步I/O的aio控制块
    unsigned char data[BSZ];//文件缓冲区，异步读时往data写数据，异步写时从data读数据
};

struct buf bufs[NBUF];		//异步操作的集合，表示最多同事有NBUF个异步操作

unsigned char translate(unsigned char c) //ROT-13翻译算法
{
    if (isalpha(c))
    {
        if (c >= 'n')
            c -= 13;
        else if (c >= 'a')
            c += 13;
        else if (c >= 'N')
            c -= 13;
        else
            c += 13;
    }
    return (c);
}

int main(int argc, char *argv[])
{
    int ifd, ofd, i, j, n, err, numop;
    struct stat sbuf;
    const struct aiocb *aiolist[NBUF];
    off_t off = 0;

    if (argc != 3)
        err_quit("usage: rot13 infile outfile");
    if ((ifd = open(argv[1], O_RDONLY)) < 0)
        err_sys("can't open %s", argv[1]);
    if ((ofd = open(argv[2], O_RDWR | O_CREAT | O_TRUNC, FILE_MODE)) < 0) //创建临时文件
        err_sys("can't create %s", argv[2]);
    if (fstat(ifd, &sbuf) < 0) //获得指定文件的相关信息
        err_sys("fstat failed");

    /* 初始化缓冲区 */
    for (i = 0; i < NBUF; i++)
    {
        bufs[i].op = UNUSED;			//标记缓冲区状态为未使用
        bufs[i].aiocb.aio_buf = bufs[i].data;	//将bufs[i].data设为用户I/O的缓冲区
        bufs[i].aiocb.aio_sigevent.sigev_notify = SIGEV_NONE;	//信号设为不通知
        aiolist[i] = NULL;	//AIO控制块数组，作为后续aio_suspend等待异步I/O操作的参数
    }

    numop = 0;		//执行中异步I/O的数量
    for (;;)
    {
        for (i = 0; i < NBUF; i++)
        {
            switch (bufs[i].op)
            {
            case UNUSED:
                //此时缓冲区未使用，因此可以用于异步I/O，此处为异步读
                if (off < sbuf.st_size)
                {
                    bufs[i].op = READ_PENDING;			//设置异步操作的状态，此时为异步读
                    bufs[i].aiocb.aio_fildes = ifd;		//要读的文件标识符
                    bufs[i].aiocb.aio_offset = off;		//要读的文件偏移位置
                    off += BSZ;							//将偏移位置向后偏移BSZ，用于其他的异步读操作
                    if (off >= sbuf.st_size) //如果此次读取到达文件尾，设置结尾标志
                        bufs[i].last = 1;
                    bufs[i].aiocb.aio_nbytes = BSZ;		//此次读的字节数
                    if (aio_read(&bufs[i].aiocb) < 0)	//发起异步读
                        err_sys("aio_read failed");
                    aiolist[i] = &bufs[i].aiocb;		//此时有异步操作，标记aiolist，用于后续aio_suspend阻塞进程
                    numop++;	//此时有一个异步操作，因此+1
                }
                break;

            case READ_PENDING:
                if ((err = aio_error(&bufs[i].aiocb)) == EINPROGRESS) //表示异步操作正在等待
                    continue;
                if (err != 0)		//异步操作出错
                {
                    if (err == -1)
                        err_sys("aio_error failed");
                    else
                        err_exit(err, "read failed");
                }
                //异步读操作成功，接下来进行异步写
                if ((n = aio_return(&bufs[i].aiocb)) < 0) //如果aio_return调用失败，则返回-1，如果大于0，则aio_return返回的是read操作的结束
                    err_sys("aio_return failed");
                if (n != BSZ && !bufs[i].last)	//如果读取的字节数小于4096并且文件没有到达文件尾，说明读取错误，返回
                    err_quit("short read (%d/%d)", n, BSZ);
                for (j = 0; j < n; j++)	//翻译异步读的字节
                    bufs[i].data[j] = translate(bufs[i].data[j]);
                bufs[i].op = WRITE_PENDING;	//设置新的异步操作，即异步写
                bufs[i].aiocb.aio_fildes = ofd;	//异步写的目标文件标识符
                bufs[i].aiocb.aio_nbytes = n;	//异步写的字节数
                if (aio_write(&bufs[i].aiocb) < 0)	//发起异步写操作
                    err_sys("aio_write failed");
                break;

            case WRITE_PENDING:
                if ((err = aio_error(&bufs[i].aiocb)) == EINPROGRESS) //同上
                    continue;
                if (err != 0)	//同上
                {
                    if (err == -1)
                        err_sys("aio_error failed");
                    else
                        err_exit(err, "write failed");
                }
                //write已完成，标记buffer为未使用
                if ((n = aio_return(&bufs[i].aiocb)) < 0)//同上
                    err_sys("aio_return failed");
                if (n != bufs[i].aiocb.aio_nbytes)		//同上
                    err_quit("short write (%d/%d)", n, BSZ);
                aiolist[i] = NULL;		//将对应的设为NULL，表示该异步I/O已完成
                bufs[i].op = UNUSED;	//将缓冲区标记为未使用，以供其他异步操作使用
                numop--;			//执行中的异步操作-1
                break;
            }
        }
        if (numop == 0)
        {
            if (off >= sbuf.st_size)        //如果进行中的异步操作为0，并且off已经到达文件尾，则退出循环
                break;
        }
        else
        {
            //如果bufs中有异步操作没有完成，则进程阻塞
            if (aio_suspend(aiolist, NBUF, NULL) < 0)
                err_sys("aio_suspend failed");
        }
    }
    bufs[0].aiocb.aio_fildes = ofd;
    if (aio_fsync(O_SYNC, &bufs[0].aiocb) < 0)
        err_sys("aio_fsync failed");
    exit(0);
}

这里使用了8个缓冲区，同时最多可以有8个异步I/O操作处于等待状态。使用off偏移量，可以实现多个异步I/O同时进程翻译文件的不同位置。

函数readv和writev

readv和writev用于一次函数调用中读、写多个非连续的缓冲区，这两个函数也称为散布读、聚集写。

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

ssize_t readv(int fd, const struct iovec *iov, int iovcnt);
ssize_t writev(int fd, const struct iovec *iov, int iovcnt);
								返回值：已读或已写的字节数；若出错，返回-1

第一个参数fd是文件描述符；

第二个参数iov是一个指向iovec结构数组的指针，第三个参数iovcnt是数组的大小（最大为IOV_MAX），iovec结构如下：

1
2
3
4
struct iovec {
    void 	*iov_base;	//缓冲的起始地址
    size_t	iov_len;	//缓冲的大小
}

下图是iovec结构的描述：

iovec结构组成

readv按上述顺序将读入的数据散布到各个缓冲区中，readv总是先填满一个缓冲区，在写入下个缓冲区。readv返回读的总字节数，如果是文件末尾，返回0。

writev按上述顺序从各个缓冲区中输出数据。writev返回输出的总字节数，通常为所谓缓冲区长度之和。

存储映射I/O

存储映射I/O将磁盘文件映射到一个缓冲区中，当从缓冲区中取数据，相当于从文件中读取相应字节数；当向缓冲区写数据，相应的字节会自动写入文件。这就可以不使用read和write的情况下I/O。

使用之前，要将给定的文件映射到一个存储区域中，该过程由mmap函数实现。

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

void *mmap(void *addr, size_t len, int prot, int flag, int fd, off_t off);
								返回值：若成功，返回映射区的起始地址；若出错，返回MAP_FAILED

参数addr指定映射区域地址，通常设为0，表示由系统分配映射区域；参数fd指映射的文件，在映射之前，必须打开该文件；参数len为映射的字节数；off为映射字节在文件中的偏移位置；prot参数为映射存储区的保护要求，如下表所示：

prot 说明
PROT_READ 映射区可读
PROT_WRITE 映射区可写
PROT_EXEC 映射区可执行
PROT_NONE 映射区不可访问

prot可设为上述参数的任意组合的按位或。对映射区的保护要求不能超过文件open模式访问权限。例如文件open只读打开，那么prot不能设为PROT_WRITE。

flag通常有3中参数：

  • MAP_FIXED：返回值必须等于addr。不建议使用该标志，这会降低可移植性，并且addr为非0，内核只是作为参考，并不保证会使用所要求的地址。addr使用0可获取最大可移植性。
  • MAP_SHARED：表示对映射区域的存储操作会修改映射文件，存储文件相当于向文件write操作
  • MAP_PRIVATE：映射区为映射文件的一个副本，对映射区的修改不会影响映射文件，只会修改文件的副本。

flag可能还有其他参数，但都是其他实现特有的。

函数mprotect可以更改一个现有映射的权限。

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

int mprotect(void *addr, size_t len, int prot);
								返回值：若成功，返回0；若出错，返回-1

注意，此处的addr必须是系统页长（linux一般为4096）的整数倍。prot与mmap中的相同。

如果mmap的flag参数设为MAP_SHARED，那么修改不会立即写回到文件，写回的时机由内核的守护进程决定。而且，就算只修改了一页中的一个字节，修改也会将整个页写回。

如果共享映射的页已修改，可以调用msync将该页冲洗到被映射的文件中。该函数与fsync相似，但仅作用于映射区，fsync冲洗整个文件。

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

int msync(void *addr, size_t len, int flags);
								返回值：若成功，返回0；若出错，返回-1

如果映射私有，则不修改映射的文件。与其他映射函数一样，addr必须是系统页长的整数倍。

flags有两个参数：

  • MS_AYNC：即简单的调试写的页，函数返回之前写操作不一定成功；
  • MS_SYNC：函数在写操作完成之后才返回。

进程终止时，会自动解除存储映射区的映射，也可以调用munmap解除映射区。注意，关闭映射区对应的文件描述符并不解除映射区。

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

int munmap(void *addr, size_t len);
								返回值：若成功，返回0；若出错，返回-1

调用munmap不会将映射区的内容写到磁盘文件上。解除映射区后，对MAP_PRIVATE存储区的修改会被丢弃。

赏

谢谢你请我吃糖果

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

很惭愧<br><br>只做了一点微小的工作<br>谢谢大家