设计模式-Observer模式

参考书籍：

《HeadFirst 设计模式》
《设计模式-可复用面向对象软件的基础》

本文主要介绍对象行为型模式——Observer（观察者）模式，介绍的内容基于HeadFirst设计模式这本书，由于这本书是通过java编写，学习C++的朋友可能有所疑惑，因此本文借助GoF将其例子通过C++进行改编。

简介

观察者模式定义了对象间一对多的依赖关系，当一个对象（本文称其为目标对象）的状态发生改变时，所有依赖它的对象（本文称其为观察者对象）都会得到通知并被自动更新。

more >>

展开全文 >>

APUE-第16章-网络IPC-套接字

套接字描述符

套接字是通信端点的抽象，和文件描述符类似，程序通过套接字描述符访问套接字。事实上，套接字在UNIX中就是一种文件描述符，许多处理文件描述符的函数（如read、write）也可以用于处理套接字。

socket函数创建一个套接字。

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

int socket(int domain, int type, int protocol);
                                返回值：若成功，返回文件描述符（套接字）；若出错，返回-1

参数domain（域）描述了通行的特性，有以下取值：

  • AF_INET：IPv4因特网域
  • AF_INET6：IPv6因特网域
  • AF_UNIX：UNIX域
  • AF_UPSPEC：未指定

参数type确定套接字类型，POSIX.1定义了以下套接字类型：

  • SOCK_DGRAM：固定长度、无连接的、不可靠的报文传递
  • SOCK_RAW：IP协议的数据报接口（POSIX.1中可选）
  • SOCK_SEQPACKET：固定长度的、有序的、可靠的、面向连接的报文传递
  • SOCK_STREAM：有序的、可靠的、双向的、面向连接的字节流

SOCK_STREAM提供字节流服务，应用程序分辨不出报文界限，从该套接字读取数据时，不会返回发送进程的所有字节数，需要多次函数调用才能获得所有数据。

SOCK_SEQPACKET提供基于报文服务，从该套接字获取的数据量与发送方一致。SCTP提供因特网域上的顺序数据包服务。

SOCK_RAW提供数据报接口，直接访问下层网络层（即IP层），应用程序自己负责构造协议头部。

参数protocol通常为0，表示给定的域和套接字类型选择默认协议。AF_INET中，type为SOCK_STREAM默认协议为TCP；AF_INET中，type为SOCK_DGRAM默认协议为UDP。

调用close可以关闭对套接字的访问，释放该套接字以重新使用。

函数shutdown可以禁止一个套接字的I/O

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

int shutdown(int sockfd, int how);
                                返回值：若成功，返回0；若出错，返回-1

参数how为SHUT_RD，表示关闭读端，无法从套接字读取数据；若为SHUT_WR，表示关闭写端，无法向套接字发送数据；若为SHUT_RDWR，表示即无法读取数据，也无法发送数据。

寻址

进程标识用于标志一个通行目标进程，由两部分组成：一部分为计算机的网络地址，标识了想要通信的计算机；另一部分为端口号，标识了特定的进程。

字节序

与不同计算机的进程通信，需要考虑字节序，字节序有大端法和小端法。Linux、FreeBSD、MAC OS为小端法，Solaris为大端法。

网络协议指定了字节序，因此计算机通信时不会被字节序混淆。TCP/IP协议为大端字节序。

对于TCP/IP应用程序，有4个函数用于主机字节序与网络字节序之间的转换。

1
2
3
4
5
6
7
8
9
10
#include <arpa/inet.h>

uint32_t htonl(uint32_t hostint32);
                                返回值：以网络字节序表示的32位整数
uint16_t htons(uint16_t hostint16);
                                返回值：以网络字节序表示的16位整数
uint32_t ntohl(uint32_t netint32);
                                返回值：以主机字节序表示的32位整数
uint16_t ntohs(uint16_t netint16);
                                返回值：以主机字节序表示的16位整数

h表示主机，n表示网络，l表示长（4字节）整数、s表示短（2字节）整数。

地址格式

为使不同地址能够传入套接字函数，地址会被强制转换成一个通用地址结构sockaddr：

1
2
3
4
5
struct sockaddr {
    sa_family_t sa_family;  //address family
    char        sa_data[];  //variable-length address
    ...
};

套接字实现可以自由添加额外成员。

在IPv4因特网域（AF_INET）中，套接字结构为sockaddr_in：

1
2
3
4
5
6
7
8
struct in_addr {
    in_addr_t   s_addr; //IPv4 address
};
struct sockaddr_in {
    sa_family_t     sin_family; //通信域
    in_port_t       sin_port;   //端口号
    struct in_addr  sin_addr;   //IPv4地址
};

IPv6因特网域（AF_INET6）套接字结构为sockaddr_in6：

1
2
3
4
5
6
7
8
9
10
struct in6_addr {
    uint8_t s6_addr[16];    //IPv6地址
};
struct sockaddr_in6 {
    sa_family_t     sin6_family;    //通信域
    in_port_t       sin6_port;      //端口号
    uint32_t        sin6_flowinfo;  //traffic class and flow info
    struct in6_addr sin6_addr;      //IPv6地址
    uint32_t        sin6_scope_id;  //set of interfaces for scope
};

尽管sockaddr_in和sockaddr_in6结构相差较大，但它们均被强制转换成sockaddr结构输入套接字程序中。

函数inet_ntop和inet_pton用于二进制地址格式和点分十进制表示（a.b.c.d）之间的转换，并且同时支持IPv4和IPv6。

1
2
3
4
5
6
7
8
9
#include <arpa/inet.h>

const char *inet_ntop(int domain, const void *restrict addr, 
                        char *restrict str, socklen_t size);
                                返回值：若成功，返回地址字符串指针；若出错，返回NULL

int inet_pton(int domain, const char *restrict str,
                void *restrict addr);
                                返回值：若成功，返回1；若格式无效，返回0；若出错，返回-1

inet_ntop将网络字节序的二进制地址转换成文本字符串格式，inet_pton将文本字符串格式转换成网络字节序的二进制地址。

参数domain只支持AF_INET和AF_INET6。

参数size指定了保存文本字符串的str的大小，INET_ADDRSTRLEN定义了足够存放IPv4的大小，INET6_ADDRSTRLEN定义了足够存放IPv6的大小。

展开全文 >>

APUE-第15章-进程间通信

管道

管道具有两个局限性：

  1. 历史上，管道是半双工的（即数据只能在一个方向上流动），目前有些系统实现全双工，但为了最大可移植性，应该假定系统支持半双工；
  2. 管道只能在具有公共祖先的两个进程之间使用。

管道通过调用pipe函数创建。

1
2
3
4
#include <unistd.h>

int pipe(int fd[2]);
								返回值：若成功，返回0；若出错，返回-1

参数fd返回两个文件描述符：fd[0]负责读、fd[1]负责写。fd[1]的输出是fd[0]的输入。对于支持全双工管道的系统，fd[0]和fd[1]以读/写方式打开。

