flask 源码专题（app.run()的背后） _源码

千磨万击还坚劲，任尔东西南北风。这篇文章主要讲述flask 源码专题：app.run()的背后相关的知识，希望能为你提供帮助。
当我们用Flask写好一个app后, 运行app.run()表示监听指定的端口, 对收到的request运行app生成response并返回. 现在分析一下, 运行app.run()后具体发生了什么事情
Flask定义的run方法如下:

def run(self, host=None, port=None, debug=None, **options): """Runs the application on a local development server.Do not use ``run()`` in a production setting. It is not intended to meet security and performance requirements for a production server. Instead, see :ref:`deployment` for WSGI server recommendations.If the :attr:`debug` flag is set the server will automatically reload for code changes and show a debugger in case an exception happened.If you want to run the application in debug mode, but disable the code execution on the interactive debugger, you can pass ``use_evalex=False`` as parameter.This will keep the debugger‘s traceback screen active, but disable code execution.It is not recommended to use this function for development with automatic reloading as this is badly supported.Instead you should be using the :command:`flask` command line script‘s ``run`` support... admonition:: Keep in MindFlask will suppress any server error with a generic error page unless it is in debug mode.As such to enable just the interactive debugger without the code reloading, you have to invoke :meth:`run` with ``debug=True`` and ``use_reloader=False``. Setting ``use_debugger`` to ``True`` without being in debug mode won‘t catch any exceptions because there won‘t be any to catch... versionchanged:: 0.10 The default port is now picked from the ``SERVER_NAME`` variable.:param host: the hostname to listen on. Set this to ``‘0.0.0.0‘`` to have the server available externally as well. Defaults to ``‘127.0.0.1‘``. :param port: the port of the webserver. Defaults to ``5000`` or the port defined in the ``SERVER_NAME`` config variable if present. :param debug: if given, enable or disable debug mode. See :attr:`debug`. :param options: the options to be forwarded to the underlying Werkzeug server.See :func:`werkzeug.serving.run_simple` for more information. """ from werkzeug.serving import run_simple if host is None: host = ‘127.0.0.1‘ if port is None: server_name = self.config[‘SERVER_NAME‘] if server_name and ‘:‘ in server_name: port = int(server_name.rsplit(‘:‘, 1)[1]) else: port = 5000 if debug is not None: self.debug = bool(debug) options.setdefault(‘use_reloader‘, self.debug) options.setdefault(‘use_debugger‘, self.debug) try: run_simple(host, port, self, **options) finally: # reset the first request information if the development server # reset normally.This makes it possible to restart the server # without reloader and that stuff from an interactive shell. self._got_first_request = False

在上面的源码中, from werkzeug.serving import run_simple引入run_simple函数, 该函数将定义好的app在指定的host, port上运行，run_simple函数如下：

def run_simple( hostname, port, application, use_reloader=False, use_debugger=False, use_evalex=True, extra_files=None, reloader_interval=1, reloader_type="auto", threaded=False, processes=1, request_handler=None, static_files=None, passthrough_errors=False, ssl_context=None, ): """ ..................... """ if not isinstance(port, int): raise TypeError("port must be an integer") if use_debugger: from .debug import DebuggedApplicationapplication = DebuggedApplication(application, use_evalex) if static_files: from .middleware.shared_data import SharedDataMiddlewareapplication = SharedDataMiddleware(application, static_files)def log_startup(sock): display_hostname = hostname if hostname not in ("", "*") else "localhost" quit_msg = "(Press CTRL+C to quit)" if sock.family == af_unix: _log("info", " * Running on %s %s", display_hostname, quit_msg) else: if ":" in display_hostname: display_hostname = "[%s]" % display_hostname port = sock.getsockname()[1] _log( "info", " * Running on %s://%s:%d/ %s", "http" if ssl_context is None else "https", display_hostname, port, quit_msg, )def inner(): try: fd = int(os.environ["WERKZEUG_SERVER_FD"]) except (LookupError, ValueError): fd = None srv = make_server( hostname, port, application, threaded, processes, request_handler, passthrough_errors, ssl_context, fd=fd, )#第二步
if fd is None: log_startup(srv.socket) srv.serve_forever() #第三步

