gunicorn源码分析

from setuptools import setup, find_packages


setup(
    entry_points="""
    [console_scripts]
    gunicorn=gunicorn.app.wsgiapp:run
    """,
)

def run():
    from gunicorn.app.wsgiapp import WSGIApplication
    WSGIApplication("%(prog)s [OPTIONS] [APP_MODULE]").run()

class WorkerConnections(Setting):
    name = "worker_connections"
    section = "Worker Processes"
    cli = ["--worker-connections"]
    meta = "INT"
    validator = validate_pos_int
    type = int
    default = 1000
    desc = """\
        The maximum number of simultaneous clients.

        This setting only affects the Eventlet and Gevent worker types.
        """

def run(self):
    try:
        Arbiter(self).run()
    except RuntimeError as e:
        print("\nError: %s\n" % e, file=sys.stderr)
        sys.stderr.flush()
        sys.exit(1)

def run(self):
    "Main master loop."
    # 1.初始化实例，并信号监听以及sock
    self.start()
    util._setproctitle("master [%s]" % self.proc_name)

    try:
        # 2.创建worker直到worker数量满足用户的指定条件，如果当前worker数量超过用户指定的条件，则会杀掉创建时间比较长的worker
        self.manage_workers()

        while True:
            # 3.负责判断该进程是否是正真的master，如果是则提升为正真的master(这一块放在最后一部分进行分析)
            self.maybe_promote_master()

            # 4.获取信号
            sig = self.SIG_QUEUE.pop(0) if self.SIG_QUEUE else None
            if sig is None:
                # 5.利用select休眠1小时
                self.sleep()
                # 6.判断worker是否超时，如果是则杀掉worker(将在Worker章节进行分析) 
                self.murder_workers()
                self.manage_workers()
                continue

            if sig not in self.SIG_NAMES:
                self.log.info("Ignoring unknown signal: %s", sig)
                continue

            signame = self.SIG_NAMES.get(sig)
            # 调用对应的信号处理
            handler = getattr(self, "handle_%s" % signame, None)
            if not handler:
                self.log.error("Unhandled signal: %s", signame)
                continue
            self.log.info("Handling signal: %s", signame)
            handler()
            # 7.这样下次循环就不会等待一秒了
            self.wakeup()
    # 8.服务异常，进行退出处理
    except (StopIteration, KeyboardInterrupt):
        # 收到用户的退出信号(按下CTRL+C) 
        self.halt()
    except HaltServer as inst:
        # Worker运行异常的时候
        self.halt(reason=inst.reason, exit_status=inst.exit_status)
    except SystemExit:
        raise
    except Exception:
        # 其它的运行异常
        self.log.info("Unhandled exception in main loop",
                      exc_info=True)
        self.stop(False)
        if self.pidfile is not None:
            self.pidfile.unlink()
        sys.exit(-1)

def init_signals(self):
    """\
    Initialize master signal handling. Most of the signals
    are queued. Child signals only wake up the master.
    """
    # close old PIPE
    for p in self.PIPE:
        os.close(p)

    # initialize the pipe
    self.PIPE = pair = os.pipe()
    for p in pair:
        util.set_non_blocking(p)
        util.close_on_exec(p)

    self.log.close_on_exec()

    # initialize all signals
    for s in self.SIGNALS:
        signal.signal(s, self.signal)
    # 捕获子进程exit前发出的sigchld信号,然后函数里面通过waitpid来收拾，避免僵尸进程的存在
    signal.signal(signal.SIGCHLD, self.handle_chld)

def signal(self, sig, frame):
    if len(self.SIG_QUEUE) < 5:
        self.SIG_QUEUE.append(sig)
        self.wakeup()

class BaseSocket(object):

    def __init__(self, address, conf, log, fd=None):
        ...
        # 初始化sock
        self.sock = self.set_options(sock, bound=bound)

    def set_options(self, sock, bound=False):
        # 设置sock的功能
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        if (self.conf.reuse_port
            and hasattr(socket, 'SO_REUSEPORT')):  # pragma: no cover
            try:
                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
            except socket.error as err:
                if err.errno not in (errno.ENOPROTOOPT, errno.EINVAL):
                    raise
        if not bound:
            self.bind(sock)
        # 设置不阻塞
        sock.setblocking(0)

        # make sure that the socket can be inherited
        if hasattr(sock, "set_inheritable"):
            sock.set_inheritable(True)
        
        # 开始监听
        sock.listen(self.conf.backlog)
        return sock

    def bind(self, sock):
        sock.bind(self.cfg_addr)