一般来讲，使用管道通常进程会调用pipe，然后调用fork，从而创建从父进程到子进程的IPC通道，如下图所示：

fork后的半双工管道

之后如果想创建从父进程和子进程的管道，父进程可以关闭读端（fd[0]），子进程关闭写端（fd[2]）。如下图所示：

从父进程到子进程的管道

当管道一端被关闭时，会有以下规则：

  1. 如果读一个写端已关闭的管道，在所有数据被读完后，read返回0，表示文件结束；
  2. 如果写一个读端已关闭的管道，会产生SIGPIPE信号。如果忽略或捕捉该信号从处理程序返回后，write返回-1，errno设置为EPIPE。

实例1

下面创建了一个从父进程到子进程的管道，并父进程从管道中向子进程传送数据。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
#include "apue.h"

int main(void)
{
    int     n;
    int     fd[2];
    pid_t   pid;
    char    line[MAXLINE];

    if (pipe(fd) < 0)
        err_sys("pipe error");
    if ((pid = fork()) < 0) {
        err_sys("fork error");
    } else if (pid > 0) {       //父进程
        close(fd[0]);           //父进程关闭读端
        write(fd[1], "hello world\n", 12);	//父进程向管道写数据
    } else {                    //子进程
        close(fd[1]);           //子进程关闭写端
        n = read(fd[0], line, MAXLINE); //子进程从管道中读数据并放入line中
        write(STDOUT_FILENO, line, n);  //子进程向标准输出写入line
    }
    exit(0);
}

函数popen和pclose

标准I/O库提供了两个函数popen和pclose，这两个函数的操作是：创建一个管道，fork一个子进程，关闭未使用的管道端，执行shell命令，然后等待命令终止。

1
2
3
4
5
6
#include <stdio.h>

FILE *popen(const char *cmdstring, const char *type);
								返回值：若成功，返回文件指针；若出错，返回NULL
int pclose(FILE *fp);
								返回值：若成功，返回cmdstring终止状态；若出错，返回-1

函数popen先执行fork，然后调用exec执行cmdstring，并返回一个文件指针。若type是“r“，则文件指针连接到cmdstring的标准输出，表示进程可以从管道里读数据；若type是”w”，则文件指针连接到cmdstring的标准输入，表示进程可以向管道里写数据，如下所示：

执行fp=popen的“r”的结果 执行fp=popen的“w”的结果

函数pclose关闭标准I/O流，等待命令终止，然后返回shell的终止状态。

实例

popen和pclose的具体实现：

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

static pid_t *childpid = NULL;  //用于存放子进程的进程ID
static int maxfd;

FILE *popen(const char *cmdstring, const char *type)
{
    int     i;
    int     pfd[2];
    pid_t   pid;
    FILE    *fp;

    /* 参数type只能是"r"或者"w" */
    if ((type[0] != 'r' && type[0] != 'w') || type[1] != 0) {
        errno = EINVAL;
        return (NULL);
    }

    if (childpid == NULL) {
        maxfd = open_max();     //文件标识符上限大小
        if ((childpid = calloc(maxfd, sizeof(pid_t))) == NULL)  //开辟空间，用于存储popen创建的子进程ID
            return (NULL);
    }

    if (pipe(pfd) < 0)
        return (NULL);
    if (pfd[0] >= maxfd || pfd[1] >= maxfd) {   //如果pipe申请的文件描述符大于上限，表示文件描述符用尽，需出错返回
        close(pfd[0]);
        close(pfd[1]);
        errno = EMFILE;     //文件描述符用尽
        return (NULL);
    }

    if ((pid = fork()) < 0) {
        return (NULL);
    } else if (pid == 0) {  //下述动作都是子进程
        if (*type == 'r') {
            close(pfd[0]);  //关闭读端，因为type为r，则cmdstring负责标准输出，此时子进程负责读
            if (pfd[1] != STDOUT_FILENO) {
                dup2(pfd[1], STDOUT_FILENO);    //将管道写端定向至标准输出，子进程的标准输出写入管道
                close(pfd[1]);
            }
        } else {    //如果type为"w"
            close(pfd[1]);      //关闭写端
            if (pfd[0] != STDIN_FILENO) {
                dup2(pfd[0], STDIN_FILENO);     //将管道读端定向至标准输入，子进程从管道中读取数据到标准输入
                close(pfd[0]);
            }
        }   

        /* 子进程释放所有文件描述符 */
        for (i = 0; i < maxfd; i++)
            if (childpid[i] > 0)
                close(i);
  
        execl("/bin/sh", "sh", "-c", cmdstring, (char *)0); //子进程通过sh执行cmdstring命令
        _exit(127);             //如果上一行的execl执行失败，则进程直接退出
    }

    if (*type == 'r') {
        close(pfd[1]);      //父进程关闭写端
        if ((fp = fdopen(pfd[0], type)) == NULL)        //根据文件描述符，获取I/O流
            return (NULL);
    } else {
        close(pfd[0]);      //父进程关闭读端
        if ((fp = fdopen(pfd[1], type)) == NULL)        //获取I/O流
            return (NULL);
    }

    childpid[fileno(fp)] = pid;     //以fp对应的文件描述符为索引，存储相应的子进程ID
    return (fp);
}

int pclose(FILE *fp)
{
    int     fd, stat;
    pid_t   pid;

    if (childpid == NULL) {
        errno = EINVAL;
        return (-1);
    }

    fd = fileno(fp);
    if (fd >= maxfd) {      //所给的文件指针只是的文件描述符大于最大文件描述符，出错并退出
        errno = EINVAL;
        return (-1);
    }
    if ((pid = childpid[fd]) == 0) {    //所给的文件描述符没有被popen函数创建
        errno = EINVAL;
        return (-1);
    }

    childpid[fd] = 0;
    if (fclose(fp) == EOF)      //关闭文件描述符
        return (-1);
    
    while (waitpid(pid, &stat, 0) < 0)  //等待子进程结束
        if (errno != EINTR)	//如果子进程不存在，waitpid返回-1，errno会设置为ECHILD
            return (-1);
    
    return (stat);
}

协同进程

UNIX系统过滤程序从标准输入读取数据，向标准输出写数据，几个过滤程序在shell管道中线性连接。当一个过滤程序既产生某个过滤程序的输入，又读取该过滤程序的输出，则它就是协同进程。

popen只提供连接到另一个进程的标准输入或标准输出的一个单项管道，而协同进程有连接到另一个进程的两个单项管道：一个连接到标准输入、一个连接到标准输出。可以将数据写到标准输入，经过处理后，在从其标准输出读取数据。示意图如下图所示：

协同进程

实例

下面是一个简单的协同进程示例，协同进程从其标准输入读取两个数，计算它们的和，然后输出到标准输出。

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"

static void sig_pipe(int);