if use_reloader: # If we‘re not running already in the subprocess that is the # reloader we want to open up a socket early to make sure the # port is actually available. if not is_running_from_reloader(): if port == 0 and not can_open_by_fd: raise ValueError( "Cannot bind to a random port with enabled " "reloader if the python interpreter does " "not support socket opening by fd." )# Create and destroy a socket so that any exceptions are # raised before we spawn a separate Python interpreter and # lose this ability. address_family = select_address_family(hostname, port) server_address = get_sockaddr(hostname, port, address_family) s = socket.socket(address_family, socket.SOCK_STREAM) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind(server_address) if hasattr(s, "set_inheritable"): s.set_inheritable(True)# If we can open the socket by file descriptor, then we can just # reuse this one and our socket will survive the restarts. if can_open_by_fd: os.environ["WERKZEUG_SERVER_FD"] = str(s.fileno()) s.listen(LISTEN_QUEUE) log_startup(s) else: s.close() if address_family == af_unix: _log("info", "Unlinking %s" % server_address) os.unlink(server_address)# Do not use relative imports, otherwise "python -m werkzeug.serving" # breaks. from ._reloader import run_with_reloaderrun_with_reloader(inner, extra_files, reloader_interval, reloader_type) else: inner() # 第一步

run_simple函数一般情况下调用内部定义的inner函数. inner调用make_server创建server实例srv, 然后调用srv.serve_forever.
server_forever函数如下:

def make_server( host=None, port=None, app=None, threaded=False, processes=1, request_handler=None, passthrough_errors=False, ssl_context=None, fd=None, ): """Create a new server instance that is either threaded, or forks or just processes one request after another. """ if threaded and processes > 1: raise ValueError("cannot have a multithreaded and multi process server.") elif threaded: return ThreadedWSGIServer( host, port, app, request_handler, passthrough_errors, ssl_context, fd=fd ) elif processes > 1: return ForkingWSGIServer( host, port, app, processes, request_handler, passthrough_errors, ssl_context, fd=fd, ) else: return BaseWSGIServer( host, port, app, request_handler, passthrough_errors, ssl_context, fd=fd )

【flask 源码专题（app.run()的背后）】

class BaseWSGIServer(HTTPServer, object):"""Simple single-threaded, single-process WSGI server."""multithread = False multiprocess = False request_queue_size = LISTEN_QUEUEdef __init__( self, host, port, app, handler=None, passthrough_errors=False, ssl_context=None, fd=None, ): if handler is None: handler = WSGIRequestHandler #第一步self.address_family = select_address_family(host, port)if fd is not None: real_sock = socket.fromfd(fd, self.address_family, socket.SOCK_STREAM) port = 0server_address = get_sockaddr(host, int(port), self.address_family)# remove socket file if it already exists if self.address_family == af_unix and os.path.exists(server_address): os.unlink(server_address) HTTPServer.__init__(self, server_address, handler)#第二步self.app = app self.passthrough_errors = passthrough_errors self.shutdown_signal = False self.host = host self.port = self.socket.getsockname()[1]# Patch in the original socket. if fd is not None: self.socket.close() self.socket = real_sock self.server_address = self.socket.getsockname()if ssl_context is not None: if isinstance(ssl_context, tuple): ssl_context = load_ssl_context(*ssl_context) if ssl_context == "adhoc": ssl_context = generate_adhoc_ssl_context()# If we are on Python 2 the return value from socket.fromfd # is an internal socket object but what we need for ssl wrap # is the wrapper around it :( sock = self.socket if PY2 and not isinstance(sock, socket.socket): sock = socket.socket(sock.family, sock.type, sock.proto, sock) self.socket = ssl_context.wrap_socket(sock, server_side=True) self.ssl_context = ssl_context else: self.ssl_context = Nonedef log(self, type, message, *args): _log(type, message, *args)def serve_forever(self):#第三步 self.shutdown_signal = False try: HTTPServer.serve_forever(self)#第四步
except KeyboardInterrupt: pass finally: self.server_close()

