深入理解子进程Python 相关源码解析

很多时候，我需要写脚本去做一些自动化操作，简单的可以直接写 Shell 脚本，但一些稍复杂的情况， 比如要用到分支语句，循环语句，或者调用一些高级函数，用 Shell 就太费劲了。 我更喜欢用一种完整的语言(比如 Python)，调用 Shell 程序并获取它的输出，执行复杂操作。

本文介绍 UNIX 进程的创建过程(fork, exec)，如何与子进程通信(pipe, pty)， 并深入分析标准库中的 subprocess 模块和著名的第三方库 sh 的源码，阐述其实现原理。

这些库提供了非常易用的接口，大部分时候只要看看文档就能解决问题。 但它们也无法彻底掩盖实现细节，The Law of Leaky Abstractions - 抽象漏洞法则 解释了许多不完美的抽象，希望这篇文章也能帮你更好地理解子进程。

UNIX 进程的创建过程

fork

fork 函数以复制调用进程的方式创建新进程，它 调用一次，返回两次，在子进程中它返回0， 在父进程中它返回子进程的 ID。

import os
def test_fork:
pid = os.fork
if pid == 0:
print('child')
os._exit(0)
else:
print('parent')
test_fork

fork 之后父子进程各自继续往下执行，所以示例中的两个 print都会执行， 子进程通过os._exit退出，否则两个进程共用标准输入，shell 就不能正常工作了。

fork是个系统调用，可通过man fork.2查看其手册。

_exit也是个系统调用，它与sys.exit的区别在于：_exit直接退出， 而sys.exit先执行一些清理工作再退出。可通过man exit.2和man exit.3查看其手册。

exec

exec 是一系列函数，总共有七个，它把当前进程执行的程序替换成新程序，正常情况下它们 调用一次，永不退出。

import os
os.execv('/bin/sh', ['sh'])

执行示例中的代码之后，进程就变成了 sh，无法再回到 python。

exec 函数中，通常 execve 是系统调用，其他几个都能通过它来实现， 可通过 man execve.2和man exec.3查看其手册。

waitpid

通常 fork 之后，子进程调用 exec 执行新程序，父进程要等待子进程的结果， 这就可以通过 waitpid 实现。如果父进程没有调用 waitpid 获取子进程的退出状态， 那么子进程退出后，状态信息会一直保留在内存中，这样的子进程也被称为僵尸进程。

import os
def system(cmd):
pid = os.fork
if pid == 0:
os.execv('/bin/sh', ['sh', '-c', cmd])
else:
return os.waitpid(pid, 0)
system('python --version')

这就是 os.system的实现原理，子进程共用父进程的标准输入输出，父进程阻塞，直到子进程结束。 因为子进程的输出是直接送到标准输出，所以无法被父进程获取。

可通过 man waitpid.2和man system.3查看其手册。

dup2

int dup2(int oldfd, int newfd);

dup2 可以复制文件描述符, 它会先把 newfd 关闭，再把 oldfd 复制到 newfd， 经常用它来修改进程的标准 I/O。

import os, sys
out = open('out.txt', 'w')
os.dup2(out.fileno, sys.stdout.fileno)
print('这段文字会输出到 out.txt 中，而不是控制台')

可通过 man dup2.2查看手册。

进程间通信

管道

进程之间有很多种通信方式，这里只讨论 管道这种方式，这也是最常用的一种方式。 管道通常是半双工的，只能一端读，另一端写，为了可移植性，不能预先假定系统支持全双工管道。

import os
r, w = os.pipe
os.write(w, b'hello')
print(os.read(r, 10))

可通过 man pipe.2查看其手册。

缓冲 I/O

I/O 可分为：无缓冲，行缓冲，全缓冲三种。

通过 read 和 write 系统调用直接读写文件，就是无缓冲模式，性能也最差。 而通过标准 I/O 库读写文件，就是缓冲模式，标准 I/O 库提供缓冲的目的是尽可能减少 read 和 write 调用的次数，提高性能。

行缓冲模式，当在输入输出中遇到换行符时，才进行实际 I/O 操作。 全缓冲模式，当填满缓冲区时，才进行实际 I/O 操作。

管道和普通文件默认是全缓冲的，标准输入和标准输出默认是行缓冲的，标准错误默认是无缓冲的。

import os
r, w = os.pipe
fw = os.fdopen(w, 'wb')
fr = os.fdopen(r, 'rb')
fw.write(b'hello')
print(fr.read)

这个例子中，读管道这步会一直阻塞。有两个原因：

1. 写管道有缓冲，没有进行实际 I/O，所以读端读不到数据

2. 读管道也有缓冲，要读满缓冲区才会返回

只要满足其中任何一个条件都会阻塞。通常写管道是在子进程中进行，我们无法控制其缓冲， 这也是管道的一个局限性。

伪终端

伪终端看起来就像一个双向管道，一端称为 master(主)，另一端称为 slave(从)。 从端看上去和真实的终端一样，能够处理所有的终端 I/O 函数。

import os
master, slave = os.openpty
os.write(master, b'hello from mastern')
print(os.read(slave, 50))
# b'hello from mastern'
os.write(slave, b'hello from slaven')
print(os.read(master, 50))
# b'hello from masterrnhello from slavern'

伪终端 echo 默认是开启的，所以最后 master 读的时候，会先读出之前写入的数据。

如果写入的数据没有换行符，就可能不会被传送到另一端， 造成读端一直阻塞（猜测是伪终端的底层实现使用了行缓冲）。

可通过 man openpty.3和man pty.7查看其手册。

subprocess 的实现

subprocess 提供了很多接口， 其核心是 Popen(Process Open)，其他部分都通过它来实现。subprocess 也支持 Windows 平台， 这里只分析它在 Unix 平台的实现。

subprocess 源码: https://github.com/python/cpython/blob/3.6/Lib/subprocess.py#L540

首先看一下接口原型(L586)，参数非常多：

def __init__(self, args, bufsize=-1, executable=None,
stdin=None, stdout=None, stderr=None,
preexec_fn=None, close_fds=_PLATFORM_DEFAULT_CLOSE_FDS,
shell=False, cwd=None, env=None, universal_newlines=False,
startupinfo=None, creationflags=0,
restore_signals=True, start_new_session=False,
pass_fds=, *, encoding=None, errors=None):
"""Create new Popen instance."""

然后看到中间(L648) 一大段注释：

# Input and output objects. The general principle is like
# this:
#
# Parent Child
# ------ -----
# p2cwrite ---stdin---> p2cread
# c2pread <--stdout--- c2pwrite
# errread <--stderr--- errwrite
#
# On POSIX, the child objects are file descriptors. On
# Windows, these are Windows file handles. The parent objects
# are file descriptors on both platforms. The parent objects
# are -1 when not using PIPEs. The child objects are -1
# when not redirecting.
(p2cread, p2cwrite,
c2pread, c2pwrite,
errread, errwrite) = self._get_handles(stdin, stdout, stderr)

父子进程通过这三对文件描述符进行通信。

再跳到 _get_handles的实现(L1144)：

def _get_handles(self, stdin, stdout, stderr):
"""Construct and return tuple with IO objects:
p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite
"""
p2cread, p2cwrite = -1, -1
c2pread, c2pwrite = -1, -1
errread, errwrite = -1, -1
if stdin is None:
pass
elif stdin == PIPE:
p2cread, p2cwrite = os.pipe
elif stdin == DEV:
p2cread = self._get_dev
elif isinstance(stdin, int):
p2cread = stdin
else:
# Assuming file-like object
p2cread = stdin.fileno

它创建了三对文件描述符， stdin，stdout，stderr这三个参数都是类似的， 分别对应一对文件描述符。当参数等于PIPE时，它就会创建一条管道，这也是最常用的参数。

回到 Popen，接着往下看(L684)：

if p2cwrite != -1:
self.stdin = io.open(p2cwrite, 'wb', bufsize)
if text_mode:
self.stdin = io.TextIOWrapper(self.stdin, write_through=True,
line_buffering=(bufsize == 1),
encoding=encoding, errors=errors)
if c2pread != -1:
self.stdout = io.open(c2pread, 'rb', bufsize)
if text_mode:
self.stdout = io.TextIOWrapper(self.stdout,
encoding=encoding, errors=errors)
if errread != -1:
self.stderr = io.open(errread, 'rb', bufsize)
if text_mode:
self.stderr = io.TextIOWrapper(self.stderr,
encoding=encoding, errors=errors)
self._execute_child(args, executable, preexec_fn, close_fds,
pass_fds, cwd, env,
startupinfo, creationflags, shell,
p2cread, p2cwrite,
c2pread, c2pwrite,
errread, errwrite,
restore_signals, start_new_session)

它把其中三个文件描述符变成了文件对象，用于在父进程中和子进程通信。 注意 bufsize参数只作用于父进程中的文件对象，子进程中的缓冲是没法控制的。

encoding和errors参数就不赘述了，可以查看 io 模块的文档。

再看 _execute_child的实现(L1198):

def _execute_child(self, args, executable, preexec_fn, close_fds,
pass_fds, cwd, env,
startupinfo, creationflags, shell,
p2cread, p2cwrite,
c2pread, c2pwrite,
errread, errwrite,
restore_signals, start_new_session):
"""Execute program (POSIX version)"""
if isinstance(args, (str, bytes)):
args = [args]
else:
args = list(args)
if shell:
args = ["/bin/sh", "-c"] args
if executable:
args[0] = executable

看到这里， args,shell,executable这几个参数就很好理解了。

args可以是字符串或列表，shell表示通过 shell 程序执行命令，executable是 shell 程序的路径，默认是/bin/sh。

当 args是字符串时，通常配合shell=True使用，用来执行任意 shell 命令。

接着往下看(L1221):

# For transferring possible exec failure from child to parent.
# Data format: "exception name:hex errno:description"
# Pickle is not used; it is complex and involves memory allocation.
errpipe_read, errpipe_write = os.pipe

这是第四对文件描述符。子进程在 fork 之后到 exec 之前会执行许多步骤， 万一这些步骤失败了，就通过这对文件描述向父进程传递异常信息，父进程收到后抛出相应的异常。

创建子进程并执行命令(L1252)：

fds_to_keep = set(pass_fds)
fds_to_keep.add(errpipe_write)
self.pid = _posixsubprocess.fork_exec(
args, executable_list,
close_fds, sorted(fds_to_keep), cwd, env_list,
p2cread, p2cwrite, c2pread, c2pwrite,
errread, errwrite,
errpipe_read, errpipe_write,
restore_signals, start_new_session, preexec_fn)
self._child_created = True

继续追踪 _posixsubprocess源码： https://github.com/python/cpython/blob/3.6/Modules/_posixsubprocess.c#L545