int main(void)
{
    int     n, fd1[2], fd2[2];
    pid_t   pid;
    char    line[MAXLINE];

    /* SIGPIPE：在管道的读进程终止时继续写进管道，产生此信号 */
    if (signal(SIGPIPE, sig_pipe) == SIG_ERR)   
        err_sys("signal error");
  
    if (pipe(fd1) < 0 || pipe(fd2) < 0)     //创建两个管道，用于连接父进程和子进程
        err_sys("pipe error");
  
    if ((pid = fork()) < 0) {
        err_sys("fork error");
    } else if (pid > 0) {       //父进程
        /* 父进程只需要保留fd1[1]、fd2[0] */
        close(fd1[0]);      
        close(fd2[1]);

        while (fgets(line, MAXLINE, stdin) != NULL) {       //父进程负责从标准输入读取数据，并从子进程中读取转换后的数据并输出到标准输出
            n = strlen(line);
            if (write(fd1[1], line, n) != n)      //向管道1写入数据
                err_sys("write error to pipe");
            if ((n = read(fd2[0], line, MAXLINE)) < 0)  //从管道2读取数据
                err_sys("read error from pipe");
            if (n == 0) {
                err_msg("child closed pipe");
                break;
            }
            line[n] = 0;
            if (fputs(line, stdout) == EOF)     //父进程将从管道读取的数据打印到标准输出
                err_sys("fputs error");
        }
        if (ferror(stdin))      //fget到达文件尾和出错都返回NULL，所有要调用ferror判断是否出错
            err_sys("fgets error on stdin");
        exit(0);
    } else {            //子进程
        close(fd1[1]);
        close(fd2[0]);
        if (fd1[0] != STDIN_FILENO) {   //将子进程的标准输入定向至管道fd1[0]
            if (dup2(fd1[0], STDIN_FILENO) != STDIN_FILENO)
                err_sys("dup2 error to stdin");
            close(fd1[0]);
        }
        if (fd2[1] != STDOUT_FILENO) {  //将子进程的标准输出定向至管道fd2[1]
            if (dup2(fd2[1], STDOUT_FILENO) != STDOUT_FILENO)
                err_sys("dup2 error to stdout");
            close(fd2[1]);
        }
        if (execl("./add2", "./add2", (char *)0) < 0)     //子进程执行add2程序
            err_sys("execl error");
    }
    exit(0);
}

static void sig_pipe(int signo)
{
    printf("SIGPIPE caught\n");
    exit(1);
}

FIFO

FIFO又称命名管道，匿名管道只能在只能用于有相同父进程的两个进程通信。而通过FIFO，不相关的进程也能交换数据。

FIFO是一种文件类型，通过stat结构的st_mode字段可以直到文件是否是FIFO。创建FIFO类似于创建文件：

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

int mkfifo(const char *path, mode_t mode);
int mkfifoat(int fd, const char *path, mode_t mode);
								返回值：若成功，返回0；若失败，返回-1

当用mkfifo或mkfifoat创建FIFO使，要用open打开它。正常的文件I/O函数都需要FIFO。

当open一个FIFO是，非阻塞标志（O_NONBLOCK）会有一下影响：

  • 在没有阻塞情况下（默认，没有指定O_NONBLOCK），只读open会阻塞到某个进程以写打开该FIFO为止。同理，只写open要阻塞到某个其他进程以读打开该FIFO为止；
  • 如果指定了O_NONBLOCK，只读open立即返回。若此时没有其他进程以读打开FIFO，则只读open返回-1，并将errno设置为ENXIO。

类似于管道，若write一个没有读进程的FIFO，则产生信号SIGPIPE。若FIFO的最后一个写进程关闭了FIFO，则为该FIFO产生一个文件结束标志。

一个给定的FIFO有多个写进程是常见的，如果要保证数据不交叉，则需要保证写操作的原子性。和管道一样，PIPE_BUF是原子地写入FIFO的最大数据量。

XSI IPC

有三种称为XSI IPC的IPC：消息队列、信号量、共享内存。

每个IPC结构关联一个ipc_perm结构，该结构定义了权限和所有者，至少包括：

1
2
3
4
5
6
7
struct ipc_perm {
    uid_t 	uid;		//拥有者的有效用户ID
    gid_t	gid;		//拥有者的有效组ID
    uid_t 	cuid;		//创建者的有效用户ID
    gid_t 	cgid;		//创建者的有效组ID
    mode_t	mode;		//访问模式
};

修改这些值，调用进程必须是IPC结构的创建者或者超级用户。

mode字段表示权限，任何IPC不存在执行权限，消息队列和共享内存使用术语“读”和“写”，信号量使用“读”和“更改”，下表是每种IPC的权限。

权限 位
用户读 0400
用户写（更改） 0200
组读 0040
组写（更改） 0020
其他读 0004
其他写（更改） 0002

消息队列

消息队列是消息的链接表，存储在内核中，有消息队列标识符标识。

msgget用于创建新队列或打开一个现有队列；msgsnd将消息添加到队列尾部；msgrcv用于从队列中取消息。消息并不一定要以先进先出的次序取，也可以按照消息的类型取消息。

每个队列都有一个msqid_ds结构与其关联：

1
2
3
4
5
6
7
8
9
10
11
struct msqid_ds {
    struct ipc_perm     msg_perm;       
    msgqnum_t           msg_qnum;       //队列中的消息数量
    msglen_t            msg_qbytes;     //队列中最大字节数
    pid_t               msg_lspid;      //最新的msgsnd的pid
    pid_t               msg_lrpid;      //最新的msgrcv的pid
    time_t              msg_stime;      //最新的msgsnd时间
    time_t              msg_rtime;      //最新的msgrcv时间
    time_t              msg_ctime;      //最新改变的时间
    ...
}

此结构定义了队列的当前状态。

调用的第一个函数通常是msgget，其功能是打开一个现有队列或创建一个新队列：

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

int msgget(key_t key, int flag);
                                返回值：若成功，返回消息队列ID；若出错，返回-1

参数key讨论了是创建一个新队列还是引用现有队列。在创建新队列时，需要初始化msqid_ds结构的下列成员：

  • ipc_perm按上述的XSI IPC所述舒适化，mode成员按flag设置相应权限位，权限XSI IPC所述；
  • msg_qnum、msg_lspid、msg_lrpid、msg_stime、msg_rtime都设置为0；
  • msg_ctime设置为当前时间；
  • msg_qbytes设置为系统限制值（如Linux是16384字节）。

若执行成功，msgget返回非负队列ID。该ID可用于后续的消息队列函数。

函数msgctl可以对队列执行多种操作。

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

int msgctl(int msqid, int cmd, struct msqid_ds *buf);
                                返回值：若成功，返回0；若出错，返回-1

参数cmd指定了msqid指定队列要执行的命令，有：

  • IPC_STAT： 取mspid所指队列的msqid_ds结构，并将其存放在buf执行的结构中
  • IPC_SET：将buf中的msg_perm.uid、msg_perm.gid、msg_perm.mode、msg_qbytes复制到msqid指定的msqid_ds结构中。此命令只能由两种进程执行：1）有效用户ID等于msg_perm.cuid或msg_perm.uid；2）拥有超级用户权限的进程。还有只有超级用户才能增加msg_qbytes的值；
  • IPC_RMID：从系统中删除该消息队列以及仍在消息队列中的所有数据。删除立即生效。仍在使用这一消息队列的进程在下一次试图操作该队列时，会得到EIDRM错误。执行此命令的进程只有两种，与上述的IPC_SET一致。
    这三条命令（IPC_STAT、IPC_SET、IPC_RMID）可以用于信号量和共享存储。