class BaseServer:"""Base class for server classes.Methods for the caller:- __init__(server_address, RequestHandlerClass) - serve_forever(poll_interval=0.5) - shutdown() - handle_request()# if you do not use serve_forever() - fileno() -> int# for selectorMethods that may be overridden:- server_bind() - server_activate() - get_request() -> request, client_address - handle_timeout() - verify_request(request, client_address) - server_close() - process_request(request, client_address) - shutdown_request(request) - close_request(request) - service_actions() - handle_error()Methods for derived classes:- finish_request(request, client_address)Class variables that may be overridden by derived classes or instances:- timeout - address_family - socket_type - allow_reuse_addressInstance variables:- RequestHandlerClass - socket"""timeout = Nonedef __init__(self, server_address, RequestHandlerClass): """Constructor.May be extended, do not override.""" self.server_address = server_address self.RequestHandlerClass = RequestHandlerClass self.__is_shut_down = threading.Event() self.__shutdown_request = Falsedef server_activate(self): """Called by constructor to activate the server.May be overridden.""" passdef serve_forever(self, poll_interval=0.5): """Handle one request at a time until shutdown.Polls for shutdown every poll_interval seconds. Ignores self.timeout. If you need to do periodic tasks, do them in another thread. """ self.__is_shut_down.clear() try: # XXX: Consider using another file descriptor or connecting to the # socket to wake this up instead of polling. Polling reduces our # responsiveness to a shutdown request and wastes cpu at all other # times. with _ServerSelector() as selector: selector.register(self, selectors.EVENT_READ)while not self.__shutdown_request: ready = selector.select(poll_interval) # bpo-35017: shutdown() called during select(), exit immediately. if self.__shutdown_request: break if ready: self._handle_request_noblock()#第一步self.service_actions() finally: self.__shutdown_request = False self.__is_shut_down.set()def shutdown(self): """Stops the serve_forever loop.Blocks until the loop has finished. This must be called while serve_forever() is running in another thread, or it will deadlock. """ self.__shutdown_request = True self.__is_shut_down.wait()def service_actions(self): """Called by the serve_forever() loop.May be overridden by a subclass / Mixin to implement any code that needs to be run during the loop. """ pass# The distinction between handling, getting, processing and finishing a # request is fairly arbitrary.Remember: # # - handle_request() is the top-level call.It calls selector.select(), #get_request(), verify_request() and process_request() # - get_request() is different for stream or datagram sockets # - process_request() is the place that may fork a new process or create a #new thread to finish the request # - finish_request() instantiates the request handler class; this #constructor will handle the request all by itselfdef handle_request(self): """Handle one request, possibly blocking.Respects self.timeout. """ # Support people who used socket.settimeout() to escape # handle_request before self.timeout was available. timeout = self.socket.gettimeout() if timeout is None: timeout = self.timeout elif self.timeout is not None: timeout = min(timeout, self.timeout) if timeout is not None: deadline = time() + timeout# Wait until a request arrives or the timeout expires - the loop is # necessary to accommodate early wakeups due to EINTR. with _ServerSelector() as selector: selector.register(self, selectors.EVENT_READ)while True: ready = selector.select(timeout) if ready: return self._handle_request_noblock() else: if timeout is not None: timeout = deadline - time() if timeout < 0: return self.handle_timeout()def _handle_request_noblock(self): """Handle one request, without blocking.I assume that selector.select() has returned that the socket is readable before this function was called, so there should be no risk of blocking in get_request(). """ try: request, client_address = self.get_request() except OSError: return if self.verify_request(request, client_address): try: self.process_request(request, client_address)#第二步 except Exception: self.handle_error(request, client_address) self.shutdown_request(request) except: self.shutdown_request(request) raise else: self.shutdown_request(request)def handle_timeout(self): """Called if no new request arrives within self.timeout.Overridden by ForkingMixIn. """ passdef verify_request(self, request, client_address): """Verify the request.May be overridden.Return True if we should proceed with this request.""" return Truedef process_request(self, request, client_address): """Call finish_request.Overridden by ForkingMixIn and ThreadingMixIn.""" self.finish_request(request, client_address)#第三步 self.shutdown_request(request)def finish_request(self, request, client_address): """Finish one request by instantiating RequestHandlerClass.""" self.RequestHandlerClass(request, client_address, self)#第四步