深入理解子进程

import os

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

test_fork()

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

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')

import os, sys

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

import os

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

import os

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

import os

master, slave = os.openpty()
os.write(master, b'hello from master\n')
print(os.read(slave, 50))
# b'hello from master\n'
os.write(slave, b'hello from slave\n')
print(os.read(master, 50))
# b'hello from master\r\nhello from slave\r\n'

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."""

# 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)

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 == DEVNULL:
        p2cread = self._get_devnull()
    elif isinstance(stdin, int):
        p2cread = stdin
    else:
        # Assuming file-like object
        p2cread = stdin.fileno()

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)

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

# 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()

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

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

# 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)

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

# 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)

# 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 sh import ls
print(ls('-l'))

# magicmodule.py
import sys
from types import ModuleType

class MagicModule(ModuleType):

    def __init__(self, name):
        super().__init__(name)

    def __getattr__(self, name):
        return name

sys.modules[__name__] = MagicModule(__name__)

>>> from magicmodule import anything
>>> anything
'anything'
>>>

from sh import ErrorReturnCode_12
from sh import SignalException_9
from sh import SignalException_SIGHUP

try:
    some_cmd()
except ErrorReturnCode_12:
    print("couldn't do X")

def get_exc_from_name(name):
    """takes an exception name, like:

        ErrorReturnCode_1
        SignalException_9
        SignalException_SIGHUP

    and returns the corresponding exception.  this is primarily used for
    importing exceptions from sh into user code, for instance, to capture those
    exceptions
    """

rc_exc_regex = re.compile("(ErrorReturnCode|SignalException)_((\d+)|SIG[a-zA-Z]+)")

exc = ErrorReturnCodeMeta(name, (base,), {"exit_code": rc})

>>> sh.which('ls')
'/usr/bin/ls'

# L528
def is_exe(fpath):
    # 存在 & 可执行 & 是个文件
    return (os.path.exists(fpath) and
            os.access(fpath, os.X_OK) and
            os.path.isfile(os.path.realpath(fpath)))

# L554
for path in paths_to_search:
    exe_file = os.path.join(path, program)
    if is_exe(exe_file):
        found_path = exe_file
        break

def resolve_command_path(program):
    # 查找可执行文件
    path = which(program)
    if not path:
        # 替换下划线为横杠，再次尝试
        if "_" in program:
            path = which(program.replace("_", "-"))
        if not path:
            return None
    return path


# 导入命令时调用此函数，创建Command对象
def resolve_command(name, baked_args=None):
    path = resolve_command_path(name)
    cmd = None
    if path:
        cmd = Command(path)
        if baked_args:
            cmd = cmd.bake(**baked_args)
    return cmd

class Command(object):
    """ represents an un-run system program, like "ls" or "cd". """
    # L1065
    _call_args = {
        "fg": False, # run command in foreground

        # run a command in the background.  commands run in the background
        # ignore SIGHUP and do not automatically exit when the parent process
        # ends
        "bg": False,

        # ...一堆参数
    }
    # L1188
    def __init__(self, path, search_paths=None):
        found = which(path, search_paths)
        # L1209
        self._path = encode_to_py3bytes_or_py2str(found)

@staticmethod
def _extract_call_args(kwargs):
    """ takes kwargs that were passed to a command's __call__ and extracts
    out the special keyword arguments, we return a tuple of special keyword
    args, and kwargs that will go to the execd command """

    kwargs = kwargs.copy()
    call_args = {}
    for parg, default in Command._call_args.items():
        key = "_" + parg

        if key in kwargs:
            call_args[parg] = kwargs[key]
            del kwargs[key]

    invalid_kwargs = special_kwarg_validator(call_args,
            Command._kwarg_validators)

    if invalid_kwargs:
        exc_msg = []
        for args, error_msg in invalid_kwargs:
            exc_msg.append("  %r: %s" % (args, error_msg))
        exc_msg = "\n".join(exc_msg)
        raise TypeError("Invalid special arguments:\n\n%s\n" % exc_msg)

    return call_args, kwargs

>>> from sh import ls
>>> lslh = ls.bake('-l', '-h')
>>> lslh()
total 56K
-rw-r--r-- 1 guyskk guyskk  155 May 12 13:00 aaa.json
-rw-r--r-- 1 guyskk guyskk  162 May 12 13:00 bbb.py
...
>>> ls
<Command '/usr/bin/ls'>
>>> lslh
<Command '/usr/bin/ls -l -h'>
>>>

def bake(self, *args, **kwargs):
    fn = type(self)(self._path)
    fn._partial = True

    call_args, kwargs = self._extract_call_args(kwargs)

    pruned_call_args = call_args
    for k, v in Command._call_args.items():
        try:
            if pruned_call_args[k] == v:
                del pruned_call_args[k]
        except KeyError:
            continue

    fn._partial_call_args.update(self._partial_call_args)
    fn._partial_call_args.update(pruned_call_args)
    fn._partial_baked_args.extend(self._partial_baked_args)
    sep = pruned_call_args.get("long_sep", self._call_args["long_sep"])
    prefix = pruned_call_args.get("long_prefix",
            self._call_args["long_prefix"])
    fn._partial_baked_args.extend(compile_args(args, kwargs, sep, prefix))
    return fn

def __call__(self, *args, **kwargs):
    # ...中间的具体实现不太好理解，先看最后一行
    return RunningCommand(cmd, call_args, stdin, stdout, stderr)

# L1373
# check if we're piping via composition
stdin = call_args["in"]
if args:
    first_arg = args.pop(0)
    if isinstance(first_arg, RunningCommand):
        if first_arg.call_args["piped"]:
            stdin = first_arg.process
        else:
            stdin = first_arg.process._pipe_queue

    else:
        args.insert(0, first_arg)

if call_args["fg"]:
    if call_args["env"] is None:
        launch = lambda: os.spawnv(os.P_WAIT, cmd[0], cmd)
    else:
        launch = lambda: os.spawnve(os.P_WAIT, cmd[0], cmd, call_args["env"])

    exit_code = launch()

# stdout redirection
stdout = call_args["out"]
if output_redirect_is_filename(stdout):
    stdout = open(str(stdout), "wb")

class RunningCommand(object):
    """this represents an executing Command object."""
    def __init__(self, cmd, call_args, stdin, stdout, stderr):
        """
            cmd is an array, where each element is encoded as bytes (PY3) or str
            (PY2)
        """

# set up which stream should write to the pipe
# TODO, make pipe None by default and limit the size of the Queue
# in oproc.OProc
pipe = OProc.STDOUT
if call_args["iter"] == "out" or call_args["iter"] is True:
    pipe = OProc.STDOUT
elif call_args["iter"] == "err":
    pipe = OProc.STDERR

if spawn_process:
    # this lock is needed because of a race condition where a background
    # thread, created in the OProc constructor, may try to access
    # self.process, but it has not been assigned yet
    process_assign_lock = threading.Lock()
    with process_assign_lock:
        self.process = OProc(self, self.log, cmd, stdin, stdout, stderr,
                self.call_args, pipe, process_assign_lock)

    if should_wait:
        self.wait()

# convenience
ca = self.call_args

>>> proc = sh.python('hello.py', _iter='out')
>>> for line in proc:
...     print(line)
...
hello world

hello world

hello world

# L1770
elif ca["tty_in"]:
    self._stdin_read_fd, self._stdin_write_fd = pty.openpty()

# tty_in=False is the default
else:
    self._stdin_write_fd, self._stdin_read_fd = os.pipe()

# L1782
# tty_out=True is the default
elif ca["tty_out"]:
    self._stdout_read_fd, self._stdout_write_fd = pty.openpty()

else:
    self._stdout_read_fd, self._stdout_write_fd = os.pipe()