函数msgsnd将数据放进消息队列中。

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

int msgsnd(int mspid, const void *ptr, size_t nbytes, int flag);
                                返回值：若成功，返回0；若出错，返回-1

每个消息由3部分组成：正的长整型字段、非负的长度、实际数据字节数。并且消息总是放在队列尾端。

参数ptr指向一个长整型数，包括整型消息类型，紧接着是消息数据（若nbyte为0，则无消息数据）。若发送的最长消息为512字节，因此可以定义以下结构：

1
2
3
4
struct mymesg {
    long mtype;         //消息类型
    char mtext[512];    //消息数据
};

ptr就可以是指向mymesg结构的指针。接受者可以通过消息类型以非先进先出的次序取消息。

参数flag可以指定为IPC_NOWAIT。这类似于I/O中的非阻塞I/O标志，若消息已满，则指定了IPC_NOWAIT的msgsnd操作会立即出错并返回EAGAIN。如果没有指定IPC_NOWAIT，则进程会一直阻塞，直到1）有空间容纳消息；2）或系统删除该消息队列，返回EIDRM错误；3）或捕捉到一个信号，并从信号处理函数返回，返回EINTR错误。

当msgsnd成功返回时，消息队列的msqid_ds结构会更新，表明调用的进程ID（msg_lspid）、调用时间（msg_stime）、新增消息（msg_qnum）。

函数msgrcv从消息队列中取消息。

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

ssize_t msgrcv(int msqid, void *ptr, size_t nbytes, long type, int flag);
                            返回值：若成功，返回消息数据部分的长度；若出错，返回-1

参数ptr与msgsnd类似，包括消息类型和消息数据缓冲区；nbytes指定缓冲区长度，若返回消息的长度大于nbytes，如果flag设置MSG_NOERROR位，则将该消息截断，如果没有设置该位，则函数出错返回E2BIG（消息仍在队列中）。

参数type有三种情况：

  • type == 0：返回队列中的第一个消息；
  • type > 0 ：返回消息队列中第一个类型为type的消息；
  • type < 0 ：返回消息类型值小于等于type绝对值的消息，若符合的消息有多个，则返回类型值最小的消息。

参数flag可指定为IPC_NOWAIT，这样，若无指定类型的消息，则msgrcv返回-1，error设置为ENOMSG。如果没有设置IPC_NOWAIT，则进程阻塞，直到1）有指定类型消息可用、2）或系统删除此队列（返回-1，error设置为EIDRM）、3）或捕捉到一个信号并从信号处理函数返回（返回-1，error设置为EINTR）。

msgrcv成功执行时，内核更新该消息队列相关的msgid_ds结构，有指示调用者的进程ID（msg_lrpid）、调用时间（msg_rtime）、指示消息数减少1（msg_qnum）。

因为：

  1. 该IPC没有引用计数，当向队列中添加消息后直接终止，该消息队列不会被删除，直到进程调用msgrcv或msgctl或删除该消息队列。直到最后一个引用FIFO的进程结束，FIFO名字仍保留在系统中；

  2. 该IPC不使用文件描述符，因此无法使用I/O多路复用函数（select、poll），这使得很难一次使用一个以上的IPC结构。

目前为止，在速度上管道和FIFO相差无几，因此，在新的程序中不应该使用FIFO。

信号量

信号量与前面的IPC不同，它是一个计数器，用于为多个进程提供对共享数据对象的访问。

为了获得共享资源，进程需要：

  1. 判断控制该资源的信号量；
  2. 若信号量为正，则进程可以使用该资源，并将信号量值减1；
  3. 若信号量为0，则进程进入休眠状态，直到信号量大于0，进程被唤醒，执行步骤1.

常用的信号量为二元信号量，它控制单个资源，初始值为1。但是信号量初值可以是任意初值，表示控制了多少个共享资源单位。

内核为每个信号量集合维护一个semid_ds结构：

1
2
3
4
5
6
struct semid_ds {
    struct ipc_perm     sem_perm;
    unsigned short      sem_nsems;
    time_t              sem_otime;  //最后调用semop()的时间
    time_t              sem_ctime;  //最后改变的时间
};

每个信号量由一个无名结构表示，它至少包括：

1
2
3
4
5
6
struct {
    unsigned short  semval;     //semaphore value, always >= 0
    pid_t           sempid;     //pid for last operation
    unsigned short  semncnt;    //processes awaiting semval>=curval
    unsigned short  semzcnt;    //processes awaiting semval==0
};

要想使用信号量，首先需要调用semget函数获得一个信号量ID。

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

int semget(key_t key, int nsems, int flag);
                            返回值：若成功，返回信号量ID；若失败，返回-1

创建一个新集合时，要对semid_ds结构的下列成员赋值：

  • 初始化ipc_perm结构，与FIFO中的msqid_ds类似；
  • sem_otime设置为0
  • sem_ctime设置为当前时间
  • sem_nsems设置为nsems

nsems是该集合中的信号量数。

函数semctl包含了多个信号量操作

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

int semctl(int semid, int semnum, int cmd, .../* union semun arg */);
                            返回值：如下

第4个参数可选，若使用，则类型是semun，如下：

1
2
3
4
5
union semun {
    int                 val;        //for SETVAL
    struct semid_ds     *buf;       //for IPC_STAT and IPC_SET
    unsigned short      *array;     //for GETALL and SETALL
};

注意这里是union，不是指向union的指针。

参数cmd有以下命令：

  • IPC_STAT：取该集合的semid_ds结构，存储在arg.buf指向的结构中
  • IPC_SET：按arg.buf指向的结构的值，设置集合中的sem_perm.uid、sem_perm.gid、sem_perm.mode字段。
  • IPC_RMID：从系统中删除该信号量集合
  • GETVAL：返回semnum的semval值
  • SETVAL：设置semnum的semval值。该值大小由arg.val指定
  • GETPID：返回semnum的sempid值
  • GETNCNT：返回semnum的semncnt值
  • GETZCNT：返回semnum的semzcnt值
  • GETALL：取该集合中所有的信号量值。这些值存储在arg.array指向的数组中
  • SETALL：将集合中所有信号量的值设置为arg.array指向的数组中的值。

除GETALL以外的所有GET命令，semctl返回相应的值。其他命令，若成功，返回0；如出错，返回-1，并设置errno。

函数semop自动执行信号量集合上的操作数组。

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

int semop(int semid, struct sembuf semoparray[], size_t nopes);
                                返回值：若成功，返回0；若出错，返回-1

semoparray是一系列信号量操作的数组：

1
2
3
4
5
struct sembuf {
    unsigned short  sem_num;    //member in set
    short           sem_op;     //operation
    short           sem_flg;    //IPC_NOWAIT, SEM_UNDO
};