def spawn_worker(self):
    # 每个新创建的Worker, age都加1, Worker越大，则代表越新
    self.worker_age += 1
    # 把一些参数传给worker
    worker = self.worker_class(self.worker_age, self.pid, self.LISTENERS,
                               self.app, self.timeout / 2.0,
                               self.cfg, self.log)
    self.cfg.pre_fork(self, worker)
    # 开始Fork
    pid = os.fork()
    if pid != 0:
        # 对于主进程，记录创建的worker
        worker.pid = pid
        self.WORKERS[pid] = worker
        return pid

    # 对于创建的子进程
    
    # 不继承其它worker的tmp文件
    for sibling in self.WORKERS.values():
        sibling.tmp.close()

    worker.pid = os.getpid()
    try:
        util._setproctitle("worker [%s]" % self.proc_name)
        self.log.info("Booting worker with pid: %s", worker.pid)
        self.cfg.post_fork(self, worker)
        # 开始运行，在init_process方法编写worker的方法，因为__init__方法属于在主进程运行
        worker.init_process()
        # exit 0 代表进程常退出
        sys.exit(0)
    except SystemExit:
        raise
    except AppImportError as e:
        # 初始化实例失败
        self.log.debug("Exception while loading the application",
                       exc_info=True)
        print("%s" % e, file=sys.stderr)
        sys.stderr.flush()
        sys.exit(self.APP_LOAD_ERROR)
    except Exception:
        # 其它失败方法
        self.log.exception("Exception in worker process")
        if not worker.booted:
            sys.exit(self.WORKER_BOOT_ERROR)
        sys.exit(-1)
    finally:
        self.log.info("Worker exiting (pid: %s)", worker.pid)
        try:
            # 不管如何，都要关闭临时文件
            worker.tmp.close()
            self.cfg.worker_exit(self, worker)
        except Exception:
            self.log.warning("Exception during worker exit:\n%s",
                             traceback.format_exc())

def spawn_workers(self):
    # 判断现在还需要多少个worker, 并创建
    for _ in range(self.num_workers - len(self.WORKERS)):
        self.spawn_worker()
        # 随机休息一段时间， 防止同时创建多个worker
        time.sleep(0.1 * random.random())

def init_process(self):
    # 初始化代码
    super().init_process()

while True:
    # 3.负责判断该进程是否是正真的master，如果是则提升为正真的master(这一块放在最后一部分进行分析)
    self.maybe_promote_master()

    # 4.获取信号
    sig = self.SIG_QUEUE.pop(0) if self.SIG_QUEUE else None
    if sig is None:
        # 5.利用select休眠1小时
        self.sleep()
        # 6.判断worker是否超时，如果是则杀掉worker(将在Worker章节进行分析) 
        self.murder_workers()
        self.manage_workers()
        continue

    if sig not in self.SIG_NAMES:
        self.log.info("Ignoring unknown signal: %s", sig)
        continue

    signame = self.SIG_NAMES.get(sig)
    # 调用对应的信号处理
    handler = getattr(self, "handle_%s" % signame, None)
    if not handler:
        self.log.error("Unhandled signal: %s", signame)
        continue
    self.log.info("Handling signal: %s", signame)
    handler()
    # 7.这样下次循环就不会等待一秒了
    self.wakeup()

def signal(self, sig, frame):
    if len(self.SIG_QUEUE) < 5:
        self.SIG_QUEUE.append(sig)
        self.wakeup()

def wakeup(self):
    try:
        os.write(self.PIPE[1], b'.')
    except IOError as e:
        if e.errno not in [errno.EAGAIN, errno.EINTR]:
            raise

