[MRG] Fix a race condition during joblib's dask backend termination

[pierreglaser]

@ogrisel this PR should fix #959 (comment). Feel free to checkout this PR and try it out on your machine using #959's repro.

task: <Task pending name='Task-6045' coro=<DaskDistributedBackend._collect() running at /home/ogrisel/code/joblib/joblib/_dask.py:197> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7fae2aaf0b50>()]> cb=[IOLoop.add_future.<locals>.<lambda>() at /home/ogrisel/miniconda3/envs/pylatest/lib/python3.8/site-packages/tornado/ioloop.py:690]>

This PR only highlights the problem (terminate will destroy _collect while _collect has not had the time to exit yet.
I'm not against using a strong synchronization primitive, but we would need to be careful about it.

[codecov]

Codecov Report

Merging #1078 into master will decrease coverage by 0.70%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##           master    #1078      +/-   ##
==========================================
- Coverage   94.38%   93.67%   -0.71%     
==========================================
  Files          47       47              
  Lines        6908     6910       +2     
==========================================
- Hits         6520     6473      -47     
- Misses        388      437      +49     

Impacted Files	Coverage Δ
joblib/_dask.py	94.52% <100.00%> (+0.05%)	⬆️
joblib/backports.py	44.73% <0.00%> (-39.48%)	⬇️
joblib/_memmapping_reducer.py	94.33% <0.00%> (-2.27%)	⬇️
joblib/test/test_memmapping.py	97.33% <0.00%> (-1.91%)	⬇️
joblib/test/test_parallel.py	95.39% <0.00%> (-1.66%)	⬇️
joblib/pool.py	86.17% <0.00%> (-1.63%)	⬇️
joblib/disk.py	90.47% <0.00%> (-1.59%)	⬇️
joblib/logger.py	85.52% <0.00%> (-1.32%)	⬇️
joblib/func_inspect.py	90.41% <0.00%> (-1.20%)	⬇️
joblib/test/test_memory.py	98.34% <0.00%> (-0.16%)	⬇️
... and 1 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 54109d9...c661a38. Read the comment docs.

[ogrisel]

That fixes it for me as well.

However there is a CI failure with:

joblib.externals.loky.process_executor.ShutdownExecutorError: cannot schedule new futures after shutdown

As for the solution itself, the 10ms overhead sound acceptable, it's unlikely people will do dask distributed parallel calls for workload that last less than a couple of seconds. However this line would need a comment to explain why and leave open an implementation that would actually check explicitly wait for the _collect corountine to complete.

[pierreglaser]

(just checking this failure is not random)

[ogrisel]

So it is random. Let me push another commit.

[ogrisel]

The test_multithreaded_parallel_termination_resource_tracker_silent failure seems to often happen on this branch. It means that the loky backend probably has a race condition that we did not know about.

Anyway the fix for the dask problem in this branch is not acceptable. It should be addressed in the dask backend itself, not in the Parallel class.

[pierreglaser]

The test_multithreaded_parallel_termination_resource_tracker_silent failure seems to often happen on this branch. It means that the loky backend probably has a race condition that we did not know about.

Yes, that is worth investigating, although not the highest in the priority list for me (I'd like to work on fixing memory caching in jupyter notebook + adding the dask benchmarks first)

[ogrisel]

LGTM if CI is happy.

[pierreglaser]

This also needs a changelog entry (@ogrisel feel free to write up something)