From 8f8445171cf47713253618e52312b4d219957de5 Mon Sep 17 00:00:00 2001 From: Delta Regeer Date: Sun, 2 Aug 2026 02:09:06 -0600 Subject: [PATCH 1/2] Shut down gracefully, and clean up when starting up fails Stopping a waitress server dropped whatever was in flight on the floor. The main loop was torn down the moment the interrupt arrived, but the task threads hand their output back to that loop and use the trigger to wake it up, so anything that had not already been written to the socket was lost. A task thread parked on outbuf_lock waiting for the loop to drain a buffer never woke up again either, and the trigger's pipe and the listening socket were left dangling. Interrupting a 4MB response used to look like this: received 327356 of 4194447 bytes 1 thread(s) still running ResourceWarning: unclosed file ResourceWarning: unclosed ResourceWarning: unclosed Shutting down now follows the steps laid out in #269: close the listening sockets, stop reading new requests off the channels that are still open, keep running the main loop until the requests that are already being serviced have been answered, and only then stop the worker threads and close the trigger. How long that may take is bounded by a new shutdown_timeout adjustment, which also replaces the timeout that was hardcoded in ThreadedTaskDispatcher.shutdown(). This is available as server.graceful_shutdown(), and as start()/stop() for running a server in a background thread. stop() signals the main loop through the trigger rather than closing things underneath it from another thread, so it is safe to call from anywhere, including from the WSGI application. Together with effective_port that covers what WebTest's StopableWSGIServer had to work around. Separately, failing to create a server leaked. The worker threads were started before the first socket was bound, and a listener that bound successfully was never closed if a later one failed, leaving threads and sockets running that the caller had no way to reach. The threads are now started only once every socket is bound, and anything created before a failure is cleaned up again. BaseWSGIServer.close() also stops the task dispatcher and closes the connections that are still open, which is what MultiSocketServer.close() already did. Fixes #480 Fixes #402 Fixes #134 Fixes #264 Fixes #290 --- CHANGES.txt | 55 +++ docs/arguments.rst | 13 + docs/runner.rst | 8 + docs/usage.rst | 82 +++++ src/waitress/adjustments.py | 9 + src/waitress/channel.py | 9 +- src/waitress/runner.py | 8 + src/waitress/server.py | 483 ++++++++++++++++++++------ src/waitress/task.py | 14 + src/waitress/trigger.py | 6 + tests/test_adjustments.py | 2 + tests/test_server.py | 660 ++++++++++++++++++++++++++++++++++-- 12 files changed, 1199 insertions(+), 150 deletions(-) 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/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: From 009ba20ea4903e5dfa8190acaf64deaec49142af Mon Sep 17 00:00:00 2001 From: Delta Regeer Date: Sun, 2 Aug 2026 02:24:33 -0600 Subject: [PATCH 2/2] Bump the version to 3.1.0.dev0 This release adds a new shutdown_timeout adjustment and the start()/ stop() API, so it is not going out as a 3.0.x. Pinning it down now avoids the documented '.. versionadded:: 3.1.0' guessing at a version that turns out to be a different one by the time it ships. --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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"