Python3.x RuntimeError: Event loop is closed

Question

I'm doing something wrong with aiohttp/asyncio. I've got the following code working fine if I just run it on a one off, when I try to call run_my_job() from another file in a loop like so:

main.py
========================================
 count = 0
    batch_count = math.ceil((abc.get_count()/100))
    print("there are {} batches to complete.".format(batch_count))
    while count < batch_count:
        print("starting batch {}...".format(count))
        abc.run_my_job()
        print("batch {} completed...".format(count))
        count += 1


abc.py
===============================
def run_my_job(self):
    self.queue_manager(self.do_stuff(all_the_tasks))

def queue_manager(self, method):
    print('starting event queue')
    loop = asyncio.get_event_loop()
    future = asyncio.ensure_future(method)
    loop.run_until_complete(future)
    loop.close()

async def async_post(self, resource, session, data):
    async with session.post(self.api_attr.api_endpoint + resource, headers=self.headers, data=data) as response:
        resp = await response.read()
    return resp

async def do_stuff(self, data):
    print('queueing tasks')

    tasks = []
    async with aiohttp.ClientSession() as session:
        for row in data:
            task = asyncio.ensure_future(self.async_post('my_api_endpoint', session, row))
            tasks.append(task)
        result = await asyncio.gather(*tasks)
        self.load_results(result)
# goes on to load_results() method that parses json and updates the DB.

I get these errors:

Traceback (most recent call last):
  File "C:/usr/PycharmProjects/api_framework/api_framework.py", line 37, in <module>
starting event queue
    abc.run_my_job()
  File "C:\usr\PycharmProjects\api_framework\api\abc\abc.py", line 77, in run_eligibility
    self.queue_manager(self.verify_eligibility(json_data))
  File "C:\usr\PycharmProjects\api_framework\api\abc\abc.py", line 187, in queue_manager
    future = asyncio.ensure_future(method)
  File "C:\Python36x64\lib\asyncio\tasks.py", line 512, in ensure_future
    task = loop.create_task(coro_or_future)
  File "C:\Python36x64\lib\asyncio\base_events.py", line 282, in create_task
    self._check_closed()
  File "C:\Python36x64\lib\asyncio\base_events.py", line 357, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed
sys:1: RuntimeWarning: coroutine 'Consumer.run_my_job' was never awaited

score 12 · Accepted Answer · answered Jul 10 '18 at 18:06

Look at this function:

def queue_manager(self, method):
    print('starting event queue')
    loop = asyncio.get_event_loop()
    future = asyncio.ensure_future(method)
    loop.run_until_complete(future)
    loop.close()

This is what you call to schedule each job. And, at the end of the function, you close the event loop. So, after the first job runs, you close the event loop.

If you try to run more jobs after that, you're obviously trying to run them on a closed event loop. (And you've got jobs that run further jobs.) Hence the error:

RuntimeError: Event loop is closed

Just remove the loop.close() and that problem goes away.

I'm not sure that will be enough to make your program work, as you haven't given us anything near a runnable example—plus, in your real code, run_my_job is apparently a coroutine, but it isn't in the code you posted here. I don't see any other blatantly obvious errors in what you posted, but I don't know how much that mean.

That fixed it. I was under the impression I needed to close the loop to prevent issues. Do I need to close the loop at all then? — hyphen, Jul 10 '18 at 18:42
@hyphen You do want to close the loop, but at the end of the program, after all the jobs are done, not after each job is done. — abarnert, Jul 10 '18 at 18:49

Python3.x RuntimeError: Event loop is closed

1 Answers1

Linked