tcpclient: cancelling TCPClient.connect

Hello Ben,
I think I found a problem with TCPClient. I have a web application which connects and disconnects to/from a TCP peer on user request. Now, when the user connects and while the TCPClient.connect method still is waiting (stream = await self.connect(self.host, self.port) already requests a stop I run into an exception. But first in order to handle start/stop requests I implemented something like this:

async def start(self):
    self.task = asyncio.ensure_future(self._run())

async def stop(self):
    self.task.cancel()
    self.task = None

async def _run(self):
    # --1-- Connect to peer
    try:
        stream = await self.connect(self.host, self.port)
    except StreamClosedError:
        return
    ... now read from the stream

The exception I see is this:

2018-03-31 16:10:11,027 [ERROR] Exception in callback functools.partial(<function wrap.<locals>.null_wrapper at 0x000001FF4ABB1400>, <Future finished exception=StreamClosedError('Stream is closed',)>)
Traceback (most recent call last):
  File "C:\Development\Maelstrom\src\libs\tornado\gen.py", line 1107, in run
    yielded = self.gen.throw(*exc_info)
  File "C:\Development\Maelstrom\src\libs\tornado\tcpclient.py", line 234, in connect
    af, addr, stream = yield connector.start(connect_timeout=timeout)
  File "C:\Development\Maelstrom\src\libs\tornado\gen.py", line 1099, in run
    value = future.result()
  File "C:\Development\Maelstrom\src\libs\tornado\tcpclient.py", line 112, in on_connect_done
    stream = future.result()
tornado.iostream.StreamClosedError: Stream is closed

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:\Development\Maelstrom\src\libs\tornado\ioloop.py", line 760, in _run_callback
    ret = callback()
  File "C:\Development\Maelstrom\src\libs\tornado\stack_context.py", line 276, in null_wrapper
    return fn(*args, **kwargs)
  File "C:\Development\Maelstrom\src\libs\tornado\gen.py", line 1199, in inner
    self.run()
  File "C:\Development\Maelstrom\src\libs\tornado\gen.py", line 1139, in run
    future_set_exc_info(self.result_future, sys.exc_info())
  File "C:\Development\Maelstrom\src\libs\tornado\concurrent.py", line 597, in future_set_exc_info
    future.set_exception(exc_info[1])
asyncio.base_futures.InvalidStateError: invalid state

The problematic part of code is in tcpclient.py:

    def on_connect_done(self, addrs, af, addr, future):
        self.remaining -= 1
        try:
            stream = future.result()
        except Exception as e:
            if self.future.done():
                return
            # Error: try again (but remember what happened so we have an
            # error to raise in the end)
            self.last_error = e
            self.try_connect(addrs)
            if self.timeout is not None:
                # If the first attempt failed, don't wait for the
                # timeout to try an address from the secondary queue.
                self.io_loop.remove_timeout(self.timeout)
                self.on_timeout()
            return

If I cancel the task then the future passed to on_connect_done is already done when this function is called (a CancelledError is set an thus future.done() is TRUE). However, you check again self.future.done() which returns FALSE since it is a different future.

Adding a check to future.done() solves the problem for me:

    def on_connect_done(self, addrs, af, addr, future):
        self.remaining -= 1
        try:
            stream = future.result()
        except Exception as e:
            if future.done():   # <-- ADDED
                return              # <-- ADDED
            if self.future.done():
                return

I'm not sure if the first check to self.future.done() is required.

Kind Regards
Ralf

Start in tcpclient.py at TCPClient.connect and Connector.on_connect_done, then follow the cancellation path described in the report. Reproduce a cancelled connection and verify that it no longer produces StreamClosedError followed by asyncio InvalidStateError.