_sock_connect_cb can be called twice resulting in InvalidStateError

@gvanrossum

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']

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.

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?

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!

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.

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.

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)

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.

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.

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.

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())

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

Ha, email race.

Regarding rev 2, the updated docstring and scheduled stop looks good along with alleviating the confusion I mentioned.

I'm not sure about your warning comment; Perhaps that's a patch I didn't lay eyes on.

Cheers.

No, I mentioned the idea of a warning in the thread on the
python-tulip mailing list, but decided not to do it after all.

I see. Seems like good discussion over there. I joined up.

OK, here's another revision of the patch, setting the timeout passed to the selector to 0 when the loop is pre-stopped.

OK, another revision, keep the mock selector.

Whoops. Hopefully this one's right.

New changeset 9b3144716d17 by Guido van Rossum in branch '3.4':
Issue bpo-25593: Change semantics of EventLoop.stop().
https://hg.python.org/cpython/rev/9b3144716d17

New changeset 158cc5701488 by Guido van Rossum in branch '3.5':
Issue bpo-25593: Change semantics of EventLoop.stop(). (Merge 3.4->3.5)
https://hg.python.org/cpython/rev/158cc5701488

New changeset 2ebe03a94f8f by Guido van Rossum in branch 'default':
Issue bpo-25593: Change semantics of EventLoop.stop(). (Merge 3.5->3.6)
https://hg.python.org/cpython/rev/2ebe03a94f8f

Hopefully this is it!

I'm not sure if you guys are still listening on this closed bug but I think I've found another issue ;) I'm using python 3.5.1 + asyncio 3.4.3 with the latest aiobotocore (which uses aiohttp 0.21.0) and had two sessions (two TCPConnectors), one doing a multitude of GetObjects via HTTP1.1, and the other doing PutObject, and the PutObject session returns error 61 (connection refused) from the same _sock_connect_cb. It feels like a similar issue to the original. I'll see if I can get small testcase.

update: its unrelated to the number of sessions or SSL, but instead to the number of concurrent aiohttp requests. When set to 500, I get the error, when set to 100 I do not.

Alexander,

That sounds unrelated. I'd treat it as a new issue until you have concrete evidence to the contrary.

Also on face value it sounds like it might just be your operating systems open file limit. On OSX I think the default open file limit is in the hundreds (256 on my box). Generally on unix-oid platforms it can be checked and changed with the ulimit -n command.

Cheers

sorry for disruption! ends up our router seems to be doing some kind of QoS limits on # of connections :(

Navigation Menu

Repository navigation

_sock_connect_cb can be called twice resulting in InvalidStateError #69779

27 remaining items

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

_sock_connect_cb can be called twice resulting in InvalidStateError #69779

Description

Activity

thehesiod commented on Nov 10, 2015

gvanrossum commented on Nov 10, 2015

thehesiod commented on Nov 11, 2015

thehesiod commented on Nov 11, 2015

thehesiod commented on Nov 11, 2015

thehesiod commented on Nov 11, 2015

gvanrossum commented on Nov 11, 2015

thehesiod commented on Nov 11, 2015

JustinMayfield commented on Nov 11, 2015

thehesiod commented on Nov 11, 2015

gvanrossum commented on Nov 11, 2015

27 remaining items

JustinMayfield commented on Nov 19, 2015

gvanrossum commented on Nov 19, 2015

JustinMayfield commented on Nov 19, 2015

gvanrossum commented on Nov 19, 2015

gvanrossum commented on Nov 19, 2015

gvanrossum commented on Nov 19, 2015

python-dev commented on Nov 19, 2015

gvanrossum commented on Nov 19, 2015

thehesiod commented on Feb 10, 2016

thehesiod commented on Feb 10, 2016

JustinMayfield commented on Feb 10, 2016

thehesiod commented on Feb 10, 2016

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions