WSGI及gunicorn指北（二）

pyg0已經大概了解了wsgi。現在他決定深入探索他們實際在生產環境里用到的web 伺服器 -gunicorn。

先來看看官網的介紹：Gunicorn 是一個運行在Unix上的python WSGI web 伺服器。它採用Pre-Fork 的worker模型 。Gunicorn

拋開那些自賣自誇的話語，我們一項一項來分析。

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("
Error: %s
" % 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:
%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): # Dont 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 arent 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 its 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。