[issue25593] _sock_connect_cb can be called twice resulting in InvalidStateError

[Alexander Mohr]

New submission from Alexander Mohr:

asyncio.selector_events.BaseSelectorEventLoop._sock_connect_cb is a callback based on the selector for a socket. There are certain situations when the selector triggers twice calling this callback twice, resulting in an InvalidStateError when it sets the Future to None. The way I triggered this was by having several parallel connections to the same host in a multiprocessing script. I suggest analyzing why this callback can be called twice and figuring out what the correct fix is. I monkey patched it by adding a fut.done() check at the top. If this information is not enough I can try to provide a sample script. Its currently reproducing in a fairly involved multiprocessing script.

----------
components: asyncio
messages: 254433
nosy: gvanrossum, haypo, thehesiod, yselivanov
priority: normal
severity: normal
status: open
title: _sock_connect_cb can be called twice resulting in InvalidStateError
type: behavior
versions: Python 3.5

_______________________________________
Python tracker <***@bugs.python.org>
<http://bugs.python.org/issue25593>
_______________________________________

[Guido van Rossum]

Guido van Rossum added the comment:

Please show us how to repro -- there's no way we can figure out how this "impossible" event could happen in your code without understanding your code. Is it possible that multiprocessing forked your event loop or something similarly esoteric?

----------

_______________________________________
Python tracker <***@bugs.python.org>
<http://bugs.python.org/issue25593>
_______________________________________

[Alexander Mohr]

Alexander Mohr added the comment:

Sorry for being obscure before, it was hard to pinpoint. I think I just figured it out! I had code like this in a subprocess:

def worker():
while True:
obj = self.queue.get()
# do work with obj using asyncio http module

def producer():
nonlocal self
obj2 = self.queue.get()
return obj2


workers = []
for i in range(FILE_OP_WORKERS):
t = asyncio.ensure_future(worker())
t.add_done_callback(op_finished)
workers.append(t)

while True:
f = loop.run_in_executor(None, producer)
obj = loop.run_until_complete(f)

t = async_queue.put(obj)
loop.run_until_complete(t)

loop.run_until_complete(asyncio.wait(workers))

where self.queue is a multiprocessing.Queue, and async_queue is an asyncio queue. The idea is that I have a process populating a multiprocessing queue, and I want to transfer it to an syncio queue while letting the workers do their thing.

Without knowing the underlying behavior, my theory is that when python blocks on the multiprocessing queue lock, it releases socket events to the async http module's selectors, and then when the async loop gets to the selectors they're released again.

If I switch the producer to instead use a queue.get_nowait and busy wait with asyncio.sleep I don't get the error...however this is not ideal is we're busy waiting.

Thanks!

----------

_______________________________________
Python tracker <***@bugs.python.org>
<http://bugs.python.org/issue25593>
_______________________________________