-
-
Notifications
You must be signed in to change notification settings - Fork 346
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #307 from njsmith/serve
Add high-level server helpers
- Loading branch information
Showing
10 changed files
with
497 additions
and
63 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,129 @@ | ||
| import errno | ||
| import logging | ||
| import os | ||
|
|
||
| import trio | ||
|
|
||
| __all__ = ["serve_listeners"] | ||
|
|
||
| # Errors that accept(2) can return, and which indicate that the system is | ||
| # overloaded | ||
| ACCEPT_CAPACITY_ERRNOS = { | ||
| errno.EMFILE, | ||
| errno.ENFILE, | ||
| errno.ENOMEM, | ||
| errno.ENOBUFS, | ||
| } | ||
|
|
||
| # How long to sleep when we get one of those errors | ||
| SLEEP_TIME = 0.100 | ||
|
|
||
| # The logger we use to complain when this happens | ||
| LOGGER = logging.getLogger("trio.serve_listeners") | ||
|
|
||
|
|
||
| async def _run_handler(stream, handler): | ||
| try: | ||
| await handler(stream) | ||
| finally: | ||
| await trio.aclose_forcefully(stream) | ||
|
|
||
|
|
||
| async def _serve_one_listener(listener, handler_nursery, handler): | ||
| async with listener: | ||
| while True: | ||
| try: | ||
| stream = await listener.accept() | ||
| except OSError as exc: | ||
| if exc.errno in ACCEPT_CAPACITY_ERRNOS: | ||
| LOGGER.error( | ||
| "accept returned %s (%s); retrying in %s seconds", | ||
| errno.errorcode[exc.errno], | ||
| os.strerror(exc.errno), | ||
| SLEEP_TIME, | ||
| exc_info=True | ||
| ) | ||
| await trio.sleep(SLEEP_TIME) | ||
| else: | ||
| raise | ||
| else: | ||
| handler_nursery.start_soon(_run_handler, stream, handler) | ||
|
|
||
|
|
||
| async def serve_listeners( | ||
| handler, | ||
| listeners, | ||
| *, | ||
| handler_nursery=None, | ||
| task_status=trio.STATUS_IGNORED | ||
| ): | ||
| """Listen for incoming connections on ``listeners``, and for each one | ||
| start a task running ``handler(stream)``. | ||
| .. warning:: | ||
| If ``handler`` raises an exception, then this function doesn't do | ||
| anything special to catch it – so by default the exception will | ||
| propagate out and crash your server. If you don't want this, then catch | ||
| exceptions inside your ``handler``, or use a ``handler_nursery`` object | ||
| that responds to exceptions in some other way. | ||
| Args: | ||
| handler: An async callable, that will be invoked like | ||
| ``handler_nursery.start_soon(handler, stream)`` for each incoming | ||
| connection. | ||
| listeners: A list of :class:`~trio.abc.Listener` objects. | ||
| :func:`serve_listeners` takes responsibility for closing them. | ||
| handler_nursery: The nursery used to start handlers, or any object with | ||
| a ``start_soon`` method. If ``None`` (the default), then | ||
| :func:`serve_listeners` will create a new nursery internally and use | ||
| that. | ||
| task_status: This function can be used with ``nursery.start``, which | ||
| will return ``listeners``. | ||
| Returns: | ||
| This function never returns unless cancelled. | ||
| Resource handling: | ||
| If ``handler`` neglects to close the ``stream``, then it will be closed | ||
| using :func:`trio.aclose_forcefully`. | ||
| Error handling: | ||
| Most errors coming from :meth:`~trio.abc.Listener.accept` are allowed to | ||
| propagate out (crashing the server in the process). However, some errors – | ||
| those which indicate that the server is temporarily overloaded – are | ||
| handled specially. These are :class:`OSError`\s with one of the following | ||
| errnos: | ||
| * ``EMFILE``: process is out of file descriptors | ||
| * ``ENFILE``: system is out of file descriptors | ||
| * ``ENOBUFS``, ``ENOMEM``: the kernel hit some sort of memory limitation | ||
| when trying to create a socket object | ||
| When :func:`serve_listeners` gets one of these errors, then it: | ||
| * Logs the error to the standard library logger ``trio.serve_listeners`` | ||
| (level = ERROR, with exception information included). By default this | ||
| causes it to be printed to stderr. | ||
| * Waits 100 ms before calling ``accept`` again, in hopes that the | ||
| system will recover. | ||
| """ | ||
| async with trio.open_nursery() as nursery: | ||
| if handler_nursery is None: | ||
| handler_nursery = nursery | ||
| for listener in listeners: | ||
| nursery.start_soon( | ||
| _serve_one_listener, listener, handler_nursery, handler | ||
| ) | ||
| # The listeners are already queueing connections when we're called, | ||
| # but we wait until the end to call started() just in case we get an | ||
| # error or whatever. | ||
| task_status.started(listeners) |
Oops, something went wrong.