FileObserver setup concurrent-safe

.travis.yml

@@ -14,6 +14,8 @@ matrix:
     env: TOX_ENV=tensorflow-2
   - python: "3.6"
     env: TOX_ENV=flake8
+  - python: "2.7"
+    env: TOX_ENV=coverage
   - python: "3.5"
     env: TOX_ENV=coverage
 script:

sacred/observers/file_storage.py

@@ -4,7 +4,6 @@
 import json
 import os
 import os.path
-import tempfile
 
 from shutil import copyfile
 
@@ -51,16 +50,42 @@ def __init__(self, basedir, resource_dir, source_dir, template,
         self.cout = ""
         self.cout_write_cursor = 0
 
-    def queued_event(self, ex_info, command, host_info, queue_time, config,
-                     meta_info, _id):
-        if not os.path.exists(self.basedir):
-            os.makedirs(self.basedir)
+    @staticmethod
+    def _makedirs(name, mode=0o777, exist_ok=False):
+        """ Wrapper of os.makedirs with fallback
+            for exist_ok on python 2.
+        """
+        try:
+            os.makedirs(name, mode, exist_ok)
+        except TypeError:
+            if not os.path.exists(name):
+                os.makedirs(name, mode)
+
+    def _make_run_dir(self, _id):
+        self._makedirs(self.basedir, exist_ok=True)
         if _id is None:
-            self.dir = tempfile.mkdtemp(prefix='run_', dir=self.basedir)
+            for i in range(200):
+                dir_nrs = [int(d) for d in os.listdir(self.basedir)
+                           if os.path.isdir(os.path.join(self.basedir, d)) and
+                           d.isdigit()]
+                _id = max(dir_nrs + [0]) + 1
+                self.dir = os.path.join(self.basedir, str(_id))
+                try:
+                    os.mkdir(self.dir)
+                    break
+                except FileExistsError:  # Catch race conditions
+                    if i > 100:
+                        # After some tries,
+                        # expect that something other went wrong
+                        raise
         else:
             self.dir = os.path.join(self.basedir, str(_id))
             os.mkdir(self.dir)
 
+    def queued_event(self, ex_info, command, host_info, queue_time, config,
+                     meta_info, _id):
+        self._make_run_dir(_id)
+
         self.run_entry = {
             'experiment': dict(ex_info),
             'command': command,
@@ -91,26 +116,7 @@ def save_sources(self, ex_info):
 
     def started_event(self, ex_info, command, host_info, start_time, config,
                       meta_info, _id):
-        if not os.path.exists(self.basedir):
-            os.makedirs(self.basedir)
-        if _id is None:
-            for i in range(200):
-                dir_nrs = [int(d) for d in os.listdir(self.basedir)
-                           if os.path.isdir(os.path.join(self.basedir, d)) and
-                           d.isdigit()]
-                _id = max(dir_nrs + [0]) + 1
-                self.dir = os.path.join(self.basedir, str(_id))
-                try:
-                    os.mkdir(self.dir)
-                    break
-                except FileExistsError:  # Catch race conditions
-                    if i > 100:
-                        # After some tries,
-                        # expect that something other went wrong
-                        raise
-        else:
-            self.dir = os.path.join(self.basedir, str(_id))
-            os.mkdir(self.dir)
+        self._make_run_dir(_id)
 
         ex_info['sources'] = self.save_sources(ex_info)
 
@@ -137,8 +143,7 @@ def started_event(self, ex_info, command, host_info, start_time, config,
         return os.path.relpath(self.dir, self.basedir) if _id is None else _id
 
     def find_or_save(self, filename, store_dir):
-        if not os.path.exists(store_dir):
-            os.makedirs(store_dir)
+        self._makedirs(store_dir, exist_ok=True)
         source_name, ext = os.path.splitext(os.path.basename(filename))
         md5sum = get_digest(filename)
         store_name = source_name + '_' + md5sum + ext

tests/test_observers/test_file_storage_observer.py

@@ -8,6 +8,10 @@
 from copy import copy
 import pytest
 import json
+try:  # unittest.mock requires python >=3.3
+    from unittest import mock
+except ImportError:
+    mock = None
 
 from sacred.observers.file_storage import FileStorageObserver
 from sacred.serializer import restore
@@ -115,6 +119,16 @@ def test_fs_observer_started_event_creates_rundir(dir_obs, sample_run):
         "status": "RUNNING"
     }
 
+    def mkdir_raises_file_exists(name):
+        raise FileExistsError
+
+    # unittest.mock requires python >=3.3
+    if mock is not None:
+        with pytest.raises(FileExistsError):
+            with mock.patch('os.mkdir', mkdir_raises_file_exists):
+                sample_run['_id'] = None
+                _id = obs.started_event(**sample_run)
+
 
 def test_fs_observer_started_event_stores_source(dir_obs, sample_run, tmpfile):
     basedir, obs = dir_obs