_sock_connect_cb can be called twice resulting in InvalidStateError

[thehesiod]

BPO	25593
Nosy	@gvanrossum, @asvetlov, @1st1, @thehesiod
Files	asyncio_invalid_state_bt.txt: Backtrace of my InvalidStateError exception Issue25593_repro_client.py: Repro client script (requires server script) Issue25593_repro_server.py: Server repro script test_app.py run_once_testfix_for_Issue25593.patch: Run loop._ready callback loops at start of _run_once Issue25593_fix.patch: Patch submission for using stop flag issue25593_revised.patch issue25593_revised_2.patch issue25593_revised_3.patch issue25593_revised_4.patch issue25593_revised_5.patch

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = 'https://github.com/gvanrossum'
closed_at = <Date 2015-11-19.21:35:54.971>
created_at = <Date 2015-11-10.07:16:34.030>
labels = ['type-bug', 'expert-asyncio']
title = '_sock_connect_cb can be called twice resulting in InvalidStateError'
updated_at = <Date 2016-02-10.22:39:48.267>
user = 'https://github.com/thehesiod'

bugs.python.org fields:

activity = <Date 2016-02-10.22:39:48.267>
actor = 'vstinner'
assignee = 'gvanrossum'
closed = True
closed_date = <Date 2015-11-19.21:35:54.971>
closer = 'gvanrossum'
components = ['asyncio']
creation = <Date 2015-11-10.07:16:34.030>
creator = 'thehesiod'
dependencies = []
files = ['41015', '41016', '41017', '41018', '41019', '41059', '41080', '41082', '41084', '41086', '41087']
hgrepos = []
issue_num = 25593
keywords = ['patch']
message_count = 50.0
messages = ['254433', '254462', '254484', '254488', '254495', '254496', '254497', '254498', '254500', '254501', '254502', '254503', '254504', '254505', '254506', '254507', '254508', '254509', '254510', '254512', '254513', '254514', '254518', '254535', '254536', '254538', '254544', '254551', '254775', '254779', '254781', '254783', '254786', '254910', '254913', '254914', '254921', '254923', '254924', '254925', '254926', '254930', '254933', '254934', '254936', '254937', '259971', '259972', '260027', '260042']
nosy_count = 6.0
nosy_names = ['gvanrossum', 'asvetlov', 'python-dev', 'yselivanov', 'thehesiod', 'Justin Mayfield']
pr_nums = []
priority = 'normal'
resolution = 'fixed'
stage = None
status = 'closed'
superseder = None
type = 'behavior'
url = 'https://bugs.python.org/issue25593'
versions = ['Python 3.4', 'Python 3.5', 'Python 3.6']

[thehesiod]

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.

[gvanrossum]

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?

[thehesiod]

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!

[thehesiod]

I'm going to close this as I've found a work-around, if I find a better test-case I'll open a new bug.

[thehesiod]

Actually, I just realized I had fixed it locally by changing the callback to the following:
429 def _sock_connect_cb(self, fut, sock, address):
430 if fut.cancelled() or fut.done():
431 return

so a fix is still needed, and I also verified this happens with python3.4 as well.

[thehesiod]

clarification, adding the fut.done() check, or monkey patching:
orig_sock_connect_cb = asyncio.selector_events.BaseSelectorEventLoop._sock_connect_cb
def _sock_connect_cb(self, fut, sock, address):
if fut.done(): return
return orig_sock_connect_cb(self, fut, sock, address)

[gvanrossum]

Sorry, the code you posted is still incomprehensible. E.g. I suppose your
worker doesn't really have

obj = self.queue.get()

but rather something like

obj = yield from async_queue.get()

But in the end, even with that hypothesis, I can't explain what you're
seeing, and I believe there is a bug related to bad mixing multiprocessing
and asyncio in some code you're not showing, and your "fix" just masks the
problem. Note that the code you posted doesn't touch sockets in any way,
while the issue you're seeing is related to sockets. So there *must* be
more to it.

[thehesiod]

self.queue is not an async queue, as I stated above its a multiprocessing queue. This code is to multiplex a multiprocessing queue to a async queue.

[JustinMayfield]

I believe I'm seeing this bug in a non-threaded and non-forked env.

System:
OSX 10.11.1 (15B42)
Python 3.5.0 (from brew install)

