Repository: joblib/joblib

Issue:
not possible to use joblib as background thread
when I try to start parallel loop as background thread I'm getting 

/usr/anaconda3/lib/python3.7/site-packages/joblib/parallel.py:673: UserWarning: Loky-backed parallel loops cannot be nested below threads, setting n_jobs=1
  **self._backend_args)
[Parallel(n_jobs=8)]: Using backend SequentialBackend with 1 concurrent workers.

Attempt to rename thread to 'MainThread' don't help

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=h1) Report
> Merging [#934](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=desc) into [master](https://codecov.io/gh/joblib/joblib/commit/fd585ce36faba338fa1c71949cf29ed8e8f5ea19?src=pr&el=desc) will **decrease** coverage by `21.89%`.
> The diff coverage is `54.34%`.

[![Impacted file tree graph](https://codecov.io/gh/joblib/joblib/pull/934/graphs/tree.svg?width=650&token=gA6LF5DGTW&height=150&src=pr)](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##           master    #934      +/-   ##
=========================================
- Coverage    95.4%   73.5%   -21.9%     
=========================================
  Files          45      45              
  Lines        6460    6492      +32     
=========================================
- Hits         6163    4772    -1391     
- Misses        297    1720    +1423
```


| [Impacted Files](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [joblib/test/test\_parallel.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9wYXJhbGxlbC5weQ==) | `60.16% <27.27%> (-36.81%)` | :arrow_down: |
| [joblib/\_parallel\_backends.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL19wYXJhbGxlbF9iYWNrZW5kcy5weQ==) | `52.17% <62.5%> (-42.23%)` | :arrow_down: |
| [joblib/parallel.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3BhcmFsbGVsLnB5) | `84.61% <87.5%> (-12.67%)` | :arrow_down: |
| [joblib/test/test\_dask.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9kYXNrLnB5) | `4.47% <0%> (-94.03%)` | :arrow_down: |
| [joblib/\_dask.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL19kYXNrLnB5) | `24.29% <0%> (-71.19%)` | :arrow_down: |
| [joblib/\_memory\_helpers.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL19tZW1vcnlfaGVscGVycy5weQ==) | `3.07% <0%> (-70.77%)` | :arrow_down: |
| [joblib/backports.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL2JhY2twb3J0cy5weQ==) | `29.16% <0%> (-64.59%)` | :arrow_down: |
| [joblib/pool.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3Bvb2wucHk=) | `28.44% <0%> (-62.94%)` | :arrow_down: |
| [joblib/executor.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL2V4ZWN1dG9yLnB5) | `37.93% <0%> (-62.07%)` | :arrow_down: |
| [joblib/test/test\_memmapping.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9tZW1tYXBwaW5nLnB5) | `37.46% <0%> (-61.99%)` | :arrow_down: |
| ... and [23 more](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=footer). Last update [fd585ce...2c13ef6](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

codecov[bot]: # [Codecov](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=h1) Report
> Merging [#934](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=desc) into [master](https://codecov.io/gh/joblib/joblib/commit/fd585ce36faba338fa1c71949cf29ed8e8f5ea19?src=pr&el=desc) will **decrease** coverage by `21.89%`.
> The diff coverage is `54.34%`.

[![Impacted file tree graph](https://codecov.io/gh/joblib/joblib/pull/934/graphs/tree.svg?width=650&token=gA6LF5DGTW&height=150&src=pr)](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=tree)

```diff
@@            Coverage Diff            @@
##           master    #934      +/-   ##
=========================================
- Coverage    95.4%   73.5%   -21.9%     
=========================================
  Files          45      45              
  Lines        6460    6492      +32     
=========================================
- Hits         6163    4772    -1391     
- Misses        297    1720    +1423
```


| [Impacted Files](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [joblib/test/test\_parallel.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9wYXJhbGxlbC5weQ==) | `60.16% <27.27%> (-36.81%)` | :arrow_down: |
| [joblib/\_parallel\_backends.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL19wYXJhbGxlbF9iYWNrZW5kcy5weQ==) | `52.17% <62.5%> (-42.23%)` | :arrow_down: |
| [joblib/parallel.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3BhcmFsbGVsLnB5) | `84.61% <87.5%> (-12.67%)` | :arrow_down: |
| [joblib/test/test\_dask.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9kYXNrLnB5) | `4.47% <0%> (-94.03%)` | :arrow_down: |
| [joblib/\_dask.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL19kYXNrLnB5) | `24.29% <0%> (-71.19%)` | :arrow_down: |
| [joblib/\_memory\_helpers.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL19tZW1vcnlfaGVscGVycy5weQ==) | `3.07% <0%> (-70.77%)` | :arrow_down: |
| [joblib/backports.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL2JhY2twb3J0cy5weQ==) | `29.16% <0%> (-64.59%)` | :arrow_down: |
| [joblib/pool.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3Bvb2wucHk=) | `28.44% <0%> (-62.94%)` | :arrow_down: |
| [joblib/executor.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL2V4ZWN1dG9yLnB5) | `37.93% <0%> (-62.07%)` | :arrow_down: |
| [joblib/test/test\_memmapping.py](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9tZW1tYXBwaW5nLnB5) | `37.46% <0%> (-61.99%)` | :arrow_down: |
| ... and [23 more](https://codecov.io/gh/joblib/joblib/pull/934/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=footer). Last update [fd585ce...2c13ef6](https://codecov.io/gh/joblib/joblib/pull/934?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index 4fbd41e..0d12cbb 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -4,6 +4,9 @@ Latest changes
 Release 0.14.0
 --------------
 
+- Add official support for Python 3.8: fixed protocol number in `Hasher`
+  and updated tests.
+
 - Warn users that they should never use `joblib.load` with files from
   untrusted sources. Fix security related API change introduced in numpy
   1.6.3 that would prevent using joblib with recent numpy versions.
@@ -17,6 +20,10 @@ Release 0.14.0
   on module import.
   https://github.com/joblib/joblib/pull/893
 
+- Fix the oversubscription protection to only protect against nested
+  `Parallel` calls. This allows `joblib` to be run in background threads.
+  https://github.com/joblib/joblib/pull/934
+
 Release 0.13.2
 --------------
 
diff --git a/joblib/_parallel_backends.py b/joblib/_parallel_backends.py
index 0f0bcf0..34efffb 100644
--- a/joblib/_parallel_backends.py
+++ b/joblib/_parallel_backends.py
@@ -30,10 +30,10 @@ if mp is not None:
 class ParallelBackendBase(with_metaclass(ABCMeta)):
     """Helper abc which defines all methods a ParallelBackend must implement"""
 
+    nesting_level = None
     supports_timeout = False
-    nesting_level = 0
 
-    def __init__(self, nesting_level=0):
+    def __init__(self, nesting_level=None):
         self.nesting_level = nesting_level
 
     SUPPORTED_CLIB_VARS = [
@@ -160,6 +160,10 @@ class ParallelBackendBase(with_metaclass(ABCMeta)):
             if var_value is None:
                 os.environ[var] = str(n_threads)
 
+    @staticmethod
+    def in_main_thread():
+        return isinstance(threading.current_thread(), threading._MainThread)
+
 
 class SequentialBackend(ParallelBackendBase):
     """A ParallelBackend which will execute all batches sequentially.
@@ -251,9 +255,10 @@ class AutoBatchingMixin(object):
     _DEFAULT_EFFECTIVE_BATCH_SIZE = 1
     _DEFAULT_SMOOTHED_BATCH_DURATION = 0.0
 
-    def __init__(self):
+    def __init__(self, **kwargs):
         self._effective_batch_size = self._DEFAULT_EFFECTIVE_BATCH_SIZE
         self._smoothed_batch_duration = self._DEFAULT_SMOOTHED_BATCH_DURATION
+        super(AutoBatchingMixin, self).__init__(**kwargs)
 
     def compute_batch_size(self):
         """Determine the optimal batch size"""
@@ -409,7 +414,7 @@ class MultiprocessingBackend(PoolManagerMixin, AutoBatchingMixin,
                     stacklevel=3)
             return 1
 
-        if not isinstance(threading.current_thread(), threading._MainThread):
+        elif not (self.in_main_thread() or self.nesting_level == 0):
             # Prevent posix fork inside in non-main posix threads
             if n_jobs != 1:
                 warnings.warn(
@@ -493,7 +498,7 @@ class LokyBackend(AutoBatchingMixin, ParallelBackendBase):
                     ' multiprocessing, setting n_jobs=1',
                     stacklevel=3)
             return 1
-        elif not isinstance(threading.current_thread(), threading._MainThread):
+        elif not (self.in_main_thread() or self.nesting_level == 0):
             # Prevent posix fork inside in non-main posix threads
             if n_jobs != 1:
                 warnings.warn(
diff --git a/joblib/hashing.py b/joblib/hashing.py
index e1f60f0..670bd9d 100644
--- a/joblib/hashing.py
+++ b/joblib/hashing.py
@@ -58,8 +58,7 @@ class Hasher(Pickler):
         self.stream = io.BytesIO()
         # By default we want a pickle protocol that only changes with
         # the major python version and not the minor one
-        protocol = (pickle.DEFAULT_PROTOCOL if PY3_OR_LATER
-                    else pickle.HIGHEST_PROTOCOL)
+        protocol = 3 if PY3_OR_LATER else 2
         Pickler.__init__(self, self.stream, protocol=protocol)
         # Initialise the hash obj
         self._hash = hashlib.new(hash_name)
diff --git a/joblib/parallel.py b/joblib/parallel.py
index 03dcd92..876a448 100644
--- a/joblib/parallel.py
+++ b/joblib/parallel.py
@@ -12,12 +12,10 @@ import sys
 from math import sqrt
 import functools
 import time
-import inspect
 import threading
 import itertools
 from numbers import Integral
 import warnings
-from functools import partial
 
 from ._multiprocessing_helpers import mp
 
@@ -92,11 +90,13 @@ def get_active_backend(prefer=None, require=None, verbose=0):
     if backend_and_jobs is not None:
         # Try to use the backend set by the user with the context manager.
         backend, n_jobs = backend_and_jobs
+        nesting_level = backend.nesting_level
         supports_sharedmem = getattr(backend, 'supports_sharedmem', False)
         if require == 'sharedmem' and not supports_sharedmem:
             # This backend does not match the shared memory constraint:
             # fallback to the default thead-based backend.
-            sharedmem_backend = BACKENDS[DEFAULT_THREAD_BACKEND]()
+            sharedmem_backend = BACKENDS[DEFAULT_THREAD_BACKEND](
+                nesting_level=nesting_level)
             if verbose >= 10:
                 print("Using %s as joblib.Parallel backend instead of %s "
                       "as the latter does not provide shared memory semantics."
@@ -108,14 +108,14 @@ def get_active_backend(prefer=None, require=None, verbose=0):
 
     # We are outside of the scope of any parallel_backend context manager,
     # create the default backend instance now.
-    backend = BACKENDS[DEFAULT_BACKEND]()
+    backend = BACKENDS[DEFAULT_BACKEND](nesting_level=0)
     supports_sharedmem = getattr(backend, 'supports_sharedmem', False)
     uses_threads = getattr(backend, 'uses_threads', False)
     if ((require == 'sharedmem' and not supports_sharedmem) or
             (prefer == 'threads' and not uses_threads)):
         # Make sure the selected default backend match the soft hints and
         # hard constraints:
-        backend = BACKENDS[DEFAULT_THREAD_BACKEND]()
+        backend = BACKENDS[DEFAULT_THREAD_BACKEND](nesting_level=0)
     return backend, DEFAULT_N_JOBS
 
 
@@ -172,7 +172,19 @@ class parallel_backend(object):
 
             backend = BACKENDS[backend](**backend_params)
 
-        self.old_backend_and_jobs = getattr(_backend, 'backend_and_jobs', None)
+        # If the nesting_level of the backend is not set previously, use the
+        # nesting level from the previous active_backend to set it
+        current_backend_and_jobs = getattr(_backend, 'backend_and_jobs', None)
+        if backend.nesting_level is None:
+            if current_backend_and_jobs is None:
+                nesting_level = 0
+            else:
+                nesting_level = current_backend_and_jobs[0].nesting_level
+
+            backend.nesting_level = nesting_level
+
+        # Save the backends info and set the active backend
+        self.old_backend_and_jobs = current_backend_and_jobs
         self.new_backend_and_jobs = (backend, n_jobs)
 
         _backend.backend_and_jobs = (backend, n_jobs)
@@ -587,6 +599,7 @@ class Parallel(Logger):
                  prefer=None, require=None):
         active_backend, context_n_jobs = get_active_backend(
             prefer=prefer, require=require, verbose=verbose)
+        nesting_level = active_backend.nesting_level
         if backend is None and n_jobs is None:
             # If we are under a parallel_backend context manager, look up
             # the default number of jobs and use that instead:
@@ -618,22 +631,26 @@ class Parallel(Logger):
 
         if backend is None:
             backend = active_backend
+
         elif isinstance(backend, ParallelBackendBase):
-            # Use provided backend as is
-            pass
+            # Use provided backend as is, with the current nesting_level if it
+            # is not set yet.
+            if backend.nesting_level is None:
+                backend.nesting_level = nesting_level
+
         elif hasattr(backend, 'Pool') and hasattr(backend, 'Lock'):
             # Make it possible to pass a custom multiprocessing context as
             # backend to change the start method to forkserver or spawn or
             # preload modules on the forkserver helper process.
             self._backend_args['context'] = backend
-            backend = MultiprocessingBackend()
+            backend = MultiprocessingBackend(nesting_level=nesting_level)
         else:
             try:
                 backend_factory = BACKENDS[backend]
             except KeyError:
                 raise ValueError("Invalid backend: %s, expected one of %r"
                                  % (backend, sorted(BACKENDS.keys())))
-            backend = backend_factory()
+            backend = backend_factory(nesting_level=nesting_level)
 
         if (require == 'sharedmem' and
                 not getattr(backend, 'supports_sharedmem', False)):


Test patch:
diff --git a/joblib/test/test_module.py b/joblib/test/test_module.py
index c6a563f..2f04fb5 100644
--- a/joblib/test/test_module.py
+++ b/joblib/test/test_module.py
@@ -26,7 +26,7 @@ def test_no_start_method_side_effect_on_import():
 
 @pytest.mark.skipif(PY27, reason="Need python3.3+")
 def test_no_semaphore_tracker_on_import():
-    # check that importing joblib does not implicitly spawn a ressource tracker
+    # check that importing joblib does not implicitly spawn a resource tracker
     # or a semaphore tracker
     code = """if True:
         import joblib
@@ -35,6 +35,9 @@ def test_no_semaphore_tracker_on_import():
         # start_method is already set.
         msg = "multiprocessing.semaphore_tracker has been spawned on import"
         assert semaphore_tracker._semaphore_tracker._fd is None, msg"""
+    if sys.version_info >= (3, 8):
+        # semaphore_tracker was renamed in Python 3.8:
+        code = code.replace("semaphore_tracker", "resource_tracker")
     check_subprocess_call([sys.executable, '-c', code])
 
 
diff --git a/joblib/test/test_parallel.py b/joblib/test/test_parallel.py
index ff46116..aec5855 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -234,6 +234,24 @@ def test_nested_parallel_warnings(parent_backend, child_backend, expected):
         assert all(res)
 
 
+@with_multiprocessing
+@parametrize('backend', ['loky', 'multiprocessing', 'threading'])
+def test_background_thread_parallelism(backend):
+    is_run_parallel = [False]
+
+    def background_thread(is_run_parallel):
+        with warns(None) as records:
+            Parallel(n_jobs=2)(
+                delayed(sleep)(.1) for _ in range(4))
+        print(len(records))
+        is_run_parallel[0] = len(records) == 0
+
+    t = threading.Thread(target=background_thread, args=(is_run_parallel,))
+    t.start()
+    t.join()
+    assert is_run_parallel[0]
+
+
 def nested_loop(backend):
     Parallel(n_jobs=2, backend=backend)(
         delayed(square)(.01) for _ in range(2))
@@ -731,7 +749,6 @@ def get_nested_pids():
                               for _ in range(2))
 
 
-
 class MyBackend(joblib._parallel_backends.LokyBackend):
     """Backend to test backward compatibility with older backends"""
     def get_nested_backend(self, ):
@@ -787,6 +804,28 @@ def test_nested_backend_in_sequential(backend, n_jobs):
         )
 
 
+def check_nesting_level(inner_backend, expected_level):
+    with parallel_backend(inner_backend) as (backend, n_jobs):
+        assert backend.nesting_level == expected_level
+
+
+@with_multiprocessing
+@pytest.mark.parametrize('outer_backend', PARALLEL_BACKENDS)
+@pytest.mark.parametrize('inner_backend', PARALLEL_BACKENDS)
+def test_backend_nesting_level(outer_backend, inner_backend):
+    # Check that the nesting level for the backend is correctly set
+    check_nesting_level(outer_backend, 0)
+
+    Parallel(n_jobs=2, backend=outer_backend)(
+        delayed(check_nesting_level)(inner_backend, 1)
+        for _ in range(10)
+    )
+
+    with parallel_backend(inner_backend, n_jobs=2):
+        Parallel()(delayed(check_nesting_level)(inner_backend, 1)
+                   for _ in range(10))
+
+
 @with_multiprocessing
 def test_retrieval_context():
     import contextlib