def sleep(self):
    try:
        # 等待事件返回
        ready = select.select([self.PIPE[0]], [], [], 1.0)
        if not ready[0]:
            # 没有事件代表超时了
            return
        # 获取剩余的事件，不然下次的sleep是无效的
        while os.read(self.PIPE[0], 1):
            pass
    except (select.error, OSError) as e:
        # TODO: select.error is a subclass of OSError since Python 3.3.
        error_number = getattr(e, 'errno', e.args[0])
        # 如果是中断或者重试错误，则直接return
        if error_number not in [errno.EAGAIN, errno.EINTR]:
            raise
    except KeyboardInterrupt:
        sys.exit()

def handle_usr2(self):
    # 收到信号，执行对应的函数 
    self.reexec()

def reexec(self):
    # 防止多个信号同时处理
    if self.reexec_pid != 0:
        self.log.warning("USR2 signal ignored. Child exists.")
        return

    if self.master_pid != 0:
        self.log.warning("USR2 signal ignored. Parent exists.")
        return

    master_pid = os.getpid()
    self.reexec_pid = os.fork()
    if self.reexec_pid != 0:
        # 对于主进程，不用再执行对应的代码
        return

    # 对于fork出来的子进程，执行的代码逻辑
    self.cfg.pre_exec(self)

    # 复制当前Master进程初始化时的环境变量（不被后面运行时的变量影响）
    environ = self.cfg.env_orig.copy()
    # 设置当前Master进程的PID
    environ['GUNICORN_PID'] = str(master_pid)

    # 设置文件描述符地址到环境变量
    if self.systemd:
        environ['LISTEN_PID'] = str(os.getpid())
        environ['LISTEN_FDS'] = str(len(self.LISTENERS))
    else:
        environ['GUNICORN_FD'] = ','.join(
            str(l.fileno()) for l in self.LISTENERS)

    # 更改cwd
    os.chdir(self.START_CTX['cwd'])

    # exec the process using the original environment
    # Gunicorn通过os.execvpe来运行一个新的Master进程
    # execvpe会通过PATH环境变量来查找用户指定的程序，并用当前的Python运行环境来运行程序，同时使用的是指定的环境变量
    # 其中用到的参数分别是：
    #   0是当前的Python运行环境
    #   args是参数
    #   environ是环境变量
    os.execvpe(self.START_CTX[0], self.START_CTX['args'], environ)

def start(self):
    if 'GUNICORN_PID' in os.environ:
        # 代表这是一个被一个master进程创建的master进程
        self.master_pid = int(os.environ.get('GUNICORN_PID'))
        self.proc_name = self.proc_name + ".2"
        self.master_name = "Master.2"

    self.pid = os.getpid()
    if self.cfg.pidfile is not None:
        pidname = self.cfg.pidfile
        if self.master_pid != 0:
            pidname += ".2"
            # 重新启动时创建新的pidfile
        self.pidfile = Pidfile(pidname)
        self.pidfile.create(self.pid)

def start(self):
    ...
    # 省去之前的代码
    elif self.master_pid:
        # 获取之前进程的fd
        fds = []
        for fd in os.environ.pop('GUNICORN_FD').split(','):
            fds.append(int(fd))

    self.LISTENERS = sock.create_sockets(self.cfg, self.log, fds)

# 来自官方文档
  PID USER      PR  NI  VIRT  RES  SHR S  %CPU %MEM    TIME+  COMMAND
20844 benoitc   20   0 54808  11m 3352 S   0.0  0.1   0:00.36 gunicorn: master [test:app]
20849 benoitc   20   0 54808 9.9m 1500 S   0.0  0.1   0:00.02 gunicorn: worker [test:app]
20850 benoitc   20   0 54808 9.9m 1500 S   0.0  0.1   0:00.01 gunicorn: worker [test:app]
20851 benoitc   20   0 54808 9.9m 1500 S   0.0  0.1   0:00.01 gunicorn: worker [test:app]
20854 benoitc   20   0 55748  12m 3348 S   0.0  0.2   0:00.35 gunicorn: master [test:app]
20859 benoitc   20   0 55748  11m 1500 S   0.0  0.1   0:00.01 gunicorn: worker [test:app]
20860 benoitc   20   0 55748  11m 1500 S   0.0  0.1   0:00.00 gunicorn: worker [test:app]
20861 benoitc   20   0 55748  11m 1500 S   0.0  0.1   0:00.01 gunicorn: worker [test:app]

