Cancelled recv() calls are logged as errors

[Badg]

So first of all, this is a pretty minor thing. But I'd love it if my logging weren't getting populated by a bunch of spurious "Exception in connection handler" when I'm gracefully closing my event loop using task cancellation.

An example "log pollution":

ERROR:websockets.server:Exception in connection handler
Traceback (most recent call last):
  File "C:\Python35\lib\site-packages\websockets\server.py", line 78, in handler
    yield from self.ws_handler(self, path)
  File "C:\Python35\lib\asyncio\coroutines.py", line 105, in __next__
    return self.gen.send(None)
  File "c:\py_hypergolix\hypergolix\comms.py", line 167, in _handle_connection
    msg = await websocket.recv()
  File "C:\Python35\lib\site-packages\websockets\protocol.py", line 266, in recv
    loop=self.loop, return_when=asyncio.FIRST_COMPLETED)
  File "C:\Python35\lib\asyncio\tasks.py", line 341, in wait
    return (yield from _wait(fs, timeout, return_when, loop))
  File "C:\Python35\lib\asyncio\tasks.py", line 424, in _wait
    yield from waiter
  File "C:\Python35\lib\asyncio\futures.py", line 358, in __iter__
    yield self  # This tells Task to wait for completion.
  File "C:\Python35\lib\asyncio\tasks.py", line 290, in _wakeup
    future.result()
  File "C:\Python35\lib\asyncio\futures.py", line 266, in result
    raise CancelledError
concurrent.futures._base.CancelledError

Note that this is strictly a problem for logging; the tasks themselves need to have the error thrown as appropriate to actually cancel the tasks, and then eventually close the client/server. But it then muddies up the logs, which again, is a relatively minor thing, but still important.

I'm also open to other options -- perhaps a CancelledError should be caught and then throw a ConnectionClosed error? But it seems to me like it's appropriate to still separate these, even if Websockets treats them similarly behind the scenes.

Thanks!

[aaugustin]

Can you try closing the server with:

        server.close()
        self.loop.run_until_complete(server.wait_closed())

before shutting down the event loop?

If you're already doing this and seeing these errors in the logs, then I'll have to try to reproduce the bug.

If you aren't and it solve the problem, then it's a documentation issue.

[Badg]

Ah, it looks like I missed a wait_closed call (this is all happening from within the loop, so I don't need to run_until_complete because I never stopped the loop). This also means I can point to a very likely cause (though admittedly I don't have time to repro it locally):