static PyObject *
subprocess_fork_exec(PyObject* self, PyObject *args){
// ...省略一堆处理参数的代码
pid = fork;
if (pid == 0) {
// L692
child_exec(exec_array, argv, envp, cwd,
p2cread, p2cwrite, c2pread, c2pwrite,
errread, errwrite, errpipe_read, errpipe_write,
close_fds, restore_signals, call_setsid,
py_fds_to_keep, preexec_fn, preexec_fn_args_tuple);
}
// L733
return PyLong_FromPid(pid);

在子进程中执行 child_exec，父进程返回 pid。

继续看 child_exec 的实现(L390)：

static void
child_exec(/*一堆参数*/)
{
int i, saved_errno, reached_preexec = 0;
PyObject *result;
const char* err_msg = "";
/* Buffer large enough to hold a hex integer. We can't malloc. */
char hex_errno[sizeof(saved_errno)*2 1];
if (make_inheritable(py_fds_to_keep, errpipe_write) < 0)
goto error;

make_inheritable把 Popen 中pass_fds参数指定的文件描述符设为可继承， 它通过对每个文件描述符调用_Py_set_inheritable，清除close-on-exec标志。

需要注意，父进程的标准 I/O 默认是可继承的。

参考 PEP 446 – Make newly created file descriptors non-inheritable

int flags, res;
flags = fcntl(fd, F_GETFD);
if (flags == -1) { /* handle the error */ }
flags |= FD_CLOEXEC;
/* or "flags &= ~FD_CLOEXEC;" to clear the flag */
res = fcntl(fd, F_SETFD, flags);
if (res == -1) { /* handle the error */ }

close-on-exec(FD_CLOEXEC) 标志表示这个文件会在进程执行 exec 系统调用时自动关闭。

接着往下看(L430):

/* Dup fds for child.
dup2 removes the CLOEXEC flag but we must do it ourselves if dup2
would be a no-op (issue #10806). */
if (p2cread == 0) {
if (_Py_set_inheritable(p2cread, 1, ) < 0)
goto error;
}
else if (p2cread != -1)
POSIX_CALL(dup2(p2cread, 0)); /* stdin */

通过 dup2 系统调用，把 p2cread 设为子进程的标准输入。标准输出和标准错误也是类似的。

接着往下看(L463):

if (cwd)
POSIX_CALL(chdir(cwd));
if (restore_signals)
_Py_RestoreSignals;
#ifdef HAVE_SETSID
if (call_setsid)
POSIX_CALL(setsid);
#endif

cwd参数，设置当前工作目录。

restore_signals参数，把信号处理恢复为默认值，涉及信号处理的内容，这里略过。

call_setsid，即 Popen 的start_new_session参数， 创建会话并设置进程组 ID，内容太多也略过。

接着往下看(L474):

reached_preexec = 1;
if (preexec_fn != Py_None && preexec_fn_args_tuple) {
/* This is where the user has asked us to deadlock their program. */
result = PyObject_Call(preexec_fn, preexec_fn_args_tuple, );
if (result == ) {
/* Stringifying the exception or traceback would involve
* memory allocation and thus potential for deadlock.
* We've already faced potential deadlock by calling back
* into Python in the first place, so it probably doesn't
* matter but we avoid it to minimize the possibility. */
err_msg = "Exception occurred in preexec_fn.";
errno = 0; /* We don't want to report an OSError. */
goto error;
}
/* Py_DECREF(result); - We're about to exec so why bother? */
}
/* close FDs after executing preexec_fn, which might open FDs */
if (close_fds) {
/* TODO HP-UX could use pstat_getproc if anyone cares about it. */
_close_open_fds(3, py_fds_to_keep);
}

执行 preexec_fn，随后根据close_fds参数判断是否关闭打开的文件描述符。

最后，执行命令(L497)：

/* This loop matches the Lib/os.py _execvpe's PATH search when */
/* given the executable_list generated by Lib/subprocess.py. */
saved_errno = 0;
for (i = 0; exec_array[i] != ; i) {
const char *executable = exec_array[i];
if (envp) {
execve(executable, argv, envp);
} else {
execv(executable, argv);
}
if (errno != ENOENT && errno != ENOTDIR && saved_errno == 0) {
saved_errno = errno;
}
}

尝试执行命令，只要有一个成功，后面的就不会执行了，因为 exec 执行成功后永不返回。

Popen 的创建过程到这就结束了，子进程已经运行起来了，接下来分析如何与子进程通信。

跳到 Popen 中的 communicate 方法(L796):

def communicate(self, input=None, timeout=None):
if self._communication_started and input:
raise ValueError("Cannot send input after starting communication")

这个 raise ValueError是为什么呢？稍后解答。

继续往下看(L813):

# Optimization: If we are not worried about timeouts, we haven't
# started communicating, and we have one or zero pipes, using select
# or threads is unnecessary.
if (timeout is None and not self._communication_started and
[self.stdin, self.stdout, self.stderr].count(None) >= 2):
stdout = None
stderr = None
if self.stdin:
self._stdin_write(input)
elif self.stdout:
stdout = self.stdout.read
self.stdout.close
elif self.stderr:
stderr = self.stderr.read
self.stderr.close
self.wait
else:
if timeout is not None:
endtime = _time timeout
else:
endtime = None
try:
stdout, stderr = self._communicate(input, endtime, timeout)
finally:
self._communication_started = True
sts = self.wait(timeout=self._remaining_time(endtime))
return (stdout, stderr)

if部分直接和子进程通信，else部分调用了self._communicate， 用线程和 I/O 多路复用的方式与子进程通信，去掉影响也不大，这部分细节就不分析了。 注意self._stdin_write(input)很关键，在self._communicate中也是调用它向子进程写数据。

看 _stdin_write的实现(L773)：

def _stdin_write(self, input):
# L776
self.stdin.write(input)
# L787
self.stdin.close

省略了异常处理的代码，主要就这两句。可以看到写完输入之后，立即把标准输入关闭了。 结合上面的 raise ValueError，可以看出只能向子进程写入一次。

这涉及到缓冲的问题，管道是默认全缓冲的，如果不立即关闭写端，子进程读的时候就可能一直阻塞， 我们没法控制子进程使用什么类型的缓冲。同样的，如果子进程一直写输出，父进程也会读不到数据， 只有子进程退出之后，系统自动关闭它的标准输出，父进程才能读到数据。

下面是个例子：

# hello.py
from time import sleep
while True:
print('hello world')
sleep(1)
# test_hello.py
from subprocess import *
p = Popen(['python', 'hello.py'], stdin=PIPE, stdout=PIPE)
out, err = p.communicate
print(out)

运行 python test_hello.py，尽管子进程在不停地输出，但是父进程一直读取不到数据。 此时如果我们把子进程杀死，父进程便会立即读到数据。

subprocess 的核心代码到这就分析完了。 再放个自己实现的 popen（未考虑各种异常情况，仅用于说明实现原理）：

# popen.py
import os
class popen:
def __init__(self, cmd, *, cwd=None, env=None, input=None):
argv = ['/bin/sh', '-c', cmd]
# Parent Child
# ------ -----
# p2cwrite ---stdin---> p2cread
# c2pread <--stdout--- c2pwrite
# errread <--stderr--- errwrite
p2cread, p2cwrite = os.pipe
c2pread, c2pwrite = os.pipe
errread, errwrite = os.pipe
pid = os.fork
if pid == 0:
os.close(p2cwrite)
os.close(c2pread)
os.close(errread)
os.dup2(p2cread, 0)
os.dup2(c2pwrite, 1)
os.dup2(errwrite, 2)
if cwd:
os.chdir(cwd)
if env:
os.execve('/bin/sh', argv, env)
else:
os.execv('/bin/sh', argv)
else:
os.close(p2cread)
os.close(c2pwrite)
os.close(errwrite)
stdin = open(p2cwrite, 'w')
stdout = open(c2pread, 'r')
stderr = open(errread, 'r')
# communicate
if input:
stdin.write(input)
stdin.close
self.out = stdout.read
stdout.close
self.err = stderr.read
stderr.read
self.status = os.waitpid(pid, 0)[1]
def __repr__(self):
fmt = ('STATUS:{}n' 
'OUTPUT'.center(80, '-') 'n{}n' 
'ERROR'.center(80, '-') 'n{}n')
return fmt.format(self.status, self.out, self.err)

执行命令试一试：

>>> from popen import popen
>>> popen('python -c "import this"')
STATUS:0
-------------------------------------OUTPUT-------------------------------------
The Zen of Python, by Tim Peters
Beautiful is better than ugly.
Explicit is better than implicit.
Simple is better than complex.
Complex is better than complicated.
Flat is better than nested.
Sparse is better than dense.
Readability counts.
Special cases aren't special enough to break the rules.
Although practicality beats purity.
Errors should never pass silently.
Unless explicitly silenced.
In the face of ambiguity, refuse the temptation to guess.
There should be one-- and preferably only one --obvious way to do it.
Although that way may not be obvious at first unless you're Dutch.
Now is better than never.
Although never is often better than *right* now.
If the implementation is hard to explain, it's a bad idea.
If the implementation is easy to explain, it may be a good idea.
Namespaces are one honking great idea -- let's do more of those!
-------------------------------------ERROR--------------------------------------
>>> popen('python -c "import that"')
STATUS:256
-------------------------------------OUTPUT-------------------------------------
-------------------------------------ERROR--------------------------------------
Traceback (most recent call last):
File "<string>", line 1, in <module>
ModuleNotFoundError: No module named 'that'
>>> popen('python --version').out
'Python 3.6.1n'
>>>

题图：pexels，CC0 授权。