Aiohttp: Python 3.4.3 Event loop is closed Runtime Error

[digitaldavenyc]

I recently upgraded to Python 3.4.3 and also upgraded aiohttp to the latest version and experiencing a Runtime error on almost all requests after the event loop runs for a while. I did see that this issue was resolved in a previous ticket #411 but it doesn't seem to be from what I can tell.

I included source code below.

exception calling callback for <Future at 0x10bc03668 state=finished returned list>
Traceback (most recent call last):
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/concurrent/futures/_base.py", line 297, in _invoke_callbacks
    callback(self)
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/futures.py", line 408, in <lambda>
    new_future._copy_state, future))
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 462, in call_soon_threadsafe
    handle = self._call_soon(callback, args)
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 436, in _call_soon
    self._check_closed()
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 265, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

def run():
    while condition is True:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(generate(data))
        loop.close()

def request(data):
    try:
        response = aiohttp.request(method='GET', url=data['url'])

        result = yield from asyncio.wait_for(response, timeout=10)

        body = yield from result.read()
        response.close()

        return result, data, response, body
    except Exception as e:
        data['response_time'] = 10
        return e.__class__.__name__, data

def generate(data):
    coroutines = []

     for value in data:
     coroutines.append(asyncio.Task(retrieve.request(value)))

     results = yield from asyncio.gather(*coroutines, return_exceptions=True)

[popravich]

Hi, @digitaldavenyc,
First of all, the code above is incorrect:

def request(data):
    # here the response is actually the request --
    #   generator object which should be yielded from.
    response = aiohttp.request(method='GET', url=data['url'])

    # the result is actual response object.
    result = yield from asyncio.wait_for(response, timeout=10)
    body = yield from result.read()

    # this will raise GeneratorExit inside `response` generator
    #   but it is already exited so this line does nothing.
    #   this line should look like this -- result.close()
    response.close()

Also avoid creating tasks directly, use asyncio.async instead.

asyncio.Task(retrieve.request(value))  # use `asyncio.async(request(value))`

Maybe this will help.

[popravich]

If that didn't work maybe there is some problem with aiohttp.request but the traceback is poor and it is hard to tell what went wrong

[digitaldavenyc]

I updated my code to the following and still receiving the same error. It will run through the main loop 10 times in some cases, others it terminates on the first try.

    def request(data):
        try:
            r = aiohttp.request(method='GET', url=data['url'])
            response = yield from asyncio.wait_for(r, timeout=10)
            body = yield from response.read()    
            response.close()

            return response, body

    def generate():
        coroutines = []

        for value in self.data:
            coroutines.append(asyncio.async(retrieve.request(value)))

        results = yield from asyncio.gather(*coroutines, return_exceptions=True)

    def run():
        for i in range(25):
            selector = selectors.SelectSelector()
            loop = asyncio.SelectorEventLoop(selector)
            asyncio.set_event_loop(loop)
            loop.run_until_complete(generate())
            loop.close()

exception calling callback for <Future at 0x1156fc390 state=finished returned list>
Traceback (most recent call last):
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/concurrent/futures/_base.py", line 297, in _invoke_callbacks
    callback(self)
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/futures.py", line 408, in <lambda>
    new_future._copy_state, future))
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 462, in call_soon_threadsafe
    handle = self._call_soon(callback, args)
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 436, in _call_soon
    self._check_closed()
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 265, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

[fafhrd91]

instead of response.close() try yield from response.read_and_close()

[digitaldavenyc]

Changed body = yield from response.read() to body = yield from response.read_and_close() and error still occurs.

[fafhrd91]

could you post full source, i'd like to debug it locally

[digitaldavenyc]

import selectors
import aiohttp
import pandas as pd
import time
import asyncio


class Requests(object):
    def __init__(self):
        self.df = pd.DataFrame()
        self.csv = '/etc/project/urls.csv'
        self.data = []
        self.results = {}

    def run(self):

        for i in range(25):

            self.next()

            start = time.time()

            selector = selectors.SelectSelector()
            loop = asyncio.SelectorEventLoop(selector)
            asyncio.set_event_loop(loop)
            loop.run_until_complete(self.generate())
            loop.close()

            print(time.time() - start)

    def next(self):
        self.data = []
        self.results = {}

        if self.df.empty:
            self.df = pd.read_csv(self.csv)

        for index, value in self.df.iterrows():
            if self.data.__len__() >= 50:
                break

            self.data.append(value.to_dict())
            self.df = self.df.drop(index)

        self.df.to_csv(self.csv, index=False)

    def generate(self):
        coroutines = []

        for value in self.data:
            coroutines.append(asyncio.async(request(value)))

        self.results = yield from asyncio.gather(*coroutines, return_exceptions=True)