I'm using aiohttp to create several dozens of HTTP connections to the same server (an async tornado web server). Nothing special is being done around the event loop creation (standard get_event_loop()). However in my system the event loop is frequently stopped, via ioloop.stop(), and restarted via ioloop.run_forever(). I'm not sure this is related to the issue yet, but it's worth mentioning.

I can't provide simplified test code just yet, but I can reproduce in my env with nearly 100% odds when doing a full system test. Attached is a sample backtrace.

[thehesiod]

Perhaps I'm doing something really stupid, but I was able to reproduce the two issues I'm having with the following sample script. If you leave the monkey patch disabled, you get the InvalidStateError, if you enable it, you get the ServerDisconnect errors that I'm currently seeing which I work-around with retries. Ideas?

import asyncio
import aiohttp
import multiprocessing
import aiohttp.server
import logging
import traceback

# Monkey patching
import asyncio.selector_events

# http://bugs.python.org/issue25593
if False:
    orig_sock_connect_cb = asyncio.selector_events.BaseSelectorEventLoop._sock_connect_cb
    def _sock_connect_cb(self, fut, sock, address):
        if fut.done(): return
        return orig_sock_connect_cb(self, fut, sock, address)
    asyncio.selector_events.BaseSelectorEventLoop._sock_connect_cb = _sock_connect_cb


class HttpRequestHandler(aiohttp.server.ServerHttpProtocol):
    @asyncio.coroutine
    def handle_request(self, message, payload):
        response = aiohttp.Response(self.writer, 200, http_version=message.version)
        response.add_header('Content-Type', 'text/html')
        response.add_header('Content-Length', '18')
        response.send_headers()
        yield from asyncio.sleep(0.5)
        response.write(b'<h1>It Works!</h1>')
        yield from response.write_eof()


def process_worker(q):
    loop = asyncio.get_event_loop()
    #loop.set_debug(True)
    connector = aiohttp.TCPConnector(force_close=False, keepalive_timeout=8, use_dns_cache=True)
    session = aiohttp.ClientSession(connector=connector)
    async_queue = asyncio.Queue(100)

    @asyncio.coroutine
    def async_worker(session, async_queue):
        while True:
            try:
                print("blocking on asyncio queue get")
                url = yield from async_queue.get()
                print("unblocking on asyncio queue get")
                print("get aqueue size:", async_queue.qsize())
                response = yield from session.request('GET', url)
                try:
                    data = yield from response.read()
                    print(data)
                finally:
                    yield from response.wait_for_close()
            except:
                traceback.print_exc()

    def producer(q):
        print("blocking on multiprocessing queue get")
        obj2 = q.get()
        print("unblocking on multiprocessing queue get")
        print("get qempty:", q.empty())
        return obj2

    def worker_done(f):
        try:
            f.result()
            print("worker exited")
        except:
            traceback.print_exc()

    workers = []
    for i in range(100):
        t = asyncio.ensure_future(async_worker(session, async_queue))
        t.add_done_callback(worker_done)
        workers.append(t)

    @asyncio.coroutine
    def doit():
        print("start producer")
        obj = yield from loop.run_in_executor(None, producer, q)
        print("finish producer")

        print("blocking on asyncio queue put")
        yield from async_queue.put(obj)
        print("unblocking on asyncio queue put")
        print("put aqueue size:", async_queue.qsize())

    while True:
        loop.run_until_complete(doit())


def server():
    loop = asyncio.get_event_loop()
    #loop.set_debug(True)

    f = loop.create_server(lambda: HttpRequestHandler(debug=True, keep_alive=75), '0.0.0.0', '8080')

    srv = loop.run_until_complete(f)
    loop.run_forever()


if __name__ == '__main__':
    q = multiprocessing.Queue(100)

    log_proc = multiprocessing.log_to_stderr()
    log_proc.setLevel(logging.DEBUG)

    p = multiprocessing.Process(target=process_worker, args=(q,))
    p.start()

    p2 = multiprocessing.Process(target=server)
    p2.start()

    while True:
        print("blocking on multiprocessing queue put")
        q.put("http://0.0.0.0:8080")
        print("unblocking on multiprocessing queue put")

        print("put qempty:", q.empty())

[gvanrossum]

I wonder if the bug is in aiohttp? The code you show is still too complex
to debug for me.