Graceful loop shutdown with running executor tasks

[asvetlov]

When I send a long task to tread-pool executor and immediately close the loop without waiting future returned by .run_in_executor() call I get exception report in logs:

ERROR:concurrent.futures:exception calling callback for <Future at 0x7ff1ae393630 state=finished returned NoneType>
Traceback (most recent call last):
  File "/usr/lib/python3.4/concurrent/futures/_base.py", line 297, in _invoke_callbacks
    callback(self)
  File "/home/andrew/projects/asyncio/asyncio/futures.py", line 410, in <lambda>
    new_future._copy_state, future))
  File "/home/andrew/projects/asyncio/asyncio/base_events.py", line 487, in call_soon_threadsafe
    handle = self._call_soon(callback, args)
  File "/home/andrew/projects/asyncio/asyncio/base_events.py", line 461, in _call_soon
    self._check_closed()
  File "/home/andrew/projects/asyncio/asyncio/base_events.py", line 288, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

That's because task executed in thread has finished after loop closing. The task calls call_soon_threadsafe() (see futures.wrap_future() for details) and it fails on check for loop health.

Due threaded nature it's hard to avoid situations like this: threads cannot be cancelled, we need cope with these even after loop closing. That's pretty common situation, especially for client-side jobs like web page fetching: DNS resolver is executed in thread pool also.

I see two options:

1. Ignore loop closing in wrap_future call, skip call_soon_threadsafe() for closed loops
2. Do the same but only if special boolean flag passed into run_in_executor() (while I don't see situations when I don't like the behavior, maybe flag should be True by default).

[1st1]

I think the first option is fine. But I'd like wrap_future to still log somehow that there was a concurrent.future that resolved after the loop was closed.

[asvetlov]

See http://stackoverflow.com/questions/32598231/asyncio-runtimeerror-event-loop-is-closed/32615276 for yet another example for the problem

[ludovic-gasc]

Thank you @asvetlov for your AsyncIO support on StackOverflow, a coworker told me he has found a solution from one of your answer ;-)

[nchammas]

For the record, I am running into this same issue with run_in_executor() when I close the event loop. @asvetlov's solution here works for me.

[gvanrossum]

I think throwing away the callback when the loop is already closed is fine (i.e., option 1). When closing a loop we throw away all callbacks that haven't run yet, so this would just extend that behavior.

Now, arguably it is actually inconsistent that calling call_soon() on a closed loop raises an error at all -- why not just ignore the call? However, given that no further callbacks are run, we don't expect any further callbacks to be scheduled either, since only callbacks -- or coroutines, which is the same in this context -- should schedule callbacks. So it arguably doesn't matter much, and it does occasionally catch errors (I presume).

But for call_soon_threadsafe() calls made from the executor it's a different story -- we explicitly don't wait for the workers to finish (since who knows when that might be) so we should expect such calls -- but we should treat them the same regardless of whether the thread finishe just before or just after the loop is closed. So option 1 is the right one.

[gvanrossum]

Pull Request would be welcome!

[rudyryk]

Can I be helpful with that? :) I'd probably be able to implement the first option if I get some initial key points.

So, for call_soon_threadsafe() we need to check self.is_closed(), right? Is it (probably) enough?

As for test case for this, I think it's clear.

[mosquito]

When I use my implementation of ThreadPool (uses threading.pool under the hood) an exception not raises.

loop = asyncio.get_event_loop()
pool = ThreadPool(loop=loop)
loop.set_default_executor(pool)

# do something

pool.shutdown(wait=True)
loop.close()

Also when I call pool.shutdown(wait=True) for futures.ThresdPoolExecutor an exception raises.

exception calling callback for <Future at 0x1064c1f28 state=finished returned NoneType>
Traceback (most recent call last):
  File "/usr/local/Cellar/python3/3.5.2_3/Frameworks/Python.framework/Versions/3.5/lib/python3.5/concurrent/futures/_base.py", line 297, in _invoke_callbacks
    callback(self)
  File "/usr/local/Cellar/python3/3.5.2_3/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/futures.py", line 442, in _call_set_state
    dest_loop.call_soon_threadsafe(_set_state, destination, source)
  File "/usr/local/Cellar/python3/3.5.2_3/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/base_events.py", line 532, in call_soon_threadsafe
    handle = self._call_soon(callback, args)
  File "/usr/local/Cellar/python3/3.5.2_3/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/base_events.py", line 506, in _call_soon
    self._check_closed()
  File "/usr/local/Cellar/python3/3.5.2_3/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/base_events.py", line 334, in _check_closed
    raise RuntimeError('Event loop is closed')
RuntimeError: Event loop is closed

For uvloop the same.

exception calling callback for <Future at 0x11019cfd0 state=finished returned NoneType>
Traceback (most recent call last):
  File "/usr/local/Cellar/python3/3.5.2_3/Frameworks/Python.framework/Versions/3.5/lib/python3.5/concurrent/futures/_base.py", line 297, in _invoke_callbacks
    callback(self)
  File "/usr/local/Cellar/python3/3.5.2_3/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/futures.py", line 442, in _call_set_state
    dest_loop.call_soon_threadsafe(_set_state, destination, source)
  File "uvloop/loop.pyx", line 1023, in uvloop.loop.Loop.call_soon_threadsafe (uvloop/loop.c:23215)
  File "uvloop/loop.pyx", line 500, in uvloop.loop.Loop._call_soon (uvloop/loop.c:13237)
  File "uvloop/loop.pyx", line 504, in uvloop.loop.Loop._call_soon_handle (uvloop/loop.c:13296)
  File "uvloop/loop.pyx", line 532, in uvloop.loop.Loop._check_closed (uvloop/loop.c:13856)
RuntimeError: Event loop is closed

This looks like special behaviour of futures.ThresdPoolExecutor.

[argaen]

Some tests from our stack fail because of this bug. We could mock the executor but since they are end to end tests ideally we shouldn't mock anything.

What's the current status of this? Is anyone working on it? Do we have a reference in https://bugs.python.org/?