def request(data):
    try:
        r = aiohttp.request(method='GET', url=data['url'])

        start = time.time()
        response = yield from asyncio.wait_for(r, timeout=10)
        duration = time.time() - start

        data['response_time'] = duration
        body = yield from response.read_and_close()

        return response, data, body
    except Exception as e:
        data['response_time'] = 10
        return e.__class__.__name__, data

[digitaldavenyc]

It may have to run through a few iterations before failing. Thanks for taking a look at it!

[digitaldavenyc]

I just tried enabling async debug mode to try to get some more information on this and a socket.gaierror displayed. This error has caused problems in aiohttp in the past but I'm not sure the two are related.

xception calling callback for <Future at 0x10b427c18 state=finished raised gaierror>
Traceback (most recent call last):
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/concurrent/futures/thread.py", line 54, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 505, in _getaddrinfo_debug
    addrinfo = socket.getaddrinfo(host, port, family, type, proto, flags)
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/socket.py", line 533, in getaddrinfo
    for res in _socket.getaddrinfo(host, port, family, type, proto, flags):
socket.gaierror: [Errno 8] nodename nor servname provided, or not known

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/concurrent/futures/_base.py", line 297, in _invoke_callbacks
    callback(self)
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/futures.py", line 408, in <lambda>
    new_future._copy_state, future))
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 462, in call_soon_threadsafe
    handle = self._call_soon(callback, args)
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 436, in _call_soon
    self._check_closed()
  File "/source/homebrew/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/base_events.py", line 265, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

[digitaldavenyc]

@fafhrd91 I ran some additional tests and there is a definitive correlation between TimeoutError errors firing on asyncio.wait_for and the RuntimeError on the event loop. Any thoughts on why this is and how to prevent it?

[fafhrd91]

i think this is bug in asyncio, it might be that asyncio do not cancel gataddrinfo() future on timeout.

@asvetlov what do you think?

[digitaldavenyc]

I found a bug in Python! How does one fix a bug in Python?

[popravich]

Heh, looks like it is a really tricky bug:

1. getaddrinfo() is blocking and so executed in ThreadPoolExecutor;
2. the aiohttp.request gets cancelled with timeout and returned with all other results by asyncio.gather();
3. next step the loop is being closed however ThreadPoolExecutor is still blocked by getaddrinfo()
   (which is trying to resolve unresolvable host);
4. loop.close shutdowns executor in non-blocking manner (does not wait for threads to shutdown);
5. and finally getaddrinfo() raises error and tries to pass exception to parent future which, eventually, raises RuntimeError

[popravich]

(this is basically what I can tell from looking through the code, maybe a test case can reproduce it)

[popravich]

Test to reproduce the issue:

import time
import asyncio

@asyncio.coroutine
def go(loop):
    def long_task():
        time.sleep(2)
        print("I'm done, wait for me!")

    f = loop.run_in_executor(None, long_task)
    w = asyncio.wait_for(f, 1, loop=loop)
    yield from asyncio.gather(w, return_exceptions=True, loop=loop)

def main():
    loop = asyncio.new_event_loop()
    loop.run_until_complete(go(loop))
    loop.close()
    print("Loop is closed")

    loop = asyncio.new_event_loop()
    loop.run_until_complete(asyncio.sleep(2, loop=loop))

main()

[digitaldavenyc]

Any thoughts on how I should approach this?

[popravich]

Hi, @digitaldavenyc
In your case you should use single event loop created at the begining of the program and closed when program is complete.

[digitaldavenyc]

I'm not sure how I would do that, this program is designed to run for a long time, I have it running infinitely until all urls are exhausted, and it's a massive list.

[asvetlov]

@digitaldavenyc well, just before loop.close() call push that trick:

loop.stop()
loop.run_forever()

It allows pending closing calls finish correctly.

[digitaldavenyc]

@asvetlov How would I pass the generate function to create the list of urls into the event loop in this situation? You can't pass anything into loop.run_forever() Can you call run_until_complete and run_forever in the same event loop?

[asvetlov]

I suggest

loop.run_until_complete(very_long_coroutine())
loop.stop()
loop.run_forever()
loop.close()

[digitaldavenyc]

So you can call both, awesome. That seems to work, for now, will need to run for a while to know if the issue in the original post occurs with the update. Thanks @asvetlov

[lock[bot]]

This thread has been automatically locked since there has not been
any recent activity after it was closed. Please open a [new issue] for
related bugs.

If you feel like there's important points made in this discussion,
please include those exceprts into that [new issue].