_sock_connect_cb can be called twice resulting in InvalidStateError

[thehesiod]

I'm posting this bug here as recommendation for Guido on the bug I logged there: http://bugs.python.org/issue25593. It seems like there's a logic error in terms of _sock_connect_cb whereby it can get called twice under certain circumstances. If I apply the simple fix to check if the future is already done I run into occasional connection errors.

Here is my simple testcase:

import asyncio
import aiohttp
import multiprocessing
import traceback
import http.server
import time

# Monkey patching
# http://bugs.python.org/issue25593
if False:
    import asyncio.selector_events
    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


def process_worker(q):
    loop = asyncio.get_event_loop()
    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(async_queue):

        while True:
            try:
                url = yield from async_queue.get()
                response = yield from session.request('GET', url)

                try:
                    data = yield from response.read()
                    print(data)
                finally:
                    yield from response.wait_for_close()
                    pass
            except:
                traceback.print_exc()

    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(async_queue))
        t.add_done_callback(worker_done)
        workers.append(t)

    def producer(q):
        obj2 = q.get()
        return obj2

    @asyncio.coroutine
    def doit():
        obj = yield from loop.run_in_executor(None, producer, q)
        yield from async_queue.put(obj)

    while True:
        loop.run_until_complete(doit())


class GetHandler(http.server.SimpleHTTPRequestHandler):
    def do_GET(self):
        f = self.send_head()
        time.sleep(0.5)
        if f:
            try:
                self.copyfile(f, self.wfile)
            finally:
                f.close()


def run_server(server_class=http.server.HTTPServer, handler_class=GetHandler):
    server_address = ('0.0.0.0', 8080)
    httpd = server_class(server_address, handler_class)
    httpd.serve_forever()


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

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

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

    while True:
        q.put("http://0.0.0.0:8080")

[asvetlov]

Fixed by upstream http://bugs.python.org/issue25593

[thehesiod]

@asvetlov we just got this in production with aiohttp 3.3.1 + python 3.6.6 :( Unfortunately not sure how to reproduce yet.

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/asyncio/events.py", line 145, in _run
    self._callback(*self._args)
  File "/usr/local/lib/python3.6/asyncio/selector_events.py", line 487, in _sock_connect_cb
    fut.set_result(None)
asyncio.base_futures.InvalidStateError: invalid state

[asvetlov]

Could you describe how the problem is related to aiohttp?
AFAIK aiohttp never uses loop.sock_connect() API.
Maybe you have to blame another library?

[thehesiod]

unable to place exact blame, just an FYI since this is happening on our aiohttp server but we have no more details yet as it only happens sporadically.

[asvetlov]

Sorry for my attitude, it is not constructive.
sock_connect may still have a bug (at least looking at sources).
Thank you for raising the issue.
Please create a new bug on bugs.python.org and add me to noise list.

[thehesiod]

will do as soon as we have a reproduce-able case. info here in case anyone else runs into it so we can pool resources.

[kingangelAOA]

2019-01-25 17:30:50 Desktop-493 asyncio[3728] ERROR Exception in callback BaseSelectorEventLoop._sock_connect_cb(, <socket.socke....236', 12138)>, ('10.40.40.236', 12138))
handle: <Handle BaseSelectorEventLoop._sock_connect_cb(, <socket.socke....236', 12138)>, ('10.40.40.236', 12138))>
Traceback (most recent call last):
File "/home/jyan/.pyenv/versions/3.7.1/lib/python3.7/asyncio/events.py", line 88, in _run
self._context.run(self._callback, *self._args)
File "/home/jyan/.pyenv/versions/3.7.1/lib/python3.7/asyncio/selector_events.py", line 512, in _sock_connect_cb
fut.set_result(None)
asyncio.base_futures.InvalidStateError: invalid state

aiohttp: 3.5.4

tasks = []
for i in range(self.users):
tasks.append(self.async_worker())
self.loop.run_until_complete(asyncio.gather(*tasks))

and put it in multiprocessing.Process

[asvetlov]

multiprocessing?
I'm not surprised, asyncio is not supposed to work from a child process.
I mean asyncio loop is not supposed to be forked, this is the problem.

[thehesiod]

as long as each sub-process has a new event loop it should work. I've worked with python maintainers before to ensure this workflow.

[thehesiod]

looks like a PR for official support is stalled: https://bugs.python.org/issue22087, this discussion: python/asyncio#452 let to this fix: python/asyncio#497