diff --git a/CHANGES.txt b/CHANGES.txt index c7f32ea4..3629e626 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,9 +1,64 @@ Unreleased ---------- +Features +~~~~~~~~ + +- Waitress now shuts down gracefully. Upon being interrupted it closes its + listening sockets, stops reading new requests off the connections that are + still open, and keeps running the main loop until the requests that are + already being serviced have finished and their responses have been written + back to the client. Only then are the worker threads stopped and the + remaining connections closed. + + Previously the main loop was stopped immediately, which meant that any + response that had not been fully written out yet was dropped, and that a + task thread that was waiting on the main loop to drain an output buffer was + left blocked forever. + + This is also exposed as ``server.graceful_shutdown()`` for those that drive + the server themselves instead of using ``server.run()``. See + https://github.com/Pylons/waitress/issues/198 and + https://github.com/Pylons/waitress/issues/264 + +- Servers now have public ``start()`` and ``stop()`` methods that run the main + loop in a background thread and shut it down again. ``stop()`` is safe to + call from any thread, including from within the WSGI application. This + removes the need for the kind of workaround that WebTest's + ``StopableWSGIServer`` had to resort to, and combined with + ``server.effective_port`` it allows binding to port ``0`` and reading back + the port that was picked without racing anybody else for it. See + https://github.com/Pylons/waitress/issues/290 + +- Added a new ``shutdown_timeout`` adjustment (default: 5 seconds) that bounds + how long the graceful shutdown described above may take, and that replaces + the previously hardcoded timeout used when stopping the worker threads. Set + it to ``0`` to tear everything down immediately instead. See + https://github.com/Pylons/waitress/issues/134 + Bugfix ~~~~~~ +- Waitress no longer leaks threads, sockets and socket map entries when it + fails to create a server, for example because the port it was asked to + listen on is already in use. The worker threads are no longer started until + every listening socket has been successfully bound, and anything that was + created before the failure is now cleaned up rather than being left running + with no way for the caller to get at it. See + https://github.com/Pylons/waitress/issues/480 + +- ``BaseWSGIServer.close()`` now shuts down the task dispatcher and closes the + connections that are still open, matching what ``MultiSocketServer.close()`` + already did. Previously closing a server that was listening on a single + socket left its worker threads running. See + https://github.com/Pylons/waitress/issues/402 and + https://github.com/Pylons/waitress/issues/480 + +- Pulling the trigger after it has been closed is now a no-op instead of an + error. A task thread that finished right as the server was going away could + previously write to a file descriptor that had already been closed, and + potentially handed out to something else in the meantime. + - Renamed the HTTP header "Trailers" to "Trailer" to fix a typo and comply with the correct header name as specified in RFC 7230. diff --git a/docs/arguments.rst b/docs/arguments.rst index b8a856aa..c5fe6b62 100644 --- a/docs/arguments.rst +++ b/docs/arguments.rst @@ -270,6 +270,19 @@ channel_timeout Default: ``120`` +shutdown_timeout + Maximum seconds to spend shutting down gracefully (integer). Once waitress + is asked to stop it stops accepting new connections and keeps running its + main loop for at most this long, so that requests that are already being + serviced get a chance to finish and have their response written back to the + client. Connections that are still open once this expires are closed, and + any tasks that are still queued are cancelled. Set to ``0`` to tear + everything down immediately instead. + + Default: ``5`` + + .. versionadded:: 3.1.0 + log_socket_errors Set to ``False`` to not log premature client disconnect tracebacks. diff --git a/docs/runner.rst b/docs/runner.rst index d128d969..16a6a5f2 100644 --- a/docs/runner.rst +++ b/docs/runner.rst @@ -264,6 +264,14 @@ Tuning options: 120. 'Inactive' is defined as 'has received no data from the client and has sent no data to the client'. +``--shutdown-timeout=INT`` + Maximum number of seconds to spend shutting down gracefully. Default is 5. + Once waitress is asked to stop it stops accepting new connections and keeps + running its main loop for at most this long, so that requests that are + already being serviced get a chance to finish and have their response + written back to the client. Set to ``0`` to tear everything down + immediately instead. + ``--channel-request-lookahead=INT`` Sets the amount of requests we can continue to read from the socket, while we are processing current requests. The default value won't allow any diff --git a/docs/usage.rst b/docs/usage.rst index 01ff2878..060d696b 100644 --- a/docs/usage.rst +++ b/docs/usage.rst @@ -80,6 +80,88 @@ can be used in development and in situations where the likes of # Listen on only IPv4 on port 8041 waitress-serve --port=8041 myapp:wsgifunc +.. _shutdown: + +Shutting down +------------- + +When ``server.run()`` is interrupted, either by :kbd:`Ctrl-C` or by a +``SystemExit``, Waitress shuts down gracefully: + +- it closes its listening sockets, so the port is released and no new + connections are accepted; +- it stops reading new requests off the connections that are still open; +- it keeps running its main loop so that requests that are already being + serviced can finish and have their response written back to the client; +- once every connection has been dealt with, it stops the worker threads and + closes up shop. + +Requests that have not finished within ``shutdown_timeout`` seconds (5 by +default) do not get to hold up the shutdown any further, and their connections +are closed. Interrupting a second time skips the wait entirely. See +:ref:`arguments` for ``shutdown_timeout``. + +Waitress does not install any signal handlers of its own. To shut down on a +``SIGTERM`` from a process manager, install a handler that raises +``SystemExit`` in the main thread and let ``run()`` take it from there: + +.. code-block:: python + + import signal + import sys + + from waitress.server import create_server + + server = create_server(app, listen="*:8080") + signal.signal(signal.SIGTERM, lambda *args: sys.exit(0)) + server.run() + +The shutdown is also available directly as ``server.graceful_shutdown()``, for +when you drive the main loop yourself rather than through ``run()``. Note that +it runs the main loop while it drains, so it has to be called from the same +thread the loop runs on. + +``server.close()`` is the non-graceful counterpart: it stops the worker threads +and closes every connection immediately, without giving the requests that are +in flight a chance to finish. + +.. _running-in-a-thread: + +Running in a background thread +------------------------------ + +``server.run()`` blocks. To run a server alongside something else, for example +to serve an application under test, use ``start()`` and ``stop()`` instead: + +.. code-block:: python + + from waitress.server import create_server + + server = create_server(app, host="127.0.0.1", port=0) + server.start() + try: + ... + finally: + server.stop() + +``start()`` runs the main loop in a daemon thread and returns immediately. +``stop()`` shuts the server down gracefully and waits for it to finish, and is +the one method that may be called from any thread, including from within the +WSGI application itself. + +Because the server is already listening by the time ``start()`` returns, asking +for port ``0`` and then reading back the port that the operating system picked +is race free — there is no window in which another process could take it: + +.. code-block:: python + + server = create_server(app, host="127.0.0.1", port=0) + server.start() + url = f"http://{server.effective_host}:{server.effective_port}/" + +With several listening sockets, ``server.effective_listen`` holds the full list +of ``(host, port)`` pairs instead. + Heroku ------ diff --git a/pyproject.toml b/pyproject.toml index 32abfcb5..8f8234fb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "waitress" -version = "3.0.2" +version = "3.1.0.dev0" authors = [{ name = "Zope Foundation and Contributors", email = "zope-dev@zope.org" }] maintainers = [{ name = "Pylons Project", email = "pylons-discuss@googlegroups.com" }] description = "Waitress WSGI server" diff --git a/src/waitress/adjustments.py b/src/waitress/adjustments.py index 8b26eb87..ad8f2271 100644 --- a/src/waitress/adjustments.py +++ b/src/waitress/adjustments.py @@ -138,6 +138,7 @@ class Adjustments: ("connection_limit", int), ("cleanup_interval", int), ("channel_timeout", int), + ("shutdown_timeout", int), ("log_socket_errors", asbool), ("max_request_header_size", int), ("max_request_body_size", int), @@ -253,6 +254,14 @@ class Adjustments: # Maximum seconds to leave an inactive connection open. channel_timeout = 120 + # Maximum seconds to spend shutting down gracefully. Once the server is + # asked to stop it no longer accepts new connections, and keeps running the + # main loop for at most this long so that requests that are already being + # serviced get a chance to finish and have their response written back to + # the client. Set to 0 to disable the graceful shutdown entirely and tear + # everything down immediately. + shutdown_timeout = 5 + # Boolean: turn off to not log premature client disconnects. log_socket_errors = True diff --git a/src/waitress/channel.py b/src/waitress/channel.py index bcd18e0d..0ac47152 100644 --- a/src/waitress/channel.py +++ b/src/waitress/channel.py @@ -48,6 +48,7 @@ class HTTPChannel(wasyncore.dispatcher): sent_continue = False # used as a latch after sending 100 continue total_outbufs_len = 0 # total bytes ready to send current_outbuf_count = 0 # total bytes written to current outbuf + draining = False # set to True to stop reading new requests off the socket # # ASYNCHRONOUS METHODS (including __init__) @@ -142,13 +143,17 @@ def readable(self): # 1. We're not already about to close the connection. # 2. We're not waiting to flush remaining data before closing the # connection - # 3. There are not too many tasks already queued (if lookahead is enabled) - # 4. There's no data in the output buffer that needs to be sent + # 3. We're not draining the channel for a graceful shutdown, during + # which we finish the requests we already have but don't take on + # any new ones. + # 4. There are not too many tasks already queued (if lookahead is enabled) + # 5. There's no data in the output buffer that needs to be sent # before we potentially create a new task. return not ( self.will_close or self.close_when_flushed + or self.draining or len(self.requests) > self.adj.channel_request_lookahead or self.total_outbufs_len ) diff --git a/src/waitress/runner.py b/src/waitress/runner.py index 69af8460..a72266cf 100644 --- a/src/waitress/runner.py +++ b/src/waitress/runner.py @@ -232,6 +232,14 @@ Default is 120. 'Inactive' is defined as 'has received no data from the client and has sent no data to the client'. + --shutdown-timeout=INT + Maximum number of seconds to spend shutting down gracefully. + Default is 5. Once waitress is asked to stop it stops accepting + new connections and keeps running its main loop for at most this + long, so that requests that are already being serviced get a + chance to finish and have their response written back to the + client. Set to '0' to tear everything down immediately instead. + --channel-request-lookahead=INT Sets the amount of requests we can continue to read from the socket, while we are processing current requests. The default value won't allow diff --git a/src/waitress/server.py b/src/waitress/server.py index c56530a0..699c48d9 100644 --- a/src/waitress/server.py +++ b/src/waitress/server.py @@ -15,14 +15,15 @@ import os import os.path import socket +import threading import time from waitress import trigger from waitress.adjustments import Adjustments from waitress.channel import HTTPChannel from waitress.compat import IPPROTO_IPV6, IPV6_V6ONLY -from waitress.task import ThreadedTaskDispatcher -from waitress.utilities import cleanup_unix_socket +from waitress.task import ThreadedTaskDispatcher, in_task_thread +from waitress.utilities import cleanup_unix_socket, logger from . import wasyncore from .proxy_headers import proxy_headers_middleware @@ -54,90 +55,284 @@ def create_server( dispatcher = _dispatcher if dispatcher is None: dispatcher = ThreadedTaskDispatcher() - dispatcher.set_thread_count(adj.threads) - if adj.unix_socket and hasattr(socket, "AF_UNIX"): - sockinfo = (socket.AF_UNIX, socket.SOCK_STREAM, None, None) - return UnixWSGIServer( - application, - map, - _start, - _sock, - dispatcher=dispatcher, - adj=adj, - sockinfo=sockinfo, - ) - - effective_listen = [] - last_serv = None - if not adj.sockets: - for sockinfo in adj.listen: - # When TcpWSGIServer is called, it registers itself in the map. This - # side-effect is all we need it for, so we don't store a reference to - # or return it to the user. - last_serv = TcpWSGIServer( - application, - map, - _start, - _sock, - dispatcher=dispatcher, - adj=adj, - sockinfo=sockinfo, - ) - effective_listen.append( - (last_serv.effective_host, last_serv.effective_port) - ) - - for sock in adj.sockets: - sockinfo = (sock.family, sock.type, sock.proto, sock.getsockname()) - if sock.family == socket.AF_INET or sock.family == socket.AF_INET6: - last_serv = TcpWSGIServer( - application, - map, - _start, - sock, - dispatcher=dispatcher, - adj=adj, - bind_socket=False, - sockinfo=sockinfo, - ) - effective_listen.append( - (last_serv.effective_host, last_serv.effective_port) - ) - elif hasattr(socket, "AF_UNIX") and sock.family == socket.AF_UNIX: - last_serv = UnixWSGIServer( - application, - map, - _start, - sock, - dispatcher=dispatcher, - adj=adj, - bind_socket=False, - sockinfo=sockinfo, + # Everything below here may fail: a socket may not be bindable, a Unix + # socket path may not be writable, etc. Whatever we managed to create + # before that happened needs to be cleaned up again, otherwise we leave + # behind open sockets and running threads that the caller has no way of + # getting a reference to. See + # https://github.com/Pylons/waitress/issues/480 + servers = [] + + try: + if adj.unix_socket and hasattr(socket, "AF_UNIX"): + sockinfo = (socket.AF_UNIX, socket.SOCK_STREAM, None, None) + servers.append( + UnixWSGIServer( + application, + map, + _start, + _sock, + dispatcher=dispatcher, + adj=adj, + sockinfo=sockinfo, + ) ) - effective_listen.append( - (last_serv.effective_host, last_serv.effective_port) + else: + if not adj.sockets: + for sockinfo in adj.listen: + # When TcpWSGIServer is called, it registers itself in the + # map. This side-effect is all we need it for, so we don't + # return it to the user, we only hold on to it so that we + # are able to clean it up if a later server fails to start. + servers.append( + TcpWSGIServer( + application, + map, + _start, + _sock, + dispatcher=dispatcher, + adj=adj, + sockinfo=sockinfo, + ) + ) + + for sock in adj.sockets: + sockinfo = (sock.family, sock.type, sock.proto, sock.getsockname()) + if sock.family == socket.AF_INET or sock.family == socket.AF_INET6: + servers.append( + TcpWSGIServer( + application, + map, + _start, + sock, + dispatcher=dispatcher, + adj=adj, + bind_socket=False, + sockinfo=sockinfo, + ) + ) + elif hasattr(socket, "AF_UNIX") and sock.family == socket.AF_UNIX: + servers.append( + UnixWSGIServer( + application, + map, + _start, + sock, + dispatcher=dispatcher, + adj=adj, + bind_socket=False, + sockinfo=sockinfo, + ) + ) + + if not servers: + raise ValueError( + "There are no sockets to listen on, both 'listen' and 'sockets' " + "are empty." ) - # We are running a single server, so we can just return the last server, - # saves us from having to create one more object - if len(effective_listen) == 1: - # In this case we have no need to use a MultiSocketServer - return last_serv - - log_info = last_serv.log_info + effective_listen = [(s.effective_host, s.effective_port) for s in servers] + + if _dispatcher is None: + # Only start the worker threads once we know every socket is + # usable, so that a failure above never leaves threads behind. + dispatcher.set_thread_count(adj.threads) + except BaseException: + for server in servers: + server.close() + dispatcher.shutdown(timeout=adj.shutdown_timeout) + raise + + # We are running a single server, so we can just return it, this saves us + # from having to create one more object + if len(servers) == 1: + # In this case we have no need to use a MultiSocketServer. Hand the + # task dispatcher over to the server we return, it is the only handle + # the caller has on it and closing it needs to stop the threads. + servers[0].own_task_dispatcher = True + + return servers[0] + + log_info = servers[0].log_info # Return a class that has a utility function to print out the sockets it's # listening on, and has a .run() function. All of the TcpWSGIServers # registered themselves in the map above. - return MultiSocketServer(map, adj, effective_listen, dispatcher, log_info) + return MultiSocketServer( + map, adj, effective_listen, dispatcher, log_info, servers=servers + ) + + +def _drain(servers, map, adj, asyncore=wasyncore): + """ + Stop accepting new connections and then run the main loop until every + connection that is still open has been dealt with. + + The task threads hand their output back to the main loop, and rely on the + trigger to wake it up when they do. That means we can't stop the loop or + tear the trigger down the moment we are asked to shut down: doing so drops + the responses of everything that was in flight on the floor. Instead we + keep polling until the channels have written out what they had left and + closed themselves, giving up after ``adj.shutdown_timeout`` seconds. + """ + + # Close the listening sockets first. This releases the port right away and + # makes sure no new connections show up while we are draining the ones we + # already have. Everything else, in particular the trigger, is left alone. + for server in servers: + server.stop_accepting() + + if adj.shutdown_timeout <= 0: + return + + deadline = time.time() + adj.shutdown_timeout + + try: + while True: + channels = {} + + for server in servers: + channels.update(server.active_channels) + + if not channels: + break + + now = time.time() + + if now >= deadline: + logger.warning( + "Graceful shutdown timed out with %d connection(s) still " + "open, closing them now", + len(channels), + ) + + break + + for channel in channels.values(): + # Stop taking on any new work, we only want to finish what we + # already have. + channel.draining = True + + # A channel that is not waiting on a request to be serviced has + # nothing left to do for us, so let it flush whatever output it + # still has queued up and then close. Channels that do have + # requests outstanding are left running until their task thread + # is done with them, at which point they end up here too. + with channel.requests_lock: + if not channel.requests: + channel.close_when_flushed = True + + asyncore.loop( + timeout=min(adj.asyncore_loop_timeout, deadline - now), + map=map, + use_poll=adj.asyncore_use_poll, + count=1, + ) + except (SystemExit, KeyboardInterrupt): + # Interrupted a second time, the user is not interested in waiting for + # a clean shutdown anymore. + logger.warning("Graceful shutdown interrupted, closing connections now") + + +class _ServerRunner: + """ + Running the main loop, and stopping it again, shared by both flavours of + server. Subclasses provide ``_map``, ``adj``, ``pull_trigger()`` and + ``graceful_shutdown()``. + """ + + asyncore = wasyncore # test shim + + # Set by stop() to ask the main loop to return. Only ever written from + # outside the thread running the loop, hence the flag rather than a lock. + _stopping = False + + # The thread start() is running the main loop in, if there is one. + _thread = None + + def run(self): + """ + Run the main loop until the server is asked to stop, then shut it down + gracefully. This blocks until the shutdown has completed. + + The server stops when :meth:`stop` is called, when the process is + interrupted, or when a ``SystemExit`` is raised in this thread. + """ + + try: + while self._map and not self._stopping: + self.asyncore.loop( + timeout=self.adj.asyncore_loop_timeout, + map=self._map, + use_poll=self.adj.asyncore_use_poll, + count=1, + ) + except (SystemExit, KeyboardInterrupt): + pass + + self.graceful_shutdown() + + def start(self): + """ + Run the main loop in a background thread and return straight away. + + The server is listening by the time this returns, so + ``effective_host``/``effective_port`` can be used to find out which + address it ended up on when asking for port ``0``. + + The thread is a daemon thread, so it won't keep the process alive by + itself. Call :meth:`stop` to shut the server down again. + """ + + if self._thread is not None: + raise RuntimeError("This server has already been started") + + self._thread = threading.Thread( + target=self.run, name="waitress-main", daemon=True + ) + self._thread.start() + + def stop(self, timeout=None): + """ + Ask the server to shut down gracefully, see :meth:`graceful_shutdown`. + + Unlike the other shutdown methods this one may be called from any + thread, including from within the WSGI application itself: it only + signals the main loop, which then does the work. + + If the server was started with :meth:`start`, this waits up to + ``timeout`` seconds (forever by default) for it to finish. Returns + ``True`` if the server has completely stopped by the time it returns, + and ``False`` if it is still busy finishing up. + + It never waits when called from a thread the shutdown itself has to + wait for, such as from the WSGI application: doing so would deadlock. + In that case it returns ``False`` and the server stops as soon as the + request that called it has been answered. + """ + + self._stopping = True + # Wake the main loop up so that it notices, rather than having to wait + # for its select() to time out. + self.pull_trigger() + + thread, self._thread = self._thread, None + + if thread is None or thread is threading.current_thread() or in_task_thread(): + # Either nobody is running the loop for us, or we are being called + # from a thread the shutdown is going to wait on, which will get + # around to it once it is back in control. + return False + + thread.join(timeout) + + return not thread.is_alive() # This class is only ever used if we have multiple listen sockets. It allows # the serve() API to call .run() which starts the wasyncore loop, and catches # SystemExit/KeyboardInterrupt so that it can attempt to cleanly shut down. -class MultiSocketServer: - asyncore = wasyncore # test shim - +class MultiSocketServer(_ServerRunner): def __init__( self, map=None, @@ -145,6 +340,7 @@ def __init__( effective_listen=None, dispatcher=None, log_info=None, + servers=None, ): self.adj = adj self.map = map @@ -152,6 +348,19 @@ def __init__( self.task_dispatcher = dispatcher self.log_info = log_info + if servers is None: + # Not passed the listening sockets we are in charge of, so pick + # them back out of the socket map. + servers = [s for s in map.values() if isinstance(s, BaseWSGIServer)] + + self.servers = servers + + @property + def _map(self): + # The socket map is called `map` here for backwards compatibility, but + # _ServerRunner shares its name with wasyncore.dispatcher's. + return self.map + def print_listen(self, format_str): # pragma: nocover for l in self.effective_listen: l = list(l) @@ -161,27 +370,33 @@ def print_listen(self, format_str): # pragma: nocover self.log_info(format_str.format(*l)) - def run(self): - try: - self.asyncore.loop( - timeout=self.adj.asyncore_loop_timeout, - map=self.map, - use_poll=self.adj.asyncore_use_poll, - ) - except (SystemExit, KeyboardInterrupt): - self.close() + def pull_trigger(self): + for server in self.servers: + server.pull_trigger() + + def graceful_shutdown(self): + """ + Stop accepting new connections, let the requests that are already being + serviced finish, and then shut the server down. See ``_drain``. + """ + _drain(self.servers, self.map, self.adj, self.asyncore) + self.close() def close(self): - self.task_dispatcher.shutdown() + self.task_dispatcher.shutdown(timeout=self.adj.shutdown_timeout) wasyncore.close_all(self.map) -class BaseWSGIServer(wasyncore.dispatcher): +class BaseWSGIServer(_ServerRunner, wasyncore.dispatcher): channel_class = HTTPChannel next_channel_cleanup = 0 socketmod = socket # test shim - asyncore = wasyncore # test shim in_connection_overflow = False + trigger = None + # Whether closing this server should also shut down the task dispatcher. + # This is only true when nobody else holds a reference to the dispatcher, + # in which case we would otherwise leave its threads running forever. + own_task_dispatcher = False def __init__( self, @@ -225,28 +440,51 @@ def __init__( self.socktype = sockinfo[1] self.application = application self.adj = adj - self.trigger = trigger.trigger(map) + self.server_name = adj.server_name + self.active_channels = {} + + # Initialise the wasyncore dispatcher before acquiring anything else so + # that self.close() is always safe to call from the error handling + # below, even if we never got as far as creating a socket. + self.asyncore.dispatcher.__init__(self, _sock, map=map) + if dispatcher is None: + # Nobody else knows about this dispatcher, so we are the ones that + # have to shut it down again when we get closed. dispatcher = ThreadedTaskDispatcher() - dispatcher.set_thread_count(self.adj.threads) + self.own_task_dispatcher = True self.task_dispatcher = dispatcher - self.asyncore.dispatcher.__init__(self, _sock, map=map) - if _sock is None: - self.create_socket(self.family, self.socktype) - if self.family == socket.AF_INET6: # pragma: nocover - self.socket.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, 1) - self.set_reuse_addr() + try: + self.trigger = trigger.trigger(map) - if bind_socket: - self.bind_server_socket() + if _sock is None: + self.create_socket(self.family, self.socktype) + if self.family == socket.AF_INET6: # pragma: nocover + self.socket.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, 1) - self.effective_host, self.effective_port = self.getsockname() - self.server_name = adj.server_name - self.active_channels = {} - if _start: - self.accept_connections() + self.set_reuse_addr() + + if bind_socket: + self.bind_server_socket() + + self.effective_host, self.effective_port = self.getsockname() + + if self.own_task_dispatcher: + # Wait until we know that we have a working socket before + # starting any threads, otherwise a failure above would leave + # them running with no way to reach them. + self.task_dispatcher.set_thread_count(self.adj.threads) + + if _start: + self.accept_connections() + except BaseException: + # Don't leak the trigger, the socket, our entry in the socket map, + # or the task dispatcher's threads if we can't finish starting up. + # See https://github.com/Pylons/waitress/issues/480 + self.close() + raise def bind_server_socket(self): raise NotImplementedError # pragma: no cover @@ -320,15 +558,23 @@ def handle_accept(self): addr = self.fix_addr(addr) self.channel_class(self, conn, addr, self.adj, map=self._map) - def run(self): - try: - self.asyncore.loop( - timeout=self.adj.asyncore_loop_timeout, - map=self._map, - use_poll=self.adj.asyncore_use_poll, - ) - except (SystemExit, KeyboardInterrupt): - self.task_dispatcher.shutdown() + def graceful_shutdown(self): + """ + Stop accepting new connections, let the requests that are already being + serviced finish, and then shut the server down. See ``_drain``. + """ + _drain([self], self._map, self.adj, self.asyncore) + self.close() + + def stop_accepting(self): + """ + Close the listening socket, but leave everything else running. + + No new connections are accepted after this, while the channels that are + already open, the trigger the task threads use to wake up the main loop, + and the task dispatcher all keep working. + """ + wasyncore.dispatcher.close(self) def pull_trigger(self): self.trigger.pull_trigger() @@ -354,7 +600,22 @@ def print_listen(self, format_str): # pragma: no cover self.log_info(format_str.format(self.effective_host, self.effective_port)) def close(self): - self.trigger.close() + # Stop the worker threads first, they may still be holding on to a + # channel and they need the trigger to hand their output back to us. + if self.own_task_dispatcher: + self.task_dispatcher.shutdown(timeout=self.adj.shutdown_timeout) + + # Anything still connected at this point is not going to get an answer + # from us, drop it. Note that we deliberately don't use close_all() on + # the whole map here: it would call back into this method. + for channel in list(self.active_channels.values()): + channel.handle_close() + + # self.trigger is None only when we are being closed by an __init__ + # that failed before it got that far. + if self.trigger is not None: + self.trigger.close() + return wasyncore.dispatcher.close(self) diff --git a/src/waitress/task.py b/src/waitress/task.py index bda3ae50..9260680a 100644 --- a/src/waitress/task.py +++ b/src/waitress/task.py @@ -39,6 +39,19 @@ ) +def in_task_thread(): + """ + Return True if the calling thread is one of the threads a + ``ThreadedTaskDispatcher`` services tasks in, which is to say one of the + threads the WSGI application runs in. + + Waiting for a shutdown to complete from one of these deadlocks: the + shutdown is waiting for this very thread to finish what it is doing. + """ + + return getattr(threading.current_thread(), "waitress_task_thread", False) + + class ThreadedTaskDispatcher: """A Task Dispatcher that creates a thread for each task.""" @@ -59,6 +72,7 @@ def start_new_thread(self, target, thread_no): target=target, name=f"waitress-{thread_no}", args=(thread_no,) ) t.daemon = True + t.waitress_task_thread = True t.start() def handler_thread(self, thread_no): diff --git a/src/waitress/trigger.py b/src/waitress/trigger.py index 73ac31c3..8d667c0a 100644 --- a/src/waitress/trigger.py +++ b/src/waitress/trigger.py @@ -90,6 +90,12 @@ def close(self): self._close() # subclass does OS-specific stuff def pull_trigger(self, thunk=None): + if self._closed: + # There is no main loop left to wake up, and the file descriptor we + # would be writing to may well have been handed out to somebody + # else by now. + return + if thunk: with self.lock: self.thunks.append(thunk) diff --git a/tests/test_adjustments.py b/tests/test_adjustments.py index 86bf5ded..96282479 100644 --- a/tests/test_adjustments.py +++ b/tests/test_adjustments.py @@ -123,6 +123,7 @@ def test_goodvars(self): connection_limit="1000", cleanup_interval="1100", channel_timeout="1200", + shutdown_timeout="15", log_socket_errors="true", max_request_header_size="1300", max_request_body_size="1400", @@ -152,6 +153,7 @@ def test_goodvars(self): self.assertEqual(inst.connection_limit, 1000) self.assertEqual(inst.cleanup_interval, 1100) self.assertEqual(inst.channel_timeout, 1200) + self.assertEqual(inst.shutdown_timeout, 15) self.assertTrue(inst.log_socket_errors) self.assertEqual(inst.max_request_header_size, 1300) self.assertEqual(inst.max_request_body_size, 1400) diff --git a/tests/test_server.py b/tests/test_server.py index cede49a7..f32ad6bb 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,11 +1,29 @@ import errno +import select import socket +import sys +import threading +import time import unittest dummy_app = object() class TestWSGIServer(unittest.TestCase): + def setUp(self): + # Keep track of every server that gets created so that tearDown can + # close all of them, not just the last one a test held on to. + self.insts = [] + + def tearDown(self): + for inst in self.insts: + inst.close() + + def _register(self, inst): + self.insts.append(inst) + + return inst + def _makeOne( self, application=dummy_app, @@ -20,16 +38,17 @@ def _makeOne( ): from waitress.server import create_server - self.inst = create_server( - application, - host=host, - port=port, - map=map, - _dispatcher=_dispatcher, - _start=_start, - _sock=_sock, + return self._register( + create_server( + application, + host=host, + port=port, + map=map, + _dispatcher=_dispatcher, + _start=_start, + _sock=_sock, + ) ) - return self.inst def _makeOneWithMap( self, adj=None, _start=True, host="127.0.0.1", port=0, app=dummy_app @@ -55,15 +74,16 @@ def _makeOneWithMulti( map = {} from waitress.server import create_server - self.inst = create_server( - app, - listen=listen, - map=map, - _dispatcher=task_dispatcher, - _start=_start, - _sock=sock, + return self._register( + create_server( + app, + listen=listen, + map=map, + _dispatcher=task_dispatcher, + _start=_start, + _sock=sock, + ) ) - return self.inst def _makeWithSockets( self, @@ -80,22 +100,19 @@ def _makeWithSockets( _sockets = [] if sockets is not None: _sockets = sockets - self.inst = create_server( - application, - map=map, - _dispatcher=_dispatcher, - _start=_start, - _sock=_sock, - sockets=_sockets, - ) - return self.inst - def tearDown(self): - if self.inst is not None: - self.inst.close() + return self._register( + create_server( + application, + map=map, + _dispatcher=_dispatcher, + _start=_start, + _sock=_sock, + sockets=_sockets, + ) + ) def test_ctor_app_is_None(self): - self.inst = None self.assertRaises(ValueError, self._makeOneWithMap, app=None) def test_ctor_start_true(self): @@ -247,10 +264,6 @@ def test_handle_accept_noerror(self): def test_maintenance(self): inst = self._makeOneWithMap() - - class DummyChannel: - requests = [] - zombie = DummyChannel() zombie.last_activity = 0 zombie.running_tasks = False @@ -263,10 +276,10 @@ def test_backward_compatibility(self): from waitress.server import TcpWSGIServer, WSGIServer self.assertIs(WSGIServer, TcpWSGIServer) - self.inst = WSGIServer(None, _start=False, port=1234) + inst = self._register(WSGIServer(None, _start=False, port=1234)) # Ensure the adjustment was actually applied. self.assertNotEqual(Adjustments.port, 1234) - self.assertEqual(self.inst.adj.port, 1234) + self.assertEqual(inst.adj.port, 1234) def test_create_with_one_tcp_socket(self): from waitress.server import TcpWSGIServer @@ -311,6 +324,531 @@ def test_create_with_one_socket_handle_accept_noerror(self): self.assertListEqual(innersock.opts, [("level", "optname", "value")]) self.assertListEqual(L, [(inst, innersock, None, inst.adj)]) + def test_close_shuts_down_task_dispatcher(self): + inst = self._makeOne(_start=False, map={}) + dispatcher = inst.task_dispatcher + self.assertTrue(inst.own_task_dispatcher) + self.assertEqual(len(dispatcher.threads), inst.adj.threads) + inst.close() + self.assertEqual(dispatcher.threads, set()) + + def test_close_is_idempotent(self): + inst = self._makeOne(_start=False, map={}) + inst.close() + inst.close() + + def test_close_closes_active_channels(self): + inst = self._makeOneWithMap() + channel = DummyChannel() + inst.active_channels[100] = channel + inst.close() + self.assertTrue(channel.closed) + + def test_multi_close_shuts_down_task_dispatcher(self): + inst = self._makeOneWithMulti() + inst.close() + self.assertTrue(inst.task_dispatcher.was_shutdown) + self.assertEqual( + inst.task_dispatcher.shutdown_timeout, inst.adj.shutdown_timeout + ) + + def test_multi_finds_its_servers_in_the_map(self): + from waitress.server import BaseWSGIServer, MultiSocketServer + + inst = self._makeOneWithMulti() + # The servers can also be passed in, which is what create_server does, + # but they are discoverable from the socket map alone. + rediscovered = MultiSocketServer(inst.map, inst.adj, inst.effective_listen) + self.assertTrue(rediscovered.servers) + self.assertListEqual( + rediscovered.servers, + [s for s in inst.map.values() if isinstance(s, BaseWSGIServer)], + ) + + def test_ctor_does_not_start_threads_until_bound(self): + from waitress.server import TcpWSGIServer + + class FailingServer(TcpWSGIServer): + def bind_server_socket(self): + raise OSError(errno.EADDRINUSE, "Address already in use") + + map = {} + threads_before = threading.active_count() + + with self.assertRaises(OSError): + FailingServer(dummy_app, map=map, host="127.0.0.1", port=0) + + self.assertEqual(threading.active_count(), threads_before) + self.assertDictEqual(map, {}) + + +class TestServerCleanup(unittest.TestCase): + """ + A failure to create a server must not leave any threads, sockets or socket + map entries behind: the caller has no way of getting at them to clean them + up itself. See https://github.com/Pylons/waitress/issues/480 + """ + + def setUp(self): + self.insts = [] + + def tearDown(self): + for inst in self.insts: + inst.close() + + def _makeOne(self, **kw): + from waitress.server import create_server + + kw.setdefault("map", {}) + inst = create_server(dummy_app, **kw) + self.insts.append(inst) + + return inst + + def _makeOneListening(self): + """Create a server holding on to an ephemeral port.""" + inst = self._makeOne(host="127.0.0.1", port=0) + + return inst.effective_host, inst.effective_port + + @unittest.skipIf( + sys.platform.startswith("win"), + "Windows allows rebinding a port that is already being listened on", + ) + def test_bind_failure_does_not_leak(self): + from waitress.server import create_server + + host, port = self._makeOneListening() + + map = {} + threads_before = threading.active_count() + + with self.assertRaises(OSError) as cm: + create_server(dummy_app, host=host, port=port, map=map) + + self.assertEqual(cm.exception.errno, errno.EADDRINUSE) + # No worker threads were started, and no trigger or server socket was + # left registered in the socket map. + self.assertEqual(threading.active_count(), threads_before) + self.assertDictEqual(map, {}) + + @unittest.skipIf( + sys.platform.startswith("win"), + "Windows allows rebinding a port that is already being listened on", + ) + def test_partial_bind_failure_closes_the_servers_that_did_bind(self): + from waitress.server import create_server + + host, port = self._makeOneListening() + + map = {} + threads_before = threading.active_count() + + # The first of these binds successfully, the second one does not. The + # first one is not returned to us, so create_server has to close it. + with self.assertRaises(OSError) as cm: + create_server( + dummy_app, listen=f"127.0.0.1:0 {host}:{port}", map=map, threads=2 + ) + + self.assertEqual(cm.exception.errno, errno.EADDRINUSE) + self.assertEqual(threading.active_count(), threads_before) + self.assertDictEqual(map, {}) + + def test_no_sockets_to_listen_on(self): + dispatcher = DummyTaskDispatcher() + + with self.assertRaises(ValueError): + self._makeOne(listen="", _dispatcher=dispatcher) + + self.assertTrue(dispatcher.was_shutdown) + + def test_create_server_hands_dispatcher_to_single_server(self): + inst = self._makeOne(host="127.0.0.1", port=0) + self.assertTrue(inst.own_task_dispatcher) + + def test_create_server_keeps_dispatcher_for_multisocket(self): + from waitress.server import BaseWSGIServer + + map = {} + inst = self._makeOne(listen="127.0.0.1:0 127.0.0.1:0", map=map) + + for server in map.values(): + if isinstance(server, BaseWSGIServer): + # MultiSocketServer owns the dispatcher, closing a single one + # of the listening sockets must not stop the worker threads. + self.assertFalse(server.own_task_dispatcher) + + self.assertEqual(len(inst.task_dispatcher.threads), inst.adj.threads) + inst.close() + self.assertEqual(inst.task_dispatcher.threads, set()) + + +class TestDrain(unittest.TestCase): + """ + Tests for the loop that keeps the server running long enough for the + requests that are already in flight to finish. + """ + + def _callFUT(self, servers, map=None, asyncore=None, **kw): + from waitress.server import _drain + + if map is None: + map = {} + + if asyncore is None: + asyncore = DummyAsyncoreLoop() + + return _drain(servers, map, DummyShutdownAdj(**kw), asyncore) + + def test_stops_accepting_before_anything_else(self): + server = DummyDrainServer() + asyncore = DummyAsyncoreLoop() + self._callFUT([server], asyncore=asyncore) + self.assertTrue(server.stopped_accepting) + + def test_no_channels_does_not_run_the_loop(self): + asyncore = DummyAsyncoreLoop() + self._callFUT([DummyDrainServer()], asyncore=asyncore) + self.assertEqual(asyncore.calls, 0) + + def test_disabled_by_a_zero_timeout(self): + server = DummyDrainServer({1: DummyChannel()}) + asyncore = DummyAsyncoreLoop() + self._callFUT([server], asyncore=asyncore, shutdown_timeout=0) + self.assertTrue(server.stopped_accepting) + self.assertEqual(asyncore.calls, 0) + + def test_idle_channel_is_asked_to_close_once_flushed(self): + channel = DummyChannel() + server = DummyDrainServer({1: channel}) + # The loop "closes" the channel on the first pass through. + asyncore = DummyAsyncoreLoop(on_loop=lambda: server.active_channels.clear()) + self._callFUT([server], asyncore=asyncore) + self.assertTrue(channel.draining) + self.assertTrue(channel.close_when_flushed) + self.assertEqual(asyncore.calls, 1) + + def test_busy_channel_is_left_running(self): + channel = DummyChannel(requests=["a request being serviced"]) + server = DummyDrainServer({1: channel}) + asyncore = DummyAsyncoreLoop() + + with self.assertLogs("waitress", level="WARNING") as logged: + self._callFUT([server], asyncore=asyncore, shutdown_timeout=0.01) + + # We stop reading new requests off it, but we do not pull the rug out + # from underneath the task that is still running. + self.assertTrue(channel.draining) + self.assertFalse(channel.close_when_flushed) + self.assertIn("Graceful shutdown timed out", logged.output[0]) + + def test_second_interrupt_gives_up(self): + channel = DummyChannel(requests=["a request being serviced"]) + server = DummyDrainServer({1: channel}) + + def interrupt(): + raise KeyboardInterrupt + + asyncore = DummyAsyncoreLoop(on_loop=interrupt) + + with self.assertLogs("waitress", level="WARNING") as logged: + self._callFUT([server], asyncore=asyncore) + + self.assertIn("Graceful shutdown interrupted", logged.output[0]) + + +class TestGracefulShutdown(unittest.TestCase): + """ + End to end tests: a real socket, a real task dispatcher and a real WSGI + application, driven by a real wasyncore loop. + """ + + def _makeServer(self, app, **kw): + from waitress.server import create_server + + self.map = {} + server = create_server( + app, host="127.0.0.1", port=0, map=self.map, threads=1, **kw + ) + self.addCleanup(server.close) + + return server + + def _connect(self, server): + client = socket.create_connection( + (server.effective_host, int(server.effective_port)), timeout=10 + ) + self.addCleanup(client.close) + + return client + + def _pump(self, until, timeout=10): + """Run the main loop until ``until()`` is true.""" + from waitress import wasyncore + + deadline = time.time() + timeout + + while not until() and time.time() < deadline: + wasyncore.loop(timeout=0.01, map=self.map, count=1) + + self.assertTrue(until(), "timed out waiting for the server") + + def _read_all(self, client): + chunks = [] + + while True: + chunk = client.recv(4096) + + if not chunk: + return b"".join(chunks) + chunks.append(chunk) + + def test_in_flight_request_gets_its_response(self): + started = threading.Event() + may_finish = threading.Event() + + def app(environ, start_response): + started.set() + may_finish.wait(10) + start_response( + "200 OK", + [("Content-Type", "text/plain"), ("Content-Length", "5")], + ) + + return [b"hello"] + + server = self._makeServer(app) + client = self._connect(server) + client.sendall(b"GET / HTTP/1.0\r\nHost: localhost\r\n\r\n") + + # Hand the request off to the task thread, which then blocks. + self._pump(started.is_set) + + # This is the moment a Ctrl-C would land: the response has not been + # written yet, and it only ever will be if the main loop keeps running + # for long enough to pick it up from the task thread. + may_finish.set() + server.graceful_shutdown() + + response = self._read_all(client) + self.assertTrue(response.startswith(b"HTTP/1.0 200 OK"), response) + self.assertTrue(response.endswith(b"hello"), response) + + def test_response_larger_than_the_socket_buffers_is_not_truncated(self): + # A response that doesn't fit in the socket buffers can only be written + # out by the main loop: the task thread parks itself on outbuf_lock + # until the loop has drained enough of the outbuf for it to continue. + # Stopping the loop at that point both truncates the response and + # leaves the task thread wedged forever. + body_len = 4 * 1024 * 1024 + chunk = b"x" * 65536 + started = threading.Event() + + def app(environ, start_response): + started.set() + start_response( + "200 OK", + [("Content-Type", "text/plain"), ("Content-Length", str(body_len))], + ) + + return [chunk] * (body_len // len(chunk)) + + server = self._makeServer(app, outbuf_high_watermark=8192) + client = self._connect(server) + client.sendall(b"GET / HTTP/1.0\r\nHost: localhost\r\n\r\n") + + received = [] + reader = threading.Thread( + target=lambda: received.append(self._read_all(client)) + ) + reader.daemon = True + reader.start() + + self._pump(started.is_set) + server.graceful_shutdown() + + reader.join(30) + self.assertFalse(reader.is_alive(), "the connection was never closed") + response = received[0] + self.assertTrue(response.startswith(b"HTTP/1.0 200 OK"), response[:64]) + self.assertEqual(len(response) - response.index(b"\r\n\r\n") - 4, body_len) + # And the task thread is not left parked on the outbuf lock. + self.assertEqual(server.task_dispatcher.threads, set()) + + def test_stops_accepting_new_connections(self): + server = self._makeServer(dummy_app) + host, port = server.effective_host, int(server.effective_port) + + server.graceful_shutdown() + + with self.assertRaises(OSError): + conn = socket.create_connection((host, port), timeout=10) + conn.close() + + def test_idle_keepalive_connection_is_closed(self): + def app(environ, start_response): + start_response( + "200 OK", + [("Content-Type", "text/plain"), ("Content-Length", "2")], + ) + + return [b"ok"] + + server = self._makeServer(app) + client = self._connect(server) + client.sendall(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n") + + # Let the request complete. The connection stays open for reuse. + self._pump(lambda: _peek(client)) + self.assertEqual(len(server.active_channels), 1) + + server.graceful_shutdown() + + # The connection is closed rather than left hanging around until the + # channel timeout expires. + self.assertEqual(server.active_channels, {}) + self.assertIn(b"200 OK", self._read_all(client)) + + def test_run_shuts_down_gracefully(self): + server = self._makeServer(dummy_app) + server.asyncore = DummyAsyncore() + server.run() + self.assertIsNone(server.socket) + self.assertEqual(server.task_dispatcher.threads, set()) + + def test_trigger_is_closed(self): + server = self._makeServer(dummy_app) + trigger = server.trigger + server.graceful_shutdown() + # The pipe behind the trigger is what produces the "unclosed file" + # ResourceWarning on shutdown when it is left dangling. + # https://github.com/Pylons/waitress/issues/264 + self.assertTrue(trigger._closed) + self.assertEqual(self.map, {}) + + +class TestStartStop(unittest.TestCase): + """ + Running the server in a background thread and stopping it again, the + functionality webtest's StopableWSGIServer used to have to bolt on. + """ + + def _makeOne(self, app=None, listen="127.0.0.1:0", **kw): + from waitress.server import create_server + + if app is None: + app = hello_app + server = create_server(app, listen=listen, map={}, **kw) + self.addCleanup(server.close) + + return server + + def _get(self, host, port, path="/"): + client = socket.create_connection((host, int(port)), timeout=10) + + try: + client.sendall( + f"GET {path} HTTP/1.0\r\nHost: localhost\r\n\r\n".encode("latin-1") + ) + chunks = [] + + while True: + chunk = client.recv(4096) + + if not chunk: + return b"".join(chunks) + chunks.append(chunk) + finally: + client.close() + + def test_start_serves_requests_and_stop_stops(self): + server = self._makeOne() + server.start() + + # The port was picked by the OS, but we can still find out what it is, + # without having to race anybody for it. + # https://github.com/Pylons/waitress/issues/290 + host, port = server.effective_host, server.effective_port + self.assertNotEqual(int(port), 0) + self.assertIn(b"hello", self._get(host, port)) + + self.assertTrue(server.stop(timeout=30)) + + with self.assertRaises(OSError): + self._get(host, port) + + def test_start_twice(self): + server = self._makeOne() + server.start() + self.addCleanup(server.stop, 30) + self.assertRaises(RuntimeError, server.start) + + def test_stop_can_be_called_twice(self): + server = self._makeOne() + server.start() + self.assertTrue(server.stop(timeout=30)) + # The second one has nothing left to wait for. + self.assertFalse(server.stop()) + + def test_stop_without_start_does_not_wait(self): + server = self._makeOne() + self.assertFalse(server.stop()) + + def test_stop_from_the_application(self): + # Stopping from inside a request has to work, that's a request being + # serviced that the shutdown then has to wait for. + stopped = [] + holder = [] + + def app(environ, start_response): + stopped.append(holder[0].stop()) + start_response( + "200 OK", [("Content-Type", "text/plain"), ("Content-Length", "7")] + ) + + return [b"stopped"] + + server = self._makeOne(app) + holder.append(server) + server.start() + host, port = server.effective_host, server.effective_port + thread = server._thread + + # The response still has to make it out the door. + self.assertIn(b"stopped", self._get(host, port)) + # stop() did not wait: it was called from a thread that the shutdown + # itself has to wait for, so waiting would have deadlocked. + self.assertEqual(stopped, [False]) + + thread.join(30) + self.assertFalse(thread.is_alive()) + + def test_multisocket_start_and_stop(self): + server = self._makeOne(listen="127.0.0.1:0 127.0.0.1:0") + self.assertEqual(server.__class__.__name__, "MultiSocketServer") + server.start() + + for host, port in server.effective_listen: + self.assertIn(b"hello", self._get(host, port)) + + self.assertTrue(server.stop(timeout=30)) + + for host, port in server.effective_listen: + with self.assertRaises(OSError): + self._get(host, port) + + +def hello_app(environ, start_response): + start_response("200 OK", [("Content-Type", "text/plain"), ("Content-Length", "5")]) + + return [b"hello"] + + +def _peek(client): + """Return True if there is anything to read on ``client``.""" + return bool(select.select([client], [], [], 0)[0]) + if hasattr(socket, "AF_UNIX"): @@ -463,15 +1001,62 @@ def close(self): pass +class DummyChannel: + will_close = False + close_when_flushed = False + draining = False + + def __init__(self, requests=()): + self.requests = list(requests) + self.requests_lock = threading.Lock() + self.closed = False + + def handle_close(self): + self.closed = True + + +class DummyDrainServer: + def __init__(self, active_channels=None): + self.active_channels = active_channels if active_channels is not None else {} + self.stopped_accepting = False + + def stop_accepting(self): + self.stopped_accepting = True + + +class DummyShutdownAdj: + asyncore_loop_timeout = 1 + asyncore_use_poll = False + shutdown_timeout = 5 + + def __init__(self, **kw): + self.__dict__.update(kw) + + +class DummyAsyncoreLoop: + def __init__(self, on_loop=None): + self.calls = 0 + self.on_loop = on_loop + + def loop(self, timeout=30.0, use_poll=False, map=None, count=None): + self.calls += 1 + + if self.on_loop is not None: + self.on_loop() + + class DummyTaskDispatcher: + was_shutdown = False + def __init__(self): self.tasks = [] def add_task(self, task): self.tasks.append(task) - def shutdown(self): + def shutdown(self, cancel_pending=True, timeout=5): self.was_shutdown = True + self.shutdown_timeout = timeout class DummyTask: @@ -494,6 +1079,7 @@ class DummyAdj: socket_options = [("level", "optname", "value")] cleanup_interval = 900 channel_timeout = 300 + shutdown_timeout = 5 class DummyAsyncore: