keakon的涂鸦馆

1. 本文描述的环境为 Linux 操作系统（也适用于大部分 POSIX 系统）下的 CPython，可能不适用于 Windows 操作系统或其他的 Python 实现。
   
2. 为了避免歧义，下文用「主进程」或「当前进程」表示创建子进程的那个进程，而不使用「父进程」，除非明确指明了「xx 的父进程」。
   

import os

try:
    pid = os.fork()
    if pid == 0:  # 子进程
        # 子进程的代码
    else:  # 主进程
        # 主进程的代码
except OSError:
    # 主进程处理 fork 失败的代码

• 设置子进程自己的 PID，将其 PPID 设为主进程的 PID，而 PGID、SID、UID、GID、CWD 和 UMASK 则保持和主进程一致。
  
• 复制主进程的所有地址空间到子进程。也就是说，子进程可以直接使用主进程的变量，包括 import 过的模块等。但是这种复制是写时拷贝（copy on write），也就是一旦你对它进行了改动，它和主进程中的变量就不是同一个了，因此表现出来的就像是值拷贝一样。
  
• 复制主进程的环境变量到子进程。
  
• 复制主进程的所有文件描述符（file descriptors）到子进程。文件描述符的状态在这些进程间是共享的，所以在一个进程中读写文件，会影响另一个进程的文件打开位置。但是在一个进程中关闭文件描述符，并不影响另一个进程对应的文件描述符。
  
• 复制主进程的信号处理函数（由 signal.signal() 注册）、信号唤醒 fd（由 signal.set_wakeup_fd() 设置），是否重新启动被信号中断的系统调用（由 signal.siginterrupt() 设置），以及阻塞哪些信号（由 signal.pthread_sigmask() 设置）到子进程。但是，定时器（由 signal.setitimer() 设置）只对当前进程有效，不复制到子进程。
  
• 保留通过 atexit() 和 on_exit() 注册的函数。
  
• 对于多线程程序而言，只有调用 fork() 的线程被复制到子进程，变成一个单线程的新进程。
  
• 异步 IO 被取消。
  

try:
    with open('/tmp/stdout', 'wb') as stdout:
        sys.stdout = stdout
        print 'this will be written to file'
finally:
    sys.stdout = sys.__stdout__

def redirect_fd(fd, path, mode):
    try:
        os.close(fd)
    except OSError:
        pass
    new_fd = os.open(path, mode)
    if new_fd != fd:
        os.dup2(new_fd, fd)
        os.close(new_fd)

redirect_fd(0, os.devnull, os.O_RDONLY)
redirect_fd(1, '/tmp/stdout', os.O_WRONLY | os.O_CREAT | os.O_TRUNC)
redirect_fd(2, '/tmp/stderr', os.O_WRONLY | os.O_CREAT | os.O_TRUNC)

r, w = os.pipe()
pid = os.fork()
if pid == 0:
    os.close(w)  # 如果有必要的话，可以重定向到标准输入输出
    print os.read(r, 1024)
else:
    os.close(r)
    os.write(w, 'test')

MAX_FD = os.sysconf("SC_OPEN_MAX")  # 或者 resource.getrlimit(resource.RLIMIT_NOFILE)[0]
os.closerange(3, MAX_FD)

1. 在信号处理函数中，只调用 async-signal-safe 的函数，且对于全局变量（含静态变量）的操作是可重入的。
   
2. 在整个程序中，一旦调用非 async-signal-safe 的函数，或者操作可能被信号处理函数使用的全局变量时，先阻塞信号（即不会有信号处理函数被调用）。
   

signal.signal(signal.SIGALRM, handle_alarm)
signal.siginterrupt(signal.SIGALRM, False)

• 打开的文件描述符不会被刷新。
  
• 设置的信号处理函数恢复为 SIG_DFL，但被设为 SIG_IGN 的保持不变。信号的阻塞也保持不变。
  
• 定时器被取消。
  
• 共享内存被取消映射。
  
• 通过 atexit() 和 on_exit() 注册的函数被取消。
  
• 异步 IO 被取消。
  
• 除当前线程外，该进程的其他线程都被销毁。
  

os.execv('/bin/ls', ['/bin/ls', '/'])
os.execv('/bin/ls', ['balabala', '/'])

os.execv('/bin/sh', ['/bin/sh', '-c', 'ls / && sleep 1 && ls /tmp'])

flags = fcntl.fcntl(fd, fcntl.F_GETFD)
fcntl.fcntl(fd, fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC)

def wait_pid(pid, options):
    while True:
        try:
            return os.waitpid(pid, options)
        except OSError as e:
            if e.errno == errno.EINTR:
                continue
            raise