参数nops指定了该数组中操作的数量。

semop具有原子性，对数组中的操作，它或者执行所有操作，或者一个不做。

注意，对于多个进程间共享一个资源，对单一资源加锁，我们应该使用记录锁，因为它比信号量更简单、速度更快，并且系统会管理进程结束后遗留下的锁（对于信号量要指定SEM_UNDO标志）。

共享存储

共享存储允许多个进程共享给定的存储区，因为数据不需要拷贝，因此是最快的IPC。使用共享存储时，需要同步多个进程，例如在服务进程正在写数据，那么客户进程不应该读数据。通常，信号量用于同步共享存储的访问。

XSI共享存储与内存映射文件的区别是，前者没有相关的文件，且共享存储段是内存匿名段。

内核为每个共享存储段维护一个结构，该结构至少有：

1
2
3
4
5
6
7
8
9
10
struct shmid_ds {
    struct ipc_perm shm_perm;
    size_t          shm_segsz;      //size of segment in bytes
    pid_t           shm_lpid;       //pid of last shmop()
    pid_t           shm_cpid;       //pid of creator
    shmatt_t        shm_nattch;     //number of current attaches
    time_t          shm_atime;      //last-attach time
    time_t          shm_dtime;      //last-detach time
    time_t          shm_ctime;      //last-change time
};

shmget通常是第一个调用的函数，它获得一个共享存储标识符

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

int shmget(key_t key, size_t size, int flag);
                                返回值：若成功，返回共享存储ID；若出错，返回-1

key用于表示是创建一个新共享存储段，还是引用一个现有的共享存储段。当创建一个新段时，需要初始化shmid_ds结构的以下成员（和消息队列、信号量类似）：

  • ipc_perm按之前XSI IPC描述的方式初始化
  • shm_lpid、shm_nattach、shm_atime、shm_dtime都设置为0；
  • shm_ctime设置为当前时间；
  • shm_segsz设置为请求的size。

参数size是共享存储段的长度，以字节为单位。通常为向上取整的系统页长（Linux是4096字节）的整数倍。当创建段时，指定size大小，当引用段时，size为0。

和消息队列、信号量类似，shmctl函数对共享存储段执行多种操作。

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

int shmctl(int shmid, int cmd, struct shmid_ds *buf);
                                返回值：若成功，返回0；若失败，返回-1

cmd参数指定5中命令：

  • IPC_STAT：取段的shmid_ds结构，存储在buf指向的结构中
  • IPC_SET：由buf参数指向的结构设置段的shmid_ds结构中的参数：shm_perm.uid、shm_perm.gid、shm_perm.mode。
  • IPC_RMID：从系统中删除该共享存储段。
    Linux和Solaris提供了额外两个命令，它们不是Single UNIX Specification的组成部分。
  • SHM_LOCK：在内存中对该段加锁，此命令只能由超级用户执行。
  • SHM_UNLOCK：解锁共享存储段，只能由超级用户执行。

创建了一个共享存储段后，进程可以通过函数shmat将其映射到它的地址空间中。

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

void *shmat(int shmid, const void *addr, int flag);
                                返回值：若成功，返回指向共享存储段的指针；若出错，返回-1

对于参数addr，表示共享存储映射到进程的地址，除非只计划在一个硬件上允许，否则不应该设置该值，应当指定addr为0，由内核决定地址。

参数flag如果为SHM_RDONLY，表示只读该共享存储段，否则为读写方式连接此段。

函数shmdt可以将进程与该段分离，注意，此时并没有删除其标识符和相关数据结构。标识符会一直存在，直到有进程使用IPC_RMID的调用shmctl函数删除它为止。

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

int shmdt(const void *addr);
                                返回值：若成功，返回0；若出错，返回-1

实例

下面的程序是测试存储区各个段（bss段、堆段、栈段、data段）和共享存储段的空间分布位置。

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

#define ARRAY_SIZE      4000
#define MALLOC_SIZE     100000
#define SHM_SIZE        100000
#define SHM_MODE        0600    //用户读/写

char array[ARRAY_SIZE];     //未初始化的值，bss段

int main(void)
{
    int     shmid;
    char    *ptr, *shmptr;

    printf("array[] form %p to %p\n", (void *)&array[0], (void *)&array[ARRAY_SIZE]);       //打印bss段的位置分布
    printf("stack around %p\n", (void *)&shmid); //打印栈段的存储位置

    if ((ptr = malloc(MALLOC_SIZE)) == NULL)
        err_sys("malloc error");
    printf("malloced from %p to %p\n", (void *)ptr, (void *)ptr+MALLOC_SIZE); //打印堆段的存储位置

    if ((shmid = shmget(IPC_PRIVATE, SHM_SIZE, SHM_MODE)) < 0)  //创建共享存储区，长度为100000，用户可读可写
        err_sys("shmget error");
    if ((shmptr = shmat(shmid, 0, 0)) == (void *) -1)   //将共享存储区映射至本进程
        err_sys("shmat error");
    printf("shared memory attached from %p to %p\n", (void *)shmptr, (void *)shmptr+SHM_SIZE);  //打印内核分配的共享存储段的位置

    if (shmctl(shmid, IPC_RMID, 0) < 0)     //删除共享存储段
        err_sys("shmctl error");

    exit(0);
}

在我的Linux中，其输出如下：

array[] form 0x555e5adf8060 to 0x555e5adf9000
stack around 0x7fff0cbb73a4
malloced from 0x555e5ca946b0 to 0x555e5caacd50
shared memory attached from 0x7f150fdd7000 to 0x7f150fdef6a0

可以看到，它与典型存储区分布类似：

Linux系统的存储区分布

POSIX信号量

POSIX信号量相较于XSI信号量有了优化，解决了XSI信号量的缺点：

  • POSIX信号量性能更高；
  • POSIX信号量使用更简单，没有信号量集机制；
  • POSIX在删除时表现更好。当XSI信号量被删除时，使用信号量标识符的操作会失败，并设置errno为EIDRM，而使用POSIX信号量，在信号量标识符被删除时，操作不会失败并且正常工作直到该信号量的最后一次引用被释放。

POSIX信号量有两种形式：命名的和未命名的。它们的差异在创建和销毁上，其他工作一样。

未命名信号量只存在内存中，要求使用信号量的进程必须可以访问该内存（也就是信号量所在内存位置），因此它只用于：（1）同一进程的线程；（2）不同进程将信号量所在内存映射到各自空间中的线程。

命令信号量可以通过名字访问，可以被任何一直它名字的进程使用。

函数sem_open可以创建一个新的命名信号量或使用一个现有信号量。

1
2
3
4
#include <semaphore.h>

sem_t *sem_open(const char *name, int oflag, ... /* mode_t mode, unsigned int value */);
                                返回值：若成功，返回指向信号量的指针；若出错，返回SEM_FAILED

参数name是信号量的名字；

