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。

WSGI及gunicorn指北（二）