WSGI及gunicorn指北

Posted 2020-11-03 yuwhuawang

pyg0已经大概了解了wsgi。现在他决定深入探索他们实际在生产环境里用到的web 服务器 -gunicorn。

先来看看官网的介绍：Gunicorn 是一个运行在Unix上的python WSGI web 服务器。它采用Pre-Fork 的worker模型 。Gunicorn
可以适用于多种python web 框架，实现简单，占用系用资源少，速度快。

抛开那些自卖自夸的话语，我们一项一项来分析。

Gunicorn 之所以只能运行在Unix系统上，是因为gunicorn使用了只有在*nix系统中才存在的模块和方法，例如fnctl，os.fork等等。

而Pre-Fork的worker模型是指在gunicorn启动时，会通过Master(在gunicorn里叫做arbiter)预先生成特定数量的worker， 这些worker通过arbiter管理。

当我们启动一个wsgiapp， 会初始化一个Application，在他的基类BaseApplication中有一个run方法，代码如下(节选)：

class BaseApplication(object):

    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)

主要是把任务交给Arbiter这个类来完成。我们看看这个Arbiter是如果启动和管理worker的。

class Arbiter(object):
    """
    Arbiter maintain the workers processes alive. It launches or
    kills them if needed. It also manages application reloading
    via SIGHUP/USR2.
    """
def start(self):
        """        Initialize the arbiter. Start listening and set pidfile if needed.
        """
        self.log.info("Starting gunicorn %s", __version__)

        if ‘GUNICORN_PID‘ in os.environ:
            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"
            self.pidfile = Pidfile(pidname)
            self.pidfile.create(self.pid)
        self.cfg.on_starting(self)

        self.init_signals()

        if not self.LISTENERS:
            fds = None
            listen_fds = systemd.listen_fds()
            if listen_fds:
                self.systemd = True
                fds = range(systemd.SD_LISTEN_FDS_START,
                            systemd.SD_LISTEN_FDS_START + listen_fds)

            elif self.master_pid:
                fds = []
                for fd in os.environ.pop(‘GUNICORN_FD‘).split(‘,‘):
                    fds.append(int(fd))

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

        listeners_str = ",".join([str(l) for l in self.LISTENERS])
        self.log.debug("Arbiter booted")
        self.log.info("Listening at: %s (%s)", listeners_str, self.pid)
        self.log.info("Using worker: %s", self.cfg.worker_class_str)

        # check worker class requirements
        if hasattr(self.worker_class, "check_config"):
            self.worker_class.check_config(self.cfg, self.log)

        self.cfg.when_ready(self)def run(self):
        "Main master loop."
        self.start()
        util._setproctitle("master [%s]" % self.proc_name)

        try:
            self.manage_workers()

            while True:
                self.maybe_promote_master()

                sig = self.SIG_QUEUE.pop(0) if self.SIG_QUEUE else None
                if sig is None:
                    self.sleep()
                    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()
                self.wakeup()
        except StopIteration:
            self.halt()
        except KeyboardInterrupt:
            self.halt()
        except HaltServer as inst:
            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 manage_workers(self):
        """        Maintain the number of workers by spawning or killing
        as required.
        """
        if len(self.WORKERS.keys()) < self.num_workers:
            self.spawn_workers()

        workers = self.WORKERS.items()
        workers = sorted(workers, key=lambda w: w[1].age)
        while len(workers) > self.num_workers:
            (pid, _) = workers.pop(0)
            self.kill_worker(pid, signal.SIGTERM)

        active_worker_count = len(workers)
        if self._last_logged_active_worker_count != active_worker_count:
            self._last_logged_active_worker_count = active_worker_count
            self.log.debug("{0} workers".format(active_worker_count),
                           extra={"metric": "gunicorn.workers",
                                  "value": active_worker_count,
                                  "mtype": "gauge"})

    def spawn_worker(self):
        self.worker_age += 1
        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)
        pid = os.fork()
        if pid != 0:
            worker.pid = pid
            self.WORKERS[pid] = worker
            return pid

        # Do not inherit the temporary files of other workers
        for sibling in self.WORKERS.values():
            sibling.tmp.close()

        # Process Child
        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)
            worker.init_process()
            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:
            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:
                self.log.warning("Exception during worker exit:\n%s",
                                  traceback.format_exc())

Arbiter 初始化时， 首先要读取配置，比如worker的数量，worker的模式，然后创建socket，创建worker并且进入循环管理worker信号，如果worker不响应了，就干掉创建一个新的。主进程Arbiter大概就是这么简单。

在Arbiter创建worker的方法里，通过worker.init_process()进入worker的消息循环。

现在再来看一下Worker进程。Gunicorn支持多种worker模式，默认的为sync，就像名字表达的一样，这是个同步阻塞的网络模型。除了sync之外，还有ggevent，gaiohttp,gthread等等。我们主要来看一下ggevent模式的worker。

ggevent 模式的worker基于gevent异步协程库，支持异步I/O模式。

我们来看代码（节选）