参数oflag指定函数动作标志，oflag如果是O_CREAT，则表示创建信号量，若信号量存在，则无额外的初始化发生，并且函数不会出错；若确保要创建信号量，则oflag设置诶O_CREAT|O_EXCL，此时如果信号量已存在，sem_open会调用失败。

另外两个参数用于创建信号量，mode指示信号量权限，value指示信号量的初始值。

函数sem_close用于释放任何与信号量相关的资源。

1
2
3
4
#include <semaphore.h>

int sem_close(sem_t *sem);
                                返回值：若成功，返回0；若出错，返回-1

如果进程没有调用sem_close后退出，则内核会自动关闭任何打开的信号量。

函数sem_unlink用于销毁一个命名信号量

1
2
3
4
#include <semaphore.h>

int sem_unlink(const char *name);
                                返回值：若成功，返回0；若出错，返回-1

如果name指示的信号量没有被引用，则该信号量被销毁；若有引用，则销毁会延迟到最后一个引用关闭。

函数sem_wait或sem_trywait用于信号量减1操作。

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

int sem_trywait(sem_t *sem);
int sem_wait(sem_t *sem);
                                返回值：若成功，返回0；若失败，返回-1

调用sem_wait函数在信号量为0时，进程会进入阻塞状态，而调用sem_trywait函数在信号量为0时，则直接出错返回-1，并将errno置为EAGAIN。

函数sem_timewait可以设定阻塞时间

1
2
3
4
5
#include <semaphore.h>
#include <time.h>

int sem_timedwait(sem_t *restrict sem, const struct timespec *restrict tsptr);
                                返回值：若成功，返回0；若出错，返回-1

如果超时信号量没能减1，则返回-1，并将errno设置为ETIMEOUT。

函数sem_post可使信号量加1，若调用sem_post时，有进程因为sem_wait阻塞，则进程被唤醒，并信号量被sem_wait减1。

1
2
3
4
#include <semaphore.h>

int sem_post(sem_t *sem);
                                返回值：若成功，返回0；若出错，返回-1

若只想在单个进程中使用POSIX信号量，使用未命名信号量更容易。函数sem_init创建一个未命名信号量。

1
2
3
4
#include <semaphore.h>

int sem_init(sem_t *sem, int pshared, unsigned int value);
                                返回值：若成功，返回0；若出错，返回-1

pshared参数表示是否要在多个进程中使用信号量，如果是，则设置其为非0。value指定信号量的初始值。

sem参数则是一个声明的sem_t类型变量的地址，而不需要通过sem_open。

未命名信号量使用完成后，调用sem_destroy丢弃它。

1
2
3
4
#include <semaphore.h>

int sem_destroy(sem_t* sem);
                                返回值：若成功，返回0；若出错，返回-1

函数sem_getvalue可以检索信号量值。

1
2
3
4
#include <semaphore.h>

int sem_getvalue(sem_t *restrict sem, int *restrict valp);
                                返回值：若成功，返回0；若出错，返回-1

成功后，valp指向的整数就是信号量值。注意，在获取到信号量值后，该值有可能已经改变，除非使用额外的同步机制避免竞争，否则该函数只适用于调试。

实例

下面通过信号量来实现互斥锁，该锁能被一个线程加锁而被另外一个线程解锁，它的结构可以是：

1
2
3
4
struct slock {
    sem_t   *semp;
    char    name[_POSIX_NAME_MAX];
};

下面是通过信号量实现互斥原语。

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
/* 头文件 */
#include <semaphore.h>
#include <fcntl.h>
#include <limits.h>
#include <sys/stat.h>

struct slock {
    sem_t   *semp;                  //用于sem类型的函数
    char    name[_POSIX_NAME_MAX];  //信号量的名字
};

struct slock *s_alloc();
void s_free(struct slock *);        //释放互斥锁的资源
int s_lock(struct slock *);         //获得锁
int s_trylock(struct slock *);      //尝试获得锁
int s_unlock(struct slock *);       //解锁

/* 源文件 */
#include "slock.h"
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>

struct slock * s_alloc()
{
    struct slock *sp;
    static int cnt;

    if ((sp = malloc(sizeof(struct slock))) == NULL)
        return (NULL);
    do {
        /* 根据进程号来创建信号量的名字 */
        snprintf(sp->name, sizeof(sp->name), "/%ld.%d", (long)getpid(), cnt++);
        /* 此时如果有两个进程同时调用s_alloc并以同名字创建互斥量，
         * 则其中一个线程会成功，另一个线程会因为O_EXCL出错返回并将errno设置为EEXIST。
         */
        sp->semp = sem_open(sp->name, O_CREAT|O_EXCL, S_IRWXU, 1);
    } while ((sp->name == SEM_FAILED) && (errno == EEXIST));
    if (sp->name == SEM_FAILED) {
        free(sp);
        return (NULL);
    }
    /* 在创建信号量后，销毁其名字，这阻止了其他进程访问该信号量
     * 如果不销户名字，其他进程可以通过该命名调用sem_open访问该信号量，这并不是我们想要的结果
     */
    sem_unlink(sp->name);   
    return (sp);
}

void s_free(struct slock *sp)
{
    sem_close(sp->semp);
    free(sp);
}

int s_lock(struct slock *sp)
{
    return (sem_wait(sp->semp));
}

int s_trylock(struct slock *sp)
{
    return (sem_trywait(sp->semp));
}

int s_unlock(struct slock *sp)
{
    return (sem_post(sp->semp));
}

小结

这一章介绍了进程间通信的方式：管道、命名管道（FIFO）、通常称为XSI IPC的3种形式的IPC（消息队列、信号量、共享存储）、POSIX提供的替代XSI IPC信号量的机制。

APUE上给出的建议是：学会使用管道和FIFO，这两种IPC技术可以有效应用于大量程序。在新程序中，避免使用消息队列及信号量，应当考虑全双工管道和记录锁代替，因为它们使用更简单。共享存储仍有它的用途，虽然mmap（存储映射I/O的函数）有同样的效果。

展开全文 >>

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存储区的修改会被丢弃。

展开全文 >>

APUE-第13章-守护进程

守护进程的特征

什么是守护进程？

守护进程通常在系统引导时启动，系统关闭时终止。它们不控制终端，在后台运行。

在我的Linux系统中输入：ps -axj，输出了以下参数：

Linux下的ps输出

其中，PPID—父进程ID、PID—进程ID、PGID—进程组ID、SID—会话ID、UID—用户ID、TTY—终端名称、COMMAND—命令字符串。

父进程ID为0的进程是内核进程，内核进程存在于系统的整个生命期中，以超级用户特权运行，无命令行。

内核守护进程名字用[]中，Linux通过名为kthreadd的内核进程来创建其他内核进程，因此该进程是其他内核进程的父进程。在这些进程中，如：

  • kswapd：称为内存换页守护进程，支持虚拟子系统将脏页面写回磁盘来回收页面。
  • flush：称为冲洗守护进程。在可用内存达到设置的最小阈值时，将脏页面冲洗到磁盘中。他也会定期清洗脏页面，从而减少系统故障时的数据丢失。
  • sync_supers：定期将文件系统元数据冲洗至磁盘。