1. Have server running
2. From within loop (see below for specifics on what I'm doing), call server.close()
3. While server is closing, independently cancel the server

That being said, according to the asyncio docs, cancellation should be an expected means of stopping tasks. Perhaps in the future websockets should support that? I agree that short-term, documentation is probably sufficient.

By the way, thanks for the great work with the library. Async is awesome and websockets is an excellent application for it.

---

My specific looping strategy looks like this. It's going to be a very long code block, and then some discussion at the end about what's relevant to this issue. Hopefully this is helpful to someone and not just white noise:

class LooperTrooper(metaclass=abc.ABCMeta):
    ''' Basically, the Arduino of event loops.
    Requires subclasses to define an async loop_init function and a 
    loop_run function. Loop_run is handled within a "while running" 
    construct.

    Optionally, async def loop_stop may be defined for cleanup.

    LooperTrooper handles threading, graceful loop exiting, etc.

    if threaded evaluates to False, must call LooperTrooper().start() to
    get the ball rolling.

    *args and **kwargs are passed to the required async def loop_init.
    '''
    def __init__(self, threaded, thread_name=None, debug=False, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._shutdown_init_flag = None
        self._shutdown_complete_flag = threading.Event()
        self._debug = debug
        self._death_timeout = 1

        if threaded:
            self._loop = asyncio.new_event_loop()
            # Set up a thread for the loop
            self._thread = threading.Thread(
                target = self.start,
                args = args,
                kwargs = kwargs,
                # This may result in errors during closing.
                # daemon = True,
                # This isn't currently stable enough to close properly.
                daemon = False,
                name = thread_name
            )
            self._thread.start()

        else:
            self._loop = asyncio.get_event_loop()
            # Declare the thread as nothing.
            self._thread = None

    async def loop_init(self, *args, **kwargs):
        ''' This will be passed any *args and **kwargs from self.start,
        either through __init__ if threaded is True, or when calling 
        self.start directly.
        '''
        pass

    @abc.abstractmethod
    async def loop_run(self):
        pass

    async def loop_stop(self):
        pass

    def start(self, *args, **kwargs):
        ''' Handles everything needed to start the loop within the 
        current context/thread/whatever. May be extended, but MUST be 
        called via super().
        '''
        try:
            self._loop.set_debug(self._debug)

            if self._thread is not None:
                asyncio.set_event_loop(self._loop)

            # Set up a shutdown event and then start the task
            self._shutdown_init_flag = asyncio.Event()
            self._looper_future = asyncio.ensure_future(
                self._execute_looper(*args, **kwargs)
            )
            self._loop.run_until_complete(self._looper_future)

        finally:
            self._loop.close()
            # stop_threadsafe could be waiting on this.
            self._shutdown_complete_flag.set()

    def halt(self):
        warnings.warn(DeprecationWarning(
            'Halt is deprecated. Use stop() or stop_threadsafe().'
        ))
        if self._thread is not None:
            self.stop_threadsafe()
        else:
            self.stop()

    def stop(self):
        ''' Stops the loop INTERNALLY.
        '''
        self._shutdown_init_flag.set()

    def stop_threadsafe(self):
        ''' Stops the loop EXTERNALLY.
        '''
        self._loop.call_soon_threadsafe(self._shutdown_init_flag.set)
        self._shutdown_complete_flag.wait()

    async def catch_interrupt(self):
        ''' Workaround for Windows not passing signals well for doing
        interrupts.

        Standard websockets stuff.

        Deprecated? Currently unused anyways.
        '''
        while not self._shutdown_init_flag.is_set():
            await asyncio.sleep(5)

    async def _execute_looper(self, *args, **kwargs):
        ''' Called by start(), and actually manages control flow for 
        everything.
        '''
        await self.loop_init(*args, **kwargs)

        try:
            while not self._shutdown_init_flag.is_set():
                await self._step_looper()

        except CancelledError:
            pass

        finally:
            # Prevent cancellation of the loop stop.
            await asyncio.shield(self.loop_stop())
            await self._kill_tasks()

    async def _step_looper(self):
        ''' Execute a single step of _execute_looper.
        '''
        task = asyncio.ensure_future(self.loop_run())
        interrupt = asyncio.ensure_future(self._shutdown_init_flag.wait())

        finished, pending = await asyncio.wait(
            fs = [task, interrupt],
            return_when = asyncio.FIRST_COMPLETED
        )

        # Note that we need to check both of these, or we have a race
        # condition where both may actually be done at the same time.
        if task in finished:
            # Raise any exception, ignore result, rinse, repeat
            self._raise_if_exc(task)
        else:
            task.cancel()

        if interrupt in finished:
            self._raise_if_exc(interrupt)
        else:
            interrupt.cancel()

    async def _kill_tasks(self):
        ''' Kill all remaining tasks. Call during shutdown. Will log any
        and all remaining tasks.
        '''
        all_tasks = asyncio.Task.all_tasks()

        for task in all_tasks:
            if task is not self._looper_future:
                logging.info('Task remains while closing loop: ' + repr(task))
                task.cancel()

        if len(all_tasks) > 0:
            await asyncio.wait(all_tasks, timeout=self._death_timeout)

    @staticmethod
    def _raise_if_exc(fut):
        if fut.exception():
            raise fut.exception()


class ConnectorBase(LooperTrooper):
    ''' Common stuff for websockets clients and servers.

    Todo: refactor websockets stuff into a mix-in so that this class can
    be used for different transports.
    '''
    def __init__(self, host, port, receiver, *args, **kwargs):
        ''' Yeah, the usual.
        host -> str: hostname for the server
        port -> int: port for the server
        receiver -> coro: handler for incoming objects. Must have async def 
            receiver.receive(), which will be passed the connection, message
        connection_class -> type: used to create new connections. Defaults to 
            _WSConnection.
        '''
        self._ws_port = port
        self._ws_host = host
        self._receiver = receiver

        super().__init__(*args, **kwargs)

    @property
    def _ws_loc(self):
        return 'ws://' + self._ws_host + ':' + str(self._ws_port) + '/'

    @property
    def connection_factory(self):
        ''' Proxy for connection factory to allow saner subclassing.
        '''
        return _WSConnection

    async def new_connection(self, websocket, path, *args, **kwargs):
        ''' Wrapper for creating a new connection. Mostly here to make
        subclasses simpler.
        '''
        logger.debug('New connection: ' + str(args) + ' ' + str(kwargs))
        return self.connection_factory(
            loop = self._loop,
            websocket = websocket, 
            path = path, 
            *args, **kwargs
        )

    async def _handle_connection(self, websocket, path=None):
        ''' This handles a single websockets connection.
        '''
        connection = await self.new_connection(websocket, path)

        try:
            while True:
                msg = await websocket.recv()
                await self._receiver(connection, msg)

        except ConnectionClosed:
            pass

        except Exception:
            await connection.close()
            raise

        else:
            await connection.close()


class WSBasicServer(ConnectorBase):
    ''' Generic websockets server.
    '''
    def __init__(self, birthday_bits=40, *args, **kwargs):
        ''' 
        Note: birthdays must be > 1000, or it will be ignored, and will
        default to a 40-bit space.
        '''
        # When creating new connection ids,
        # Select a pseudorandom number from approx 40-bit space. Should have 1%
        # collision probability at 150k connections and 25% at 800k
        self._birthdays = 2 ** birthday_bits
        self._connections = {}
        self._connid_lock = None
        self._server = None

        # Make sure to call this last, lest we drop immediately into a thread.
        super().__init__(*args, **kwargs)

    @property
    def connections(self):
        ''' Access the connections dict.
        '''
        return self._connections

    async def new_connection(self, websocket, path, *args, **kwargs):
        ''' Generates a new connection object for the current conn.

        Must be called from super() if overridden.
        '''
        # Note that this overhead happens only once per connection.
        async with self._connid_lock:
            # Grab a connid and initialize it before releasing
            connid = self._new_connid()
            # Go ahead and set it to None so we block for absolute minimum time
            self._connections[connid] = None

        connection = await super().new_connection(
            websocket = websocket, 
            path = path, 
            connid = connid,
            *args, **kwargs
        )
        self._connections[connid] = connection

        return connection

    def _new_connid(self):
        ''' Creates a new connection ID. Does not need to use CSRNG, so
        let's avoid depleting entropy.

        THIS IS NOT COOP SAFE! Must be called with a lock to avoid a 
        race condition. Release the lock AFTER registering the connid.

        Standard websockets stuff.
        '''
        # Select a pseudorandom number from approx 40-bit space. Should have 1%
        # collision probability at 150k connections and 25% at 800k
        connid = random.randint(0, self._birthdays)
        if connid in self._connections:
            connid = self._new_connid()
        return connid

    async def loop_init(self):
        await super().loop_init()
        self._connid_lock = asyncio.Lock()

    async def loop_run(self):
        self._server = await websockets.serve(
            self._handle_connection, 
            self._ws_host, 
            self._ws_port
        )
        await self._server.wait_closed()

    async def loop_stop(self):
        # Todo: add in logic to gracefully handle all of the connection objects
        # Note that if we error out before calling loop_run (for example, if 
        # the server fails to start), we actually don't have a server to close.
        if self._server is not None:
            self._server.close()
            # This was the missing line
            await self._server.wait_closed()

        await super().loop_stop()

Note that I've added the missing ```await self._server.wait_closed()" call in the above code, at the very bottom. However, the discussion below is what happens without it.

Okay, so my execution flow during a healthy close is:

1. Have a parent task loop_run call await websockets.serve(...), then await its closure
2. From either within the loop, or threadsafe from without, set the close event
3. loop_run's parent, _step_looper, detects the awaited close event, and then cancels loop_run, which passes the cancellation to the server, but also breaks out of the wait_closed() call from within loop_run.
4. Control returns to _step_looper, which is now finished, and then propagates back up the chain to its parent _execute_looper.
5. _execute_looper calls a shielded loop_stop, which closes the server, and anything else that needs closing. This is where I missed wait_closed().
6. Control returns to _execute_looper, which calls a final catch-all "close and log everything that remains" _kill_tasks just in case something has been misbehaving
7. _kill_tasks cancels the server before it is fully closed
8. The loop is complete. It stops and is closed.

I guess ultimately the question then becomes, should cancellation of the server be an acceptable method of closing it? It seems to me like maybe it should, given asyncio's admonition to be able to handle cancellation gracefully. Perhaps it would be appropriate to wrap the internals of websockets.serve(...) in a try/catch block that converts CancelledError into server.close()? Apologies if this is already the case; though that would indicate that there's a bug somewhere causing it to behave otherwise.

Cheers!

[aaugustin]

I just released version 3.2 which contains a major refactor of the server shutdown sequence. The previous implementation was far from sufficient for most practical cases.

With the new version, if you call server.close(); await server.wait_closed(), you shouldn't see these log messages anymore (at least I hope so).

You might still see some spurious output if you kill the event loop while awaiting server.wait_closed(), but I'm not sure I should try to do anything about it: the server is in the process of shutting down cleanly; if you want to go faster, well, you'll spill some pending tasks.

[aaugustin]

Tentatively closing this issue — please let me know if the behavior of version 3.2 still isn't satisfactory for your use case and I'll reopen.