Repository: esmvalgroup/esmvalcore
Issue: Failed NCL diagnostic script does not exit but hangs

**Describe the bug**
Clean exit after failed NCL diagnostic fails if in multiprocessing, resulting in hanging
First discovered via new integration test in #514 

**Steps to reproduce**
- install the latest version of `esmvalcore`
- set `max_parallel_tasks: ` > 1 in config-user
- run an NCL diagnostic script that fails for any particular reason

**Tracebacks**
```
2020-03-02 15:33:29,017 UTC [6285] INFO    Maximum memory used (estimate): 0.9 GB
2020-03-02 15:33:29,018 UTC [6285] INFO    Sampled every second. It may be inaccurate if short but high spikes in memory consumption occur.
2020-03-02 15:33:29,018 UTC [6285] ERROR   Program terminated abnormally, see stack trace below for more information
multiprocessing.pool.RemoteTraceback: 
"""
Traceback (most recent call last):
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/pool.py", line 125, in worker
    result = (True, func(*args, **kwds))
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_task.py", line 694, in _run_task
    output_files = task.run()
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_task.py", line 246, in run
    self.output_files = self._run(input_files)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_task.py", line 507, in _run
    self._control_ncl_execution(process, last_line + lines)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_task.py", line 411, in _control_ncl_execution
    raise DiagnosticError(msg)
esmvalcore._task.DiagnosticError: An error occurred during execution of NCL script /home/valeriu/esmvaltool_workshop_21Jan2019/plop.ncl, see the log in /home/valeriu/esmvaltool_workshop_21Jan2019/dryrun/recipe_dask_20200302_153310/run/validation_basic/ipython/log.txt
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_main.py", line 220, in run
    conf = main(args)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_main.py", line 156, in main
    process_recipe(recipe_file=recipe, config_user=cfg)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_main.py", line 206, in process_recipe
    recipe.run()
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_recipe.py", line 1320, in run
    run_tasks(self.tasks,
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_task.py", line 615, in run_tasks
    _run_tasks_parallel(tasks, max_parallel_tasks)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_task.py", line 661, in _run_tasks_parallel
    _copy_results(task, running[task])
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/site-packages/esmvalcore/_task.py", line 682, in _copy_results
    task.output_files, updated_products = future.get()
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/pool.py", line 768, in get
    raise self._value
esmvalcore._task.DiagnosticError: An error occurred during execution of NCL script /home/valeriu/esmvaltool_workshop_21Jan2019/plop.ncl, see the log in /home/valeriu/esmvaltool_workshop_21Jan2019/dryrun/recipe_dask_20200302_153310/run/validation_basic/ipython/log.txt
2020-03-02 15:33:29,021 UTC [6285] INFO    If you suspect this is a bug or need help, please open an issue on https://github.com/ESMValGroup/ESMValTool/issues and attach the run/recipe_*.yml and run/main_log_debug.txt files from the output directory.

```
(it is in a hanging state) - upon Ctrl-C-ing:
```
^CProcess ForkPoolWorker-2:
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/util.py", line 277, in _run_finalizers
    finalizer()
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/util.py", line 201, in __call__
    res = self._callback(*self._args, **self._kwargs)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/pool.py", line 689, in _terminate_pool
Traceback (most recent call last):
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/pool.py", line 114, in worker
    task = get()
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/queues.py", line 355, in get
    with self._rlock:
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
KeyboardInterrupt
    cls._help_stuff_finish(inqueue, task_handler, len(pool))
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/pool.py", line 674, in _help_stuff_finish
    inqueue._rlock.acquire()
KeyboardInterrupt
Process ForkPoolWorker-1:
Traceback (most recent call last):
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/pool.py", line 114, in worker
    task = get()
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/queues.py", line 356, in get
    res = self._reader.recv_bytes()
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/connection.py", line 216, in recv_bytes
    buf = self._recv_bytes(maxlength)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/connection.py", line 414, in _recv_bytes
    buf = self._recv(4)
  File "/home/valeriu/anaconda3R/envs/esmvaltool_py/lib/python3.8/multiprocessing/connection.py", line 379, in _recv
    chunk = read(handle, remaining)
KeyboardInterrupt
```