大多数守护进程都以超级用户root特权运行（即UID是0），所有守护进程没有控制终端，TTY都是？。内核守护进程以无控制终端方式启动，用户层守护进程不控制终端可能是调用了setsid的结果。用户层守护进程的父进程都是init进程。

创建守护进程

创建守护进程的编程规则：

  1. 调用umask将文件模式创建屏蔽字设置为已知值（一般为0）；
  2. 调用fork，父进程exit。保证进程不是一个进程组组长，这是setsid调用的先决条件；
  3. 调用setsid创建新会话，使调用进程：（1）成为会话首进程，（2）成为新进程组的组长，（3）不控制终端；
  4. 将当前工作页面更改为根目录；
  5. 关闭不需要的文件描述符；
  6. 打开文件描述符0、1、2，使任何读标准输入、标准输出、标准错误的库例程不产生效果。

下面是一个守护进程的初始化函数：

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
#include "apue.h"
#include <syslog.h>
#include <fcntl.h>
#include <sys/resource.h>

void
daemonize(const char *cmd)
{
    int                 i, fd0, fd1, fd2;
    pid_t               pid;
    struct rlimit       r1;
    struct sigaction    sa;
	
    //设置屏蔽字
    umask(0);
    
    //获取最大的文件描述符
    if (getrlimit(RLIMIT_NOFILE, &r1) < 0)
        err_quit("%s: can't get file limit", cmd);
    
	//子进程通过调用setsid创建新会话
    if ((pid = fork()) < 0)
        err_quit("%s: can't fork", cmd);
    else if (pid != 0)
        exit(0);
    setsid();

    /*
     * Ensure future opens won't allocate controlling TTYs.
     */
    sa.sa_handler = SIG_IGN;    //忽略信号
    sigemptyset(&sa.sa_mask);	//清楚所有信号
    sa.sa_flags = 0;
    if (sigaction(SIGHUP, &sa, NULL) < 0)
        err_quit("%s: can't ignore SIGHUP", cmd);
    if ((pid = fork()) < 0)		//这个fork的进程才是守护进程
        err_quit("%s: can't fork", cmd);
    else if (pid != 0)
        exit(0);
    
    //修改当前工作目录为“/”
    if (chdir("/") < 0)
        err_quit("%s: can't change directory to /", cmd);
    
    //释放所有文件描述符
    if (r1.rlim_max == RLIM_INFINITY)
        r1.rlim_max = 1024;
    for (i = 0; i < r1.rlim_max; i++)
        close(i);
    
    //获取文件描述符0，1，2，防止库例程试图读取标准输入、标准输出、标准错误，因为守护例程不会和终端交互
    fd0 = open("/dev/null", O_RDWR);
    fd1 = dup(0);
    fd2 = dup(0);
	
    //打开日志消息
    openlog(cmd, LOG_CONS, LOG_DAEMON);
    if (fd0 != 0 || fd1 != 1 || fd2 != 2) {
        syslog(LOG_ERR, "unexpected file descriptors %d %d %d", 
            fd0, fd1, fd2);
        exit(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
45
46
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <stdio.h>
#include <sys/stat.h>
#include <stdlib.h>
#include <fcntl.h>
#include <syslog.h>

#define LOCKFILE "/var/run/daemon.pid"
#define LOCKMODE (S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH)

//记录锁
extern int lockfile(int);

int
already_running(void)
{
    int     fd;
    char    buf[16];
	
    //以读写的方式打开LOCKFILE文件，没有则创建
    fd = open(LOCKFILE, O_RDWR|O_CREAT, LOCKMODE);
    if (fd < 0) {
        //打开失败
        syslog(LOG_ERR, "can't open %s: %s", LOCKFILE, strerror(errno));
        exit(1);
    }
    //尝试使用记录锁锁住文件
    if (lockfile(fd) < 0) {
        if (errno == EACCES || errno == EAGAIN) {
            close(fd);
            return (1);
        }
        syslog(LOG_ERR, "can't lock %s: %s", LOCKFILE, strerror(errno));
        exit(1);
    }
    /* 截断文件，防止之前的进程ID比当前进程ID长
     * 如之前的ID为12345，当前ID为9999，如果不截断，则文件留下的是99995
     */
    ftruncate(fd, 0);
    //向文件写入当前进程的ID
    sprintf(buf, "%ld", (long)getpid());
    write(fd, buf, strlen(buf) + 1);
    return (0);
}

展开全文 >>

APUE-第12章-线程控制

引言

本章主要介绍线程属性、同步原语属性、同一进程中的多个线程之间如何保持数据私有性、进程如何与线程进行交互。

线程限制

下面是线程相关的一些限制：

限制名称 描述 name参数
PTHREAD_DESTRUCTOR_ITERATIONS 线程退出时尝试销毁线程特定数据的最大次数 _SC_THREAD_DESTRUCTOR_ITERATIONS
PTHREAD_KEYS_MAX 进程可以创建的键的最大数目 _SC_THREAD_KEYS_MAX
PTHREAD_STACK_MIN 一个线程栈可用的最小字节数 _SC_THREAD_STACK_MIN
PTHREAD_THREADS_MAX 进程可以创建的最大线程数 _SC_THREAD_THREADS_MAX

下面描述了4种操作系统实现中线程限制的值，其中“没有确定限制”并不意味着值是无限的：

限制名称 FreeBSD 8.0 Linux 3.2.0 Mac OS X 10.6.8 Solaris 10
PTHREAD_DESTRUCTOR_ITERATIONS 4 4 4 没有确定限制
PTHREAD_KEYS_MAX 256 1024 512 没有确定限制
PTHREAD_STACK_MIN 2048 16384 8192 8192
PTHREAD_THREADS_MAX 没有确定限制 没有确定限制 没有确定限制 没有确定限制

线程属性

pthread接口允许我们通过关联的不同属性来细调线程和同步对象的行为。管理这些属性的行为有：

  1. 每个对象与它自己类型的属性对象进行关联（比如线程与线程属性关联，互斥量和互斥量属性关联），一个属性对象可以代表多个属性；
  2. 有一个初始化函数，把属性设置为默认值；
  3. 有一个销毁属性对象函数，销毁初始化函数分配的资源；
  4. 每个属性都有一个从属性对象中获取属性值的函数；
  5. 每个属性都有一个设置属性值的函数，属性值作为参数按值传递。

在pthread_create函数中，有一个参数是pthread_attr_t，它可以修改线程默认属性。可以使用pthread_attr_init初始化pthread_attr_t结构。在调用pthread_attr_init后，pthread_attr_t结构所包含的就是操作系统实现支持的所有线程属性的默认值。

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

int pthread_attr_init(pthread_attr_t *attr);
int pthread_attr_destroy(pthread_attr_t *attr);
								两个函数返回值：若成功，返回0；否则，返回错误编号

pthread_attr_init初始化的属性对象是动态分配的，所以需要pthread_attr_destroy来释放这些内存空间。

分离线程：如果在创建线程时就知道不需要了解线程的终止状态，就可以修改pthread_attr_t 结构中detachstate线程属性，让线程一开始就处于分离状态。detachstate具有两个合法值：PTHREAD_CREATE_DETACHED——以分离状态启动线程、PTHREAD_CREATE_JOINABLE——正常启动线程，应用程序可以获取线程的终止状态。

1
2
3
4
5
6
7
#include <pthread.h>

/* 获取当前的 detachstate 线程属性 */
int pthread_attr_getdetachstate(const pthread_attr_t *restrict attr,
                                int *detachstate);
int pthread_attr_setdetachstate(pthread_attr_t *attr, int *detachstate);
								两个函数返回值：若成功，返回0；否则，返回错误编号

线程栈，即为线程分配的栈。可以使用pthread_attr_getstack和pthread_attr_setstack对线程栈属性进行管理。

1
2
3
4
5
6
7
8
#include <pthread.h>

int pthread_attr_getstack(const pthread_attr_t *restrict attr,
                          void **restrict stackaddr,
                          size_t *restrict stacksize);
int pthread_attr_setstack(pthread_attr_t *attr,
					      void *stackaddr, size_t stacksize);
								两个函数返回值：若成功，返回0；否则，返回错误编号

对于线程，虚地址空间的大小是固定的。但对于线程，同样大小的虚地址空间必须被所有的线程共享。如果使用许多线程，则这些线程栈累计大小就超过了可用的虚地址空间，就需要减少默认的线程栈大小。如果线程的函数分配了大量的自动变量，或调用函数设计很深的栈，则需要的栈比默认的大。

如果线程栈的虚地址空间消耗完了，则需要使用malloc或mmap来为可替代栈跟配空间，并用pthread_attr_setstack函数来改变新建线程的栈位置。stackattr参数指向线程栈的最低可寻址地址，该地址与边界地址对齐。当然，stackattr不一定是站的开始地址，如果一个处理器栈从高地址向低地址增长，那么stackaddr是线程栈的结尾位置。

应用程序可以通过pthread_attr_getstacksize和pthread_attr_setstacksize读取或设置线程属性stacksize。

1
2
3
4
5
6
#include <pthread.h>

int pthread_attr_getstacksize(const pthread_attr_t *restrict attr,
                              size_t *restrict stacksize);
int pthread_attr_setstacksize(pthread_attr_t *attr, size_t stacksize);
								两个函数返回值：若成功，返回0；否则，返回错误编号

其中，设置stacksize时其大小不能小于PTHREAD_STACK_MIN。

线程属性guardsize控制线程末尾用以避免栈溢出的扩展内存大小，默认值由具体实现决定，一般为系统页大小。将guardsize设置为0，则不会提供警戒缓冲区。如果程序修改了线程属性stackaddr，则系统认为由我们自己管理栈，栈警戒缓冲区机制无效，等同于将guardsize设置为0。

1
2
3
4
5
6
#include <pthread.h>

int pthread_attr_getguardsize(const pthread_attr_t *restrict attr,
                              size_t *restrict guardsize);
int pthread_attr_setguartdsize(pthread_attr_t *attr, size_t guardsize);
								两个函数返回值：若成功，返回0；否则，返回错误编号

如果guardsize被修改，操作系统可能把它取为页大小的整数倍。如果线程的栈指针溢出至警戒区域，应用程序就可能通过信号接收到出错信息。

more >>

展开全文 >>

使用Markdown中我遇到的问题

如何编写折叠代码块

<details>
<summary>标题1</summary>
<pre><code>代码内容
</code></pre>
</details>

效果如下：

>标题1 
代码内容

这里会有一个问题，当编写C程序，C程序中的头文件<>无法显示。
<details>
<summary>标题2</summary>
<pre><code>#include <stdio.h>
</code></pre>
</details>
问题的效果如下，可以看到<>中的内容消失：

>标题2 
#include

此时可以将代码块中的<替换为&lt、>替换为&gt，如下：
<details>
<summary>&gt标题3</summary>
<pre><code>#include &ltstdio.h&gt
</code></pre>
</details>

效果如下：

>标题3 
#include <stdio.h>

展开全文 >>

APUE_第11章_线程

线程概念

在程序设计时把进程设计成某个时刻，每个线程能够处理各自独立的任务。这有很多好处：

  • 为每种事件类型分配单独的处理线程，可以简化处理异步事件的代码；
  • 多个进程必须使用操作系统提供的复杂机制才能实现内存和文件描述符的共享，而多个线程自动地可以访问相同的存储地址空间和文件描述符；
  • 分解问题从而提高整个程序的吞吐量。若是单线程进程要完成多个任务，需要把任务串行化；若是进程控制多个线程，相互独立的任务处理可以交叉进行，只需为每个任务分配一个单独的线程；
  • 交叉程序同样可以通过多线程改善响应时间，多线程可以把程序中处理用户输入输出部分和其他部分分开

即使运行在单处理上，程序也可以通过多线程进行简化。而且，即使多线程程序在串行化任务时阻塞，由于某些线程在阻塞时还有其他线程可以运行，所以多线程程序在单处理上运行还是可以改善响应时间和吞吐量的。

每个线程都包含有表示执行环境所必须的信息，其中包括：线程ID、一组寄存器值、栈、调度优先级、策略、信号屏蔽字、errno变量（每个线程拥有属于自己的局部errno，以免一个线程干扰另一个线程）、线程私有数据。一个进程的所有信息对该进程的所有线程共享，包括可执行程序的代码、程序的全局内存、堆内存、栈、文件描述符。

线程标识

每个线程都有各自的线程ID。进程ID是整个系统中唯一的，线程ID是它所属的进程下上文中才有意义。

线程ID是pthread_t数据类型表示，所有可移植操作系统不能把它作为整数处理，pthread_equal函数是用于对两个线程ID进行比较。

1
2
3
4
#include <phtread.h>

int pthread_equal(pthread_t tid1, pthread_t tid2);
								返回值：若相等，返回非0数值；否则，返回0

Linux3.2.0 使用无符号长整型（unsigned long int）表示pthread_t ；

Solaris 10 使用无符号整型（unsigned int）表示pthread_t；

FreeBSD 8.0和Mac OS X 10.6.8用一个指向pthread结构的指针表示pthread_t。

线程可以通过pthread_self函数获取自身线程ID。

1
2
3
4
#include <pthread.h>

pthread_t pthread_self(void);
								返回值：调用线程的线程ID

当线程需要识别以线程ID为标识的数据结构时，pthread_self函数和pthread_equal可以一起使用。如：

下图为主线程控制工作队列实例。可以看到，主线程可以将新作业放进工作队列中，另外3个线程组成的线程池从队列中移出作业，当然线程不能任意从队列顶端取出作业，而是由主线程控制作业分配，主线程会在每个待处理作业的结构中标志处理该作业的线程ID，每个工作线程只能移除有自己线程ID的作业。

工作队列实例

more >>

展开全文 >>

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>谢谢大家