• fflush() 函数：用于刷新 stdio 流（stream）。
  对于输出流，它强制将用户空间的缓冲区（buffer），通过这个流底层的 write() 函数写入到该流。对于关联到可定位（seekable）文件的输入流，它丢弃掉已经被底层文件读取（fetch），但还没被应用消费（consume）的 buffer（因为对于 seekable 的文件而言，只需要重新定位再读一次即可）。如果它的 FILE *stream 参数为 NULL，则刷新所有打开的流。简单来说，它的作用主要是将用户空间的缓冲区写入内核空间。
  但是用户空间的缓冲区是和所用的语言和库相关的，fflush() 刷新的只是 C 语言的 stdio 库创建的 FILE 类型的 buffer。对于 Python 而言，你应该使用 file.flush() 方法；如果没有输出缓冲区的话，直接使用 os.write() 和 socket.send() 就能写入内核空间了，不需要刷新。
  
• fsync() 和 fdatasync() 系统调用：用于将内核中缓存的对文件的修改写入磁盘。两者的区别在于后者不写入元数据（metadata）。它们都是 async-signal-safe 的函数。
  很显然，这两个函数都只对磁盘文件有效，不能用于刷新 socket。此外，磁盘也有自己的缓冲区，内核也只是将数据写入磁盘缓冲区而已，并不一定真正被写入磁盘了。macOS 的 fcntl() 系统调用提供了 F_FULLFSYNC 选项，可以要求将数据写入磁盘的永久存储设备中（只对 HFS、FAT 和 UDF 文件系统有效）。
  大部分的程序都不需要用到它们，因为只要不是断电等意外原因导致非正常关机的话，内核和磁盘驱动最终都会将文件写入磁盘的（尽管那时你可能已经关闭文件或者退出进程了）。要考虑到断电时的数据安全的话，主要也就是数据库之类的应用了。
  Python 提供了 os.fsync() 函数来封装 fsync() 系统调用。
  
• aio_fsync() 函数：用于刷新异步 IO 操作到磁盘。
  功能类似于 fsync() 和 fdatasync(2)，但是只发出请求，不等待实际的操作完成就返回。
  
• sync() 和 syncfs() 系统调用：将文件系统的缓存写入磁盘。二者区别在于后者只刷新该 fd 所对应的文件的文件系统。
  很显然，这是两个很重的系统调用，一般不需要使用。
  

class Daemon(object):
    def __init__(self):
        self._action_id = None
        self._done_action = False

    def run(self):
        while True:
            start_time = time.time()

            status = self.collect_status()
            if status and self.report_status(status) and self._done_action:
                self.clean_status()

            if not self._action_id:
                action = self.fetch_action()
                if action and self.accept_action(action):
                    self.handle_action(action)

            wait_time = FETCH_ACTION_INTERVAL - (time.time() - start_time)
            if wait_time > 0:
                self.wait(wait_time)

    def fetch_action(self):
        raise NotImplementedError

    def handle_action(self, action):
        raise NotImplementedError

    def accept_action(self, action):
        raise NotImplementedError

    def collect_status(self):
        raise NotImplementedError

    def report_status(self, status):
        raise NotImplementedError

    def clean_status(self):
        raise NotImplementedError

    @staticmethod
    def wait(wait_time):
        time.sleep(wait_time)

FETCH_ACTION_INTERVAL = 5

class Daemon(object):
    def run(self):
        try:
            self.register_signals()
            
            while True:
                start_time = time.time()

                status = self.collect_status()
                if status and self.report_status(status) and self._done_action:
                    self.clean_status()

                if not self._action_id:
                    action = self.fetch_action()
                    if action and self.accept_action(action):
                        self.handle_action(action)

                wait_time = FETCH_ACTION_INTERVAL - (time.time() - start_time)
                if wait_time > 0:
                    self.wait(wait_time)
        finally:
            self.deregister_signals()

    @staticmethod
    def ignore_signal(signum, frame):
        return

    def register_signals(self):
        signal.signal(signal.SIGALRM, self.ignore_signal)
        signal.signal(signal.SIGCHLD, self.ignore_signal)

    def deregister_signals(self):
        signal.setitimer(signal.ITIMER_REAL, 0)
        signal.signal(signal.SIGALRM, signal.SIG_DFL)
        signal.signal(signal.SIGCHLD, signal.SIG_DFL)

    def wait(self):
        signal.setitimer(signal.ITIMER_REAL, wait_time)
        signal.pause()
        signal.setitimer(signal.ITIMER_REAL, 0)