def maybe_promote_master(self):
    if self.master_pid == 0:
        # 已经是真正的`Master`进程，直接返回
        return

    if self.master_pid != os.getppid():
        # 该进程是从之前的主进程派生的，获取不到父进程id证明父进程被kill了  这时自己将晋升为主进程
        self.log.info("Master has been promoted.")
        # 设置属性，标明自己是正真的Master
        # reset master infos
        self.master_name = "Master"
        self.master_pid = 0
        self.proc_name = self.cfg.proc_name
        del os.environ['GUNICORN_PID']
        # rename the pidfile
        if self.pidfile is not None:
            self.pidfile.rename(self.cfg.pidfile)
        # reset proctitle
        util._setproctitle("master [%s]" % self.proc_name)

def handle_hup(self):
    self.log.info("Hang up: %s", self.master_name)
    self.reload()

def reload(self):
    ...
    # 省略其它代码

    # spawn new workers
    for _ in range(self.cfg.workers):
        self.spawn_worker()

    # manage workers
    self.manage_workers()

def handle_ttin(self):
    self.num_workers += 1
    self.manage_workers()

def handle_ttou(self):
    if self.num_workers <= 1:
        return
    self.num_workers -= 1
    self.manage_workers()

def reap_workers(self):
    try:
        # 注意信号不排队的问题，你可以循环检查waitpid()子进程的状态. 
        while True:
            # 获取僵死子进程的信息
            wpid, status = os.waitpid(-1, os.WNOHANG)
            if not wpid:
                break
            if self.reexec_pid == wpid:
                self.reexec_pid = 0
            else:
                # A worker was terminated. If the termination reason was
                # that it could not boot, we'll shut it down to avoid
                # infinite start/stop cycles.
                exitcode = status >> 8
                if exitcode == self.WORKER_BOOT_ERROR:
                    reason = "Worker failed to boot."
                    raise HaltServer(reason, self.WORKER_BOOT_ERROR)
                if exitcode == self.APP_LOAD_ERROR:
                    reason = "App failed to load."
                    raise HaltServer(reason, self.APP_LOAD_ERROR)
                # WIFSIGNALED(status)为非0表明进程异常终止，记录异常信息。 
                if os.WIFSIGNALED(status):
                    self.log.warning(
                        "Worker with pid %s was terminated due to signal %s",
                        wpid,
                        os.WTERMSIG(status)
                    )

                # 清除Master进程的数据
                worker = self.WORKERS.pop(wpid, None)
                if not worker:
                    continue
                # 关闭临时文件
                worker.tmp.close()
                self.cfg.child_exit(self, worker)
    except OSError as e:
        # errno.ECHILD代表调用进程没有任何子进程，该类型的错误不应该报错
        if e.errno != errno.ECHILD:
            raise

class WorkerTmp(object):

    def __init__(self, cfg):
        ...
        # 省略代码，这一步只是做初始化 
        self.spinner = 0

    def notify(self):
        # 通过更改权限的方式更新修改时间
        self.spinner = (self.spinner + 1) % 2
        os.fchmod(self._tmp.fileno(), self.spinner)

    def last_update(self):
        # 获取文件描述符的修改时间
        return os.fstat(self._tmp.fileno()).st_ctime

def murder_workers(self):
    if not self.timeout:
        return
    workers = list(self.WORKERS.items())
    for (pid, worker) in workers:
        try:
            # 通过临时文件的最后更新时间判断是否合法
            if time.time() - worker.tmp.last_update() <= self.timeout:
                continue
        except (OSError, ValueError):
            continue

        if not worker.aborted:
            # 第一次发现不合法，发送警告指令
            self.log.critical("WORKER TIMEOUT (pid:%s)", pid)
            worker.aborted = True
            # 发送ABRT信号
            self.kill_worker(pid, signal.SIGABRT)
        else:
            # 第二次发现不合法，关闭进程
            self.kill_worker(pid, signal.SIGKILL)