Changed files:
esmvalcore/_task.py

diff --git a/esmvalcore/_task.py b/esmvalcore/_task.py
index 9a897d83e..e5a20d1bb 100644
--- a/esmvalcore/_task.py
+++ b/esmvalcore/_task.py
@@ -630,13 +630,12 @@ def _run_tasks_parallel(tasks, max_parallel_tasks=None):
     scheduled = get_flattened_tasks(tasks)
     running = {}
 
-    n_scheduled, n_running = len(scheduled), len(running)
-    n_tasks = n_scheduled
+    n_tasks = n_scheduled = len(scheduled)
+    n_running = 0
 
     if max_parallel_tasks is None:
         max_parallel_tasks = os.cpu_count()
-    if max_parallel_tasks > n_tasks:
-        max_parallel_tasks = n_tasks
+    max_parallel_tasks = min(max_parallel_tasks, n_tasks)
     logger.info("Running %s tasks using %s processes", n_tasks,
                 max_parallel_tasks)
 
@@ -644,37 +643,39 @@ def _run_tasks_parallel(tasks, max_parallel_tasks=None):
         """Assume a task is done if it not scheduled or running."""
         return not (task in scheduled or task in running)
 
-    pool = Pool(processes=max_parallel_tasks)
-    while scheduled or running:
-        # Submit new tasks to pool
-        for task in sorted(scheduled, key=lambda t: t.priority):
-            if len(running) >= max_parallel_tasks:
-                break
-            if all(done(t) for t in task.ancestors):
-                future = pool.apply_async(_run_task, [task])
-                running[task] = future
-                scheduled.remove(task)
-
-        # Handle completed tasks
-        ready = {t for t in running if running[t].ready()}
-        for task in ready:
-            _copy_results(task, running[task])
-            running.pop(task)
-
-        # Wait if there are still tasks running
-        if running:
-            time.sleep(0.1)
-
-        # Log progress message
-        if len(scheduled) != n_scheduled or len(running) != n_running:
-            n_scheduled, n_running = len(scheduled), len(running)
-            n_done = n_tasks - n_scheduled - n_running
-            logger.info(
-                "Progress: %s tasks running, %s tasks waiting for ancestors, "
-                "%s/%s done", n_running, n_scheduled, n_done, n_tasks)
-
-    pool.close()
-    pool.join()
+    with Pool(processes=max_parallel_tasks) as pool:
+        while scheduled or running:
+            # Submit new tasks to pool
+            for task in sorted(scheduled, key=lambda t: t.priority):
+                if len(running) >= max_parallel_tasks:
+                    break
+                if all(done(t) for t in task.ancestors):
+                    future = pool.apply_async(_run_task, [task])
+                    running[task] = future
+                    scheduled.remove(task)
+
+            # Handle completed tasks
+            ready = {t for t in running if running[t].ready()}
+            for task in ready:
+                _copy_results(task, running[task])
+                running.pop(task)
+
+            # Wait if there are still tasks running
+            if running:
+                time.sleep(0.1)
+
+            # Log progress message
+            if len(scheduled) != n_scheduled or len(running) != n_running:
+                n_scheduled, n_running = len(scheduled), len(running)
+                n_done = n_tasks - n_scheduled - n_running
+                logger.info(
+                    "Progress: %s tasks running, %s tasks waiting for "
+                    "ancestors, %s/%s done", n_running, n_scheduled, n_done,
+                    n_tasks)
+
+        logger.info("Successfully completed all tasks.")
+        pool.close()
+        pool.join()
 
 
 def _copy_results(task, future):