class Daemon(object):
    def __init__(self):
        self._action_id = None
        self._done_action = False
        self._selector = None
        r, w = os.pipe()
        set_non_blocking(r)
        set_non_blocking(w)  # wakeup fd 需要是非阻塞的
        self._waker = r, w

    def register_signals(self):
        signal.signal(signal.SIGCHLD, self.ignore_signal)
        self._selector = selectors.DefaultSelector()
        self._selector.register(self._waker[0], selectors.EVENT_READ)
        signal.set_wakeup_fd(self._waker[1])

    def deregister_signals(self):
        signal.signal(signal.SIGALRM, signal.SIG_DFL)
        signal.set_wakeup_fd(-1)
        if self._selector:
            self._selector.close()
            self._selector = None

    def wait(self, wait_time):
        ready = self._selector.select(wait_time)
        if ready:
            waker = self._waker[0]
            # 读完并忽略所有的数据，避免 _waker[1] 被阻塞而无法写入
            try:
                while True:
                    data = os.read(waker, BUF_SIZE)
                    if not data or len(data) < BUF_SIZE:  # 没有更多数据了
                        break
            except OSError:  # 忽略 EAGAIN 和 EINTR
                pass


def set_non_blocking(fd):
    flags = fcntl.fcntl(fd, fcntl.F_GETFL)
    fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)


BUF_SIZE = 1024

class Worker(object):
    def __init__(self, action):
        try:
            os.setsid()
        except OSError:
            worker_logger.warning('failed to setsid, the worker should be created in a forked process')

        user_name = action.get('user_name')
        if not user_name:
            raise InvalidAction('user_name is required')
        if not isinstance(user_name, basestring):
            raise InvalidAction('user_name must be a string object')

        try:
            user = pwd.getpwnam(user_name)
        except KeyError:
            raise UserNotFound

        group_name = action.get('group_name')
        if group_name:
            if not isinstance(group_name, basestring):
                raise InvalidAction('group_name must be a string object')
            try:
                user_group = grp.getgrnam(group_name)
            except KeyError:
                raise GroupNotFound
            if user_name not in user_group.gr_mem:
                raise UserNotInGroup('user %s not in group %s' % (user_name, group_name))
            os.setgid(user_group.gr_gid)
        else:
            os.setgid(user.pw_gid)

        os.setuid(user.pw_uid)
        os.chdir(user.pw_dir)

        redirect_fd(0, os.devnull, os.O_RDONLY)  # 避免运行一些需要 TTY 的命令，如 top
        redirect_fd(1, STDOUT_FILE_PATH), os.O_WRONLY | os.O_CREAT | os.O_TRUNC)
        redirect_fd(2, STDERR_FILE_NAME), os.O_WRONLY | os.O_CREAT | os.O_TRUNC)

        self.action = action

        self.save_worker_pid()

    def do_action(self):
        gc.disable()  # https://bugs.python.org/issue1336
        try:
            pid = os.fork()
        except OSError:
            worker_logger.exception('failed to fork executor for action ' + self.action.id)
            gc.enable()
            raise
        else:
            if pid == 0:  # executor
                worker_logger.handlers[0].close()
                os.execve('/bin/bash', ['/bin/bash', '--norc', '-c', self.action.execution], {})  # 使用干净的环境来执行
            else:  # worker
                gc.enable()
                result = wait_pid(pid, 0)
                executor_exit_status = result[1]
                self.save_executor_exit_status(exit_status)

    def save_worker_pid(self):
        raise NotImplementedError

    def save_executor_exit_status(self, exit_status):
        raise NotImplementedError


def redirect_fd(fd, path, mode):
    try:
        os.close(fd)
    except OSError:
        pass
    new_fd = os.open(path, mode)
    if new_fd != fd:
        os.dup2(new_fd, fd)
        os.close(new_fd)


def wait_pid(pid, options):
    while True:
        try:
            return os.waitpid(pid, options)
        except OSError as e:
            if e.errno == errno.EINTR:  # 忽略等待时被中断
                continue
            raise

class Daemon(object):
    def handle_action(self, action):
        gc.disable()
        try:
            pid = os.fork()
        except OSError:
            daemon_logger.exception('failed to create worker for action ' + action.id)
            gc.enable()
            # 保存 action exit status
        else:
            if pid == 0:  # worker
                self.deregister_signals()
                os.close(self._waker[0])
                os.close(self._waker[1])
                daemon_logger.handlers[0].close()
                gc.enable()
                exit_code = 0
                try:
                    worker = Worker(action)
                    worker.do_action()
                except SystemExit as e:
                    exit_code = e.code
                except:
                    exit_code = 1  # 根据实际的设置返回不同的错误码
                finally:
                    try:
                        logging.shutdown()  # 刷新未写入的日志
                    except:
                        pass
                    os._exit(exit_code)
            else:  # daemon
                gc.enable()
                # 保存 worker PID，因为 worker 有可能还没保存 PID 就退出了