class GeventWorker(AsyncWorker):
        def run(self):
        servers = []
        ssl_args = {}

        if self.cfg.is_ssl:
            ssl_args = dict(server_side=True, **self.cfg.ssl_options)

        for s in self.sockets:
            s.setblocking(1)
            pool = Pool(self.worker_connections)
            if self.server_class is not None:
                environ = base_environ(self.cfg)
                environ.update({
                    "wsgi.multithread": True,
                    "SERVER_SOFTWARE": VERSION,
                })
                server = self.server_class(
                    s, application=self.wsgi, spawn=pool, log=self.log,
                    handler_class=self.wsgi_handler, environ=environ,
                    **ssl_args)
            else:
                hfun = partial(self.handle, s)
                server = StreamServer(s, handle=hfun, spawn=pool, **ssl_args)

            server.start()
            servers.append(server)

        while self.alive:
            self.notify()
            gevent.sleep(1.0)

        try:
            # Stop accepting requests
            for server in servers:
                if hasattr(server, ‘close‘):  # gevent 1.0
                    server.close()
                if hasattr(server, ‘kill‘):  # gevent < 1.0
                    server.kill()

            # Handle current requests until graceful_timeout
            ts = time.time()
            while time.time() - ts <= self.cfg.graceful_timeout:
                accepting = 0
                for server in servers:
                    if server.pool.free_count() != server.pool.size:
                        accepting += 1

                # if no server is accepting a connection, we can exit
                if not accepting:
                    return

                self.notify()
                gevent.sleep(1.0)

            # Force kill all active the handlers
            self.log.warning("Worker graceful timeout (pid:%s)" % self.pid)
            for server in servers:
                server.stop(timeout=1)
        except:
            pass

在worker启动的时候，针对每一个Ambiter 初始化的socket，创建一个server，每一个server都运行在一个gevent pool里，等待和处理连接的操作都是在server里面进行的。

这里的server_class是和wsgi_handler 就是gevent的WSGIserver和WSGIHander的子类

class GeventPyWSGIWorker(GeventWorker):
    "The Gevent StreamServer based workers."
    server_class = PyWSGIServer
    wsgi_handler = PyWSGIHandler

我们看一下server创建的过程

 server = self.server_class(s, application=self.wsgi, spawn=pool, log=self.log,
                    handler_class=self.wsgi_handler, environ=environ,
                    **ssl_args)

s就是Arbiter创建的socket, application 就是我们用的WSGI app（比如django，flask等框架的app）， spawn是gevent的协程池。handler_class就是gevent的wsgi_handler的子类。通过这种方式，把WSGI app交给worker来运行。

while self.alive:
    self.notify()
    gevent.sleep(1.0)

这里每隔一秒给Arbiter发一个通知，告诉arbiter worker还活着，避免被arbiter干掉。

 

真正等待和处理链接走到了gevent里的WSGIServer 和WSGIhandler。我们进入gevent具体看下：

class BaseServer(object):
    def start_accepting(self):
        if self._watcher is None:
            # just stop watcher without creating a new one?
            self._watcher = self.loop.io(self.socket.fileno(), 1)
            self._watcher.start(self._do_read)

WSGIServer用start_accepting来接收链接， 在读到socket里的信息后，用do_handle来处理链接

 def do_handle(self, *args):
        spawn = self._spawn
        handle = self._handle
        close = self.do_close

        try:
            if spawn is None:
                _handle_and_close_when_done(handle, close, args)
            else:
                spawn(_handle_and_close_when_done, handle, close, args)
        except:
            close(*args)
            raise

WSGIServer实际上是对每一个链接启动一个协程。在协程中，用self._handle, 使用WSGIHandler来处理连接的请求。我们找到WSGIHandler

class WSGIHandler(object):
    def handle_one_response(self):
        self.time_start = time.time()
        self.status = None
        self.headers_sent = False

        self.result = None
        self.response_use_chunked = False
        self.response_length = 0

        try:
            try:
                self.run_application()
            finally:
                try:
                    self.wsgi_input._discard()
                except (socket.error, IOError):
                    # Don‘t let exceptions during discarding
                    # input override any exception that may have been
                    # raised by the application, such as our own _InvalidClientInput.
                    # In the general case, these aren‘t even worth logging (see the comment
                    # just below)
                    pass
        except _InvalidClientInput:
            self._send_error_response_if_possible(400)
        except socket.error as ex:
            if ex.args[0] in (errno.EPIPE, errno.ECONNRESET):
                # Broken pipe, connection reset by peer.
                # Swallow these silently to avoid spewing
                # useless info on normal operating conditions,
                # bloating logfiles. See https://github.com/gevent/gevent/pull/377
                # and https://github.com/gevent/gevent/issues/136.
                if not PY3:
                    sys.exc_clear()
                self.close_connection = True
            else:
                self.handle_error(*sys.exc_info())
        except: # pylint:disable=bare-except
            self.handle_error(*sys.exc_info())
        finally:
            self.time_finish = time.time()
            self.log_request()

    def run_application(self):
        assert self.result is None
        try:
            self.result = self.application(self.environ, self.start_response)
            self.process_result()
        finally:
            close = getattr(self.result, ‘close‘, None)
            try:
                if close is not None:
                    close()
            finally:
                # Discard the result. If it‘s a generator this can
                # free a lot of hidden resources (if we failed to iterate
                # all the way through it---the frames are automatically
                # cleaned up when StopIteration is raised); but other cases
                # could still free up resources sooner than otherwise.
                close = None
                self.result = None

在WSGIHandler里，读取到request之后，调用handle_one_response来处理，其中run_application里，我们又看到了熟悉的

    self.result = self.application(self.environ, self.start_response)

WSGI app的标准调用。

 

pyg0看了这么一大圈，然后自己默默地做了一个总结：Gunicorn在启动的时候，使用Arbiter来预先创建相应数量和模式的worker，所有的worker都监听同一组socket。在每个worker里，创建server来监听和处理请求。而ggevent模式的worker里，每一个链接来的时候，worker会起一个协程来处理，通过协程的方式来实现异步I/O。而具体处理请求的，是用户提供的WSGI app。