命名管道 fifo 的特点

特点描述

1. 打破了匿名管道只能在父子进程间使用的限制，可以在无亲缘关系的进程之间使用同一个 fifo
2. 未指定 NONBLOCK 参数 open fifo 时可能会 block，不当的编码可能会导致进程死锁
3. 当请求读超过 fifo 中现有数据大小的数据时，只会返回现有数据大小的内容
4. wirte 数据的原子性由写入的大小与 PIPE_BUF 的关系确定与是否指定 O_NONBLOCK 标志无关，当 wirte 的大小小于或等于 PIPE_BUF 时，write 操作被保证是原子执行，不会产生数据交错，当大小超过 PIPE_BUF 时 write 操作不保证能原子执行
5. 向一个未以读模式打开的 FIFO 写入数据时，内核会向此进程发送 SIGPIPE 信号

mkfifo 函数调用的系统调用

linux 中 mkfifo 函数并不是一个系统调用而是一个 C 库函数，mkfifo 函数最终会调用 mknode 系统调用创建 fifo。

未指定 NONBLOCK 参数 open fifo 时可能会 block

以只读方式打开一个 fifo

打开一个 fifo 来读时，内核会判断此 fifo 上是否有写者，如果没有则调用 wait_for_partner 函数挂起当前进程等待对端就绪。

相关代码如下：

switch (filp->f_mode & (FMODE_READ | FMODE_WRITE)) {case FMODE_READ:/**  O_RDONLY*  POSIX.1 says that O_NONBLOCK means return with the FIFO*  opened, even when there is no process writing the FIFO.*/pipe->r_counter++;if (pipe->readers++ == 0)wake_up_partner(pipe);if (!is_pipe && !pipe->writers) {if ((filp->f_flags & O_NONBLOCK)) {/* suppress EPOLLHUP until we have* seen a writer */filp->f_version = pipe->w_counter;} else {if (wait_for_partner(pipe, &pipe->w_counter))goto err_rd;}}break;

上述代码首先递增 pipe 内部计数器 r_counter，当 readers 计数递增前值为 0 时调用 wake_up_partner 尝试唤醒阻塞在以 WRITE 模式调用 open 系统调用阻塞的进程，这些进程以 pipe 结构的 r_counter 变量为参数等待读者上线，r_counter 变量的值有更新表明成功，wait_for_partner 函数返回 0，否则继续等待。

wait_for_partner 函数代码如下：

static int wait_for_partner(struct pipe_inode_info *pipe, unsigned int *cnt)                                                                     
{DEFINE_WAIT(rdwait);int cur = *cnt;while (cur == *cnt) {prepare_to_wait(&pipe->rd_wait, &rdwait, TASK_INTERRUPTIBLE);pipe_unlock(pipe);schedule();finish_wait(&pipe->rd_wait, &rdwait);pipe_lock(pipe);if (signal_pending(current))break;}    return cur == *cnt ? -ERESTARTSYS : 0; 
}

while 循环终止的条件为计数值发生变化，这与上文的描述一致。

以只写方式打开一个 fifo

同理，以只写方式打开一个 fifo 时，内核会判断此 fifo 上是否有读者，如果没有则调用 wait_for_partner 函数挂起，直到读者调用 wake_up_partner 唤醒。

写操作 open 的实现代码如下：

case FMODE_WRITE:/**  O_WRONLY*  POSIX.1 says that O_NONBLOCK means return -1 with*  errno=ENXIO when there is no process reading the FIFO.*/ret = -ENXIO;if (!is_pipe && (filp->f_flags & O_NONBLOCK) && !pipe->readers)goto err;pipe->w_counter++;if (!pipe->writers++)wake_up_partner(pipe);if (!is_pipe && !pipe->readers) {if (wait_for_partner(pipe, &pipe->r_counter))goto err_wr;}break;

上述代码首先递增 pipe 内部计数器 w_counter，当 writers 计数递增前值为 0 时调用 wake_up_partner 尝试唤醒阻塞在以 READ 模式调用 open 系统调用阻塞的进程，这些进程以 pipe 结构的 w**_counter** 变量为参数等待写者上线，w_counter 变量的值有更新表明成功，wait_for_partner 函数返回 0，否则继续等待。

以读写模式打开 fifo

POSIX 标准并未定义以读写模式且设置 O_NONBLOCK 标志时的行为，linux 内核在这种情况下不会阻塞，会直接返回。

相关代码如下：

case FMODE_READ | FMODE_WRITE:/**  O_RDWR*  POSIX.1 leaves this case "undefined" when O_NONBLOCK is set.*  This implementation will NEVER block on a O_RDWR open, since*  the process can at least talk to itself.*/pipe->readers++;pipe->writers++;pipe->r_counter++;pipe->w_counter++;if (pipe->readers == 1 || pipe->writers == 1)wake_up_partner(pipe);break;

此处代码同时递增读写相关计数器并尝试唤醒阻塞在以读、写 open fifo 的进程。

修改 pipe 共享数据时的锁保护

上述操作均在获取了 pipe 互斥锁的条件下进行以保证共享数据的一致性。获取锁的接口为 __pipe_lock，其实现如下：

static inline void __pipe_lock(struct pipe_inode_info *pipe)
{mutex_lock_nested(&pipe->mutex, I_MUTEX_PARENT);
}

从名称上看它是一个互斥锁，保证同一时刻只有一个用户占有，同时它支持嵌套调用，即如果一个已经获取了这把锁的进程再次获取时也能够成功，而释放时也需要释放相同次数。

为什么描述的是 fifo，内核代码中却用的是 pipe?

linux 内核中的 fifo 基于 pipe 实现，核心差别在于 open 操作。fifo 与 pipe 使用同一套 file_operations，相关代码如下：

const struct file_operations pipefifo_fops = {.open		= fifo_open,.llseek		= no_llseek,.read_iter	= pipe_read,.write_iter	= pipe_write,.poll		= pipe_poll,.unlocked_ioctl	= pipe_ioctl,.release	= pipe_release,.fasync		= pipe_fasync,.splice_write	= iter_file_splice_write,
};

pipe 通过 pipe 系统调用创建，内核创建两个 file 并绑定 file_operation 为 pipefifo_fops。

多进程同时操作两个 fifo 时潜在的锁死问题

UNPV2 Figure4.17 中有如下客户端与服务器端通信的操作：

上图中创建了两个 fifo，使用这两个 fifo 能够模拟全双工通信。然而当编码不当时，上述过程可能会触发进程死锁。

上图中 parent 进程创建 fifo1 与 fifo2 两个 fifo，然后将 fifo1 以只写方式打开，将 fifo2 以只读方式打开；child 进程使用相同路径将 fifo1 以只读方式打开，将 fifo2 以只写方式打开。parent 进程与 child 打开 fifo 的顺序一致而读写的模式刚好相反就能够唤醒阻塞的进程。

如果交换 parent 进程打开 fifo1 与 fifo2 的顺序，就会触发这两个进程死锁，parent 与 child 进程都阻塞在打开不同的 fifo 上，永远不会被唤醒。