Repository: joblib/joblib
Pull request #491 (2017-02-03 11:48:20)

[MRG] Fix race condition in Memory when persisting output in parallel
Fixes https://github.com/joblib/joblib/issues/490.

raghavrv: I attempted a fix for this where instead of a temporary pickle, you create a 0-byte `joblib.lock` file in that directory and use `try...finally` to ensure it gets erased upon errors / successful completion... Now while loading the cache, if this `joblib.lock` file exists, it will not utilize the cache and would instead raise a warning and re-compute the results. If the `joblib.lock` file doesn't exist, the current sequence of check the cache, if broken delete and repair, if not utilize the cache gets executed. WDYT?

The problem with blocking operations is sometimes it is faster to compute the results again than to wait for the other thread to complete....
raghavrv: This was the fix that I had in mind but it's far from being elegant or complete. I'm linking it just in case... - https://github.com/raghavrv/scikit-learn/pull/9/commits/119873e4028d585411442ba26707b074ec54149f
jnothman: I think a simple solution, basically the one implemented but with a logarithmic sleep time starting much smaller, is an okay place to start.

However, we have the following risks in behaviours:
* In the parallel case, we want to minimise recalculation if possible, and we may want to minimise reporting "Exception while loading results for ...". This means: cache could be locked even before calculation, but maybe this is an issue for user code (???); allow relatively long waits for dumping; perhaps use more specific heuristics like checking that the process is alive or checking the lock file's timestamp; perhaps soften the error message.
* In the non-parallel case, we want the wait to be as short as possible, so we want easy ways to exit early, e.g. PID file doesn't exist after a very short wait.

To avoid some concurrency issues, one can:
* use a constant filename `output.pid` or `output.lock` which stores the PID rather than putting the PID in the filename.
* not create a directory so that we are only interested in two operations: creation of PID file and moving to end location.

I think in some ways this is closer to @raghavrv's proposal, but I'm not sure I've got a clear understanding of all the differences.
jnothman: (Note that I am using "lock" approximately; I realise true locks are inappropriate.)
GaelVaroquaux: > the one implemented but with a logarithmic sleep time starting much smaller

+1

>     This means: cache could be locked even before calculation,

I worry a lot about lots: they can easily lead to deadlocks and slowdown.

>   • In the non-parallel case, we want the wait to be as short as possible, so
>     we want easy ways to exit early, e.g. PID file doesn't exist after a very
>     short wait.

+1, to be studied.

>   • not create a directory so that we are only interested in two operations: creation of PID file and moving to end location.

Seems about right, but I would need to think more about it :$

GaelVaroquaux: OK, I've given it a bit more thoughts. Something that I think would help, would be to do the writing as follows:

- Create a temporary directory (for instance, the standard directory name, but with  `"tmp%s_" % pid` at the beginning
- Write all the files their
- Rename the directory

The rename is not atomic on all file systems, so this would not be fool proof, but I think that it would help.

It is somewhat additional to what has been discussed here. I just suspect that it would decrease the amount of errors a lot.

ogrisel: Actually there is only one important file which is output.pkl. The input.pkl is just use for debugging (I think). So instead of deleting the directory in case of corruption can just copy the recomputed version of the file. That should remove any race condition (in combination with the log timing strategy in case of empty folder).

Loic is currently working on an update of his PR to implement this.
ogrisel: > I worry a lot about locks: they can easily lead to deadlocks and slowdown.

+1. Also locks are not necessarily available on storage backends we want to support in the future (S3, hdfs...)
jnothman: (I didn't really mean locks. I was lacking terminology.)
Moving the directory seems reasonable.

On 7 Feb 2017 9:48 pm, "Olivier Grisel" <notifications@github.com> wrote:

> I worry a lot about locks: they can easily lead to deadlocks and slowdown.
>
> +1. Also locks are not necessarily available on storage backends we want
> to support in the future (S3, hdfs...)
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/joblib/joblib/pull/491#issuecomment-277963569>, or mute
> the thread
> <https://github.com/notifications/unsubscribe-auth/AAEz6zwepgqf8X1dsvBu9UIZb6Lf3SIeks5raEv3gaJpZM4L2PVj>
> .
>

codecov[bot]: # [Codecov](https://codecov.io/gh/joblib/joblib/pull/491?src=pr&el=h1) Report
> Merging [#491](https://codecov.io/gh/joblib/joblib/pull/491?src=pr&el=desc) into [master](https://codecov.io/gh/joblib/joblib/commit/c60f943374474e8b262d27d9fd3c83e7d118a54b?src=pr&el=desc) will **not impact** coverage by `-0.14%`.


```diff
@@            Coverage Diff             @@
##           master     #491      +/-   ##
==========================================
- Coverage   93.94%   93.81%   -0.14%     
==========================================
  Files          36       36              
  Lines        4575     4607      +32     
==========================================
+ Hits         4298     4322      +24     
- Misses        277      285       +8
```


| [Impacted Files](https://codecov.io/gh/joblib/joblib/pull/491?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [joblib/test/test_memory.py](https://codecov.io/gh/joblib/joblib/compare/c60f943374474e8b262d27d9fd3c83e7d118a54b...8bdafca02a54a67610e2428e7908d19603ba1cb8?src=pr&el=tree#diff-am9ibGliL3Rlc3QvdGVzdF9tZW1vcnkucHk=) | `98.58% <83.33%> (-0.92%)` | :x: |
| [joblib/memory.py](https://codecov.io/gh/joblib/joblib/compare/c60f943374474e8b262d27d9fd3c83e7d118a54b...8bdafca02a54a67610e2428e7908d19603ba1cb8?src=pr&el=tree#diff-am9ibGliL21lbW9yeS5weQ==) | `92.4% <92.3%> (-0.85%)` | :x: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/joblib/joblib/pull/491?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/491?src=pr&el=footer). Last update [c60f943...8bdafca](https://codecov.io/gh/joblib/joblib/compare/c60f943374474e8b262d27d9fd3c83e7d118a54b...8bdafca02a54a67610e2428e7908d19603ba1cb8?src=pr&el=footer&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
lesteve: I left the code as it was [here](https://github.com/lesteve/joblib/blob/b81e1aa6bb305ebac1378d9f1ce1ac7489e3c7c0/joblib/memory.py#L135) because of this [test](https://github.com/lesteve/joblib/blob/b81e1aa6bb305ebac1378d9f1ce1ac7489e3c7c0/joblib/test/test_memory.py#L428). I am not sure this is the right thing to do, comments welcome.

GaelVaroquaux: 
> According to @GaelVaroquaux and @ogrisel IRL it seems like an acceptable choice.

+1 IIL (In Internet Life)


lesteve: I am seeing a problem on Windows, investigating ...
lesteve: I found a way to tackle the problem on Windows. codecov complains mostly because we don't include the coverage from Windows.

@aabadie what was the problem I don't remember exactly ?
aabadie: > what was the problem with enabling codecov on Windows I don't remember exactly ?

Codecov is not launched from the source directory but from C: IIRC. Then codecov doesn't find the coverage file. Maybe we can just pass the right path to codecov to make it work.

Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index 6d38d9a..bec2c17 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -50,6 +50,12 @@ Loïc Estève
     n_jobs=1. See https://github.com/joblib/joblib/issues/483 for more
     details.
 
+Loïc Estève
+
+    Fix race condition when a function cached with
+    ``joblib.Memory.cache`` was used inside a ``joblib.Parallel``. See
+    https://github.com/joblib/joblib/issues/490 for more details.
+
 Release 0.10.3
 --------------
 
diff --git a/joblib/backports.py b/joblib/backports.py
index 74cdf14..7dd3df1 100644
--- a/joblib/backports.py
+++ b/joblib/backports.py
@@ -1,6 +1,10 @@
 """
 Backports of fixes for joblib dependencies
 """
+import os
+import time
+import ctypes
+import sys
 
 from distutils.version import LooseVersion
 
@@ -26,3 +30,51 @@ except ImportError:
         raise NotImplementedError(
             "'joblib.backports.make_memmap' should not be used "
             'if numpy is not installed.')
+
+
+if os.name == 'nt':
+    error_access_denied = 5
+    try:
+        from os import replace
+    except ImportError:
+        # Python 2.7
+        def replace(src, dst):
+            if not isinstance(src, unicode):  # noqa
+                src = unicode(src, sys.getfilesystemencoding())  # noqa
+            if not isinstance(dst, unicode):  # noqa
+                dst = unicode(dst, sys.getfilesystemencoding())  # noqa
+
+            movefile_replace_existing = 0x1
+            return_value = ctypes.windll.kernel32.MoveFileExW(
+                src, dst, movefile_replace_existing)
+            if return_value == 0:
+                raise ctypes.WinError()
+
+    def concurrency_safe_rename(src, dst):
+        """Renames ``src`` into ``dst`` overwriting ``dst`` if it exists.
+
+        On Windows os.replace (or for Python 2.7 its implementation
+        through MoveFileExW) can yield permission errors if executed by
+        two different processes.
+        """
+        max_sleep_time = 1
+        total_sleep_time = 0
+        sleep_time = 0.001
+        while total_sleep_time < max_sleep_time:
+            try:
+                replace(src, dst)
+                break
+            except Exception as exc:
+                if getattr(exc, 'winerror', None) == error_access_denied:
+                    time.sleep(sleep_time)
+                    total_sleep_time += sleep_time
+                    sleep_time *= 2
+                else:
+                    raise
+        else:
+            raise
+else:
+    try:
+        from os import replace as concurrency_safe_rename
+    except ImportError:
+        from os import rename as concurrency_safe_rename  # noqa
diff --git a/joblib/memory.py b/joblib/memory.py
index ac09175..f9ec1df 100644
--- a/joblib/memory.py
+++ b/joblib/memory.py
@@ -15,10 +15,6 @@ import shutil
 import time
 import pydoc
 import re
-try:
-    import cPickle as pickle
-except ImportError:
-    import pickle
 import functools
 import traceback
 import warnings
@@ -29,6 +25,7 @@ import io
 import operator
 import collections
 import datetime
+import threading
 
 # Local imports
 from . import hashing
@@ -39,6 +36,7 @@ from .logger import Logger, format_time, pformat
 from . import numpy_pickle
 from .disk import mkdirp, rm_subdirs, memstr_to_bytes
 from ._compat import _basestring, PY3_OR_LATER
+from .backports import concurrency_safe_rename
 
 FIRST_LINE_TEXT = "# first line:"
 
@@ -203,6 +201,16 @@ def _get_cache_items_to_delete(root_path, bytes_limit):
 
     return cache_items_to_delete
 
+
+def concurrency_safe_write(to_write, filename, write_func):
+    """Writes an object into a file in a concurrency-safe way."""
+    thread_id = id(threading.current_thread())
+    temporary_filename = '{}.thread-{}-pid-{}'.format(
+        filename, thread_id, os.getpid())
+    write_func(to_write, temporary_filename)
+    concurrency_safe_rename(temporary_filename, filename)
+
+
 # An in-memory store to avoid looking at the disk-based function
 # source code to check if a function definition has changed
 _FUNCTION_HASHES = weakref.WeakKeyDictionary()
@@ -489,9 +497,10 @@ class MemorizedFunc(Logger):
         # function code has changed
         output_dir, argument_hash = self._get_output_dir(*args, **kwargs)
         metadata = None
+        output_pickle_path = os.path.join(output_dir, 'output.pkl')
         # FIXME: The statements below should be try/excepted
         if not (self._check_previous_func_code(stacklevel=4) and
-                                 os.path.exists(output_dir)):
+                os.path.isfile(output_pickle_path)):
             if self._verbose > 10:
                 _, name = get_func_name(self.func)
                 self.warn('Computing func %s, argument hash %s in '
@@ -523,7 +532,6 @@ class MemorizedFunc(Logger):
                           '(args=%s, kwargs=%s)\n %s' %
                           (args, kwargs, traceback.format_exc()))
 
-                shutil.rmtree(output_dir, ignore_errors=True)
                 out, metadata = self.call(*args, **kwargs)
                 argument_hash = None
         return (out, argument_hash, metadata)
@@ -748,9 +756,11 @@ class MemorizedFunc(Logger):
         """ Persist the given output tuple in the directory.
         """
         try:
-            mkdirp(dir)
             filename = os.path.join(dir, 'output.pkl')
-            numpy_pickle.dump(output, filename, compress=self.compress)
+            mkdirp(dir)
+            write_func = functools.partial(numpy_pickle.dump,
+                                           compress=self.compress)
+            concurrency_safe_write(output, filename, write_func)
             if self._verbose > 10:
                 print('Persisting in %s' % dir)
         except OSError:
@@ -784,9 +794,14 @@ class MemorizedFunc(Logger):
         metadata = {"duration": duration, "input_args": input_repr}
         try:
             mkdirp(output_dir)
-            with open(os.path.join(output_dir, 'metadata.json'), 'w') as f:
-                json.dump(metadata, f)
-        except:
+            filename = os.path.join(output_dir, 'metadata.json')
+
+            def write_func(output, dest_filename):
+                with open(dest_filename, 'w') as f:
+                    json.dump(output, f)
+
+            concurrency_safe_write(metadata, filename, write_func)
+        except Exception:
             pass
 
         this_duration = time.time() - start_time
diff --git a/joblib/pool.py b/joblib/pool.py
index 0867782..c53a12d 100644
--- a/joblib/pool.py
+++ b/joblib/pool.py
@@ -28,7 +28,7 @@ from time import sleep
 try:
     WindowsError
 except NameError:
-    WindowsError = None
+    WindowsError = type(None)
 
 from pickle import whichmodule
 try:
@@ -605,11 +605,12 @@ class MemmapingPool(PicklingPool):
             try:
                 super(MemmapingPool, self).terminate()
                 break
-            except WindowsError as e:
-                # Workaround  occasional "[Error 5] Access is denied" issue
-                # when trying to terminate a process under windows.
-                sleep(0.1)
-                if i + 1 == n_retries:
-                    warnings.warn("Failed to terminate worker processes in "
-                                  " multiprocessing pool: %r" % e)
+            except OSError as e:
+                if isinstance(e, WindowsError):
+                    # Workaround  occasional "[Error 5] Access is denied" issue
+                    # when trying to terminate a process under windows.
+                    sleep(0.1)
+                    if i + 1 == n_retries:
+                        warnings.warn("Failed to terminate worker processes in"
+                                      " multiprocessing pool: %r" % e)
         delete_folder(self._temp_folder)


Test patch:
diff --git a/joblib/test/test_backports.py b/joblib/test/test_backports.py
index 0199d02..de90cc5 100644
--- a/joblib/test/test_backports.py
+++ b/joblib/test/test_backports.py
@@ -1,7 +1,8 @@
 import mmap
 
-from joblib.backports import make_memmap
+from joblib.backports import make_memmap, concurrency_safe_rename
 from joblib.test.common import with_numpy
+from joblib.testing import parametrize
 
 
 @with_numpy
@@ -11,3 +12,17 @@ def test_memmap(tmpdir):
     offset = mmap.ALLOCATIONGRANULARITY + 1
     memmap_obj = make_memmap(fname, shape=size, mode='w+', offset=offset)
     assert memmap_obj.offset == offset
+
+
+@parametrize('src_content', [None, 'src content'])
+def test_concurrency_safe_rename(tmpdir, src_content):
+    src_path = tmpdir.join('src')
+    src_path.write('src content')
+    dst_path = tmpdir.join('dst')
+    if src_content is not None:
+        dst_path.write('dst content')
+
+    concurrency_safe_rename(src_path.strpath, dst_path.strpath)
+    assert not src_path.exists()
+    assert dst_path.exists()
+    assert dst_path.read() == 'src content'
diff --git a/joblib/test/test_memory.py b/joblib/test/test_memory.py
index bc5c580..e10e15b 100644
--- a/joblib/test/test_memory.py
+++ b/joblib/test/test_memory.py
@@ -9,17 +9,28 @@ Test the memory module.
 import shutil
 import os
 import os.path
-import pickle
 import sys
 import time
 import datetime
+import pickle
+try:
+    # Python 2.7: use the C pickle to speed up
+    # test_concurrency_safe_write which pickles big python objects
+    import cPickle as cpickle
+except ImportError:
+    import pickle as cpickle
+import functools
+
 
 from joblib.memory import Memory, MemorizedFunc, NotMemorizedFunc
 from joblib.memory import MemorizedResult, NotMemorizedResult, _FUNCTION_HASHES
 from joblib.memory import _get_cache_items, _get_cache_items_to_delete
 from joblib.memory import _load_output, _get_func_fullname
 from joblib.memory import JobLibCollisionWarning
+from joblib.memory import concurrency_safe_write
+from joblib.parallel import Parallel, delayed
 from joblib.test.common import with_numpy, np
+from joblib.test.common import with_multiprocessing
 from joblib.testing import parametrize, raises, warns
 from joblib._compat import PY3_OR_LATER
 
@@ -722,3 +733,89 @@ def test_memory_clear(tmpdir):
     memory.clear()
 
     assert os.listdir(memory.cachedir) == []
+
+
+def fast_func_with_complex_output():
+    complex_obj = ['a' * 1000] * 1000
+    return complex_obj
+
+
+def fast_func_with_conditional_complex_output(complex_output=True):
+    complex_obj = {str(i): i for i in range(int(1e5))}
+    return complex_obj if complex_output else 'simple output'
+
+
+@with_multiprocessing
+def test_cached_function_race_condition_when_persisting_output(tmpdir, capfd):
+    # Test race condition where multiple processes are writing into
+    # the same output.pkl. See
+    # https://github.com/joblib/joblib/issues/490 for more details.
+    memory = Memory(cachedir=tmpdir.strpath)
+    func_cached = memory.cache(fast_func_with_complex_output)
+
+    Parallel(n_jobs=2)(delayed(func_cached)() for i in range(3))
+
+    stdout, stderr = capfd.readouterr()
+
+    # Checking both stdout and stderr (ongoing PR #434 may change
+    # logging destination) to make sure there is no exception while
+    # loading the results
+    exception_msg = 'Exception while loading results'
+    assert exception_msg not in stdout
+    assert exception_msg not in stderr
+
+
+@with_multiprocessing
+def test_cached_function_race_condition_when_persisting_output_2(tmpdir,
+                                                                 capfd):
+    # Test race condition in first attempt at solving
+    # https://github.com/joblib/joblib/issues/490. The race condition
+    # was due to the delay between seeing the cache directory created
+    # (interpreted as the result being cached) and the output.pkl being
+    # pickled.
+    memory = Memory(cachedir=tmpdir.strpath)
+    func_cached = memory.cache(fast_func_with_conditional_complex_output)
+
+    Parallel(n_jobs=2)(delayed(func_cached)(True if i % 2 == 0 else False)
+                       for i in range(3))
+
+    stdout, stderr = capfd.readouterr()
+
+    # Checking both stdout and stderr (ongoing PR #434 may change
+    # logging destination) to make sure there is no exception while
+    # loading the results
+    exception_msg = 'Exception while loading results'
+    assert exception_msg not in stdout
+    assert exception_msg not in stderr
+
+
+def write_func(output, filename):
+    with open(filename, 'wb') as f:
+        cpickle.dump(output, f)
+
+
+def load_func(expected, filename):
+    for i in range(10):
+        try:
+            with open(filename, 'rb') as f:
+                reloaded = cpickle.load(f)
+            break
+        except OSError:
+            # On Windows you can have WindowsError ([Error 5] Access
+            # is denied) when reading the file, probably because a
+            # writer process has a lock on the file
+            time.sleep(0.1)
+    else:
+        raise
+    assert expected == reloaded
+
+
+@with_multiprocessing
+@parametrize('backend', ['multiprocessing', 'threading'])
+def test_concurrency_safe_write(tmpdir, backend):
+    filename = tmpdir.join('test.pkl').strpath
+    obj = {str(i): i for i in range(int(1e5))}
+    funcs = [functools.partial(concurrency_safe_write, write_func=write_func)
+             if i % 3 != 2 else load_func for i in range(12)]
+    Parallel(n_jobs=2, backend=backend)(
+        delayed(func)(obj, filename) for func in funcs)