dvc: switch to flufl lock

This locks employs hardlink trick, which works on NFS and alike unlike
fcntl/open locks.

Closes #1918.

dvc/analytics.py

@@ -66,7 +66,7 @@ class Analytics(object):
                 raise
 
         self.user_id_file = os.path.join(cdir, self.USER_ID_FILE)
-        self.user_id_file_lock = Lock(cdir, self.USER_ID_FILE + ".lock")
+        self.user_id_file_lock = Lock(self.user_id_file + ".lock")
 
     @staticmethod
     def load(path):
@@ -113,7 +113,7 @@ class Analytics(object):
                 return user_id
         except LockError:
             msg = "Failed to acquire '{}'"
-            logger.debug(msg.format(self.user_id_file_lock.lock_file))
+            logger.debug(msg.format(self.user_id_file_lock.lockfile))
 
     def _collect_windows(self):
         import sys

dvc/lock.py

@@ -3,58 +3,59 @@
 from __future__ import unicode_literals
 
 import os
-import time
-import zc.lockfile
+from datetime import timedelta
+from flufl.lock import Lock as _Lock, TimeOutError
 
 from dvc.exceptions import DvcException
+from dvc.utils import makedirs
 
 
 class LockError(DvcException):
     """Thrown when unable to acquire the lock for dvc repo."""
 
 
-class Lock(object):
+class Lock(_Lock):
     """Class for dvc repo lock.
 
     Args:
-        dvc_dir (str): path to the directory that the lock should be created
+        lockfile (str): the lock filename
             in.
-        name (str): name of the lock file.
+        lifetime (int | timedelta): hold the lock for so long.
+        tmp_dir (str): a directory to store claim files.
     """
 
-    LOCK_FILE = "lock"
     TIMEOUT = 5
 
-    def __init__(self, dvc_dir, name=LOCK_FILE):
-        self.lock_file = os.path.join(dvc_dir, name)
-        self._lock = None
+    def __init__(self, lockfile, lifetime=None, tmp_dir=None):
+        if isinstance(lifetime, int):
+            lifetime = timedelta(seconds=lifetime)
 
-    def _do_lock(self):
+        self._tmp_dir = tmp_dir
+        if self._tmp_dir is not None:
+            makedirs(self._tmp_dir, exist_ok=True)
+
+        super(Lock, self).__init__(lockfile, lifetime=lifetime)
+
+    @property
+    def lockfile(self):
+        return self._lockfile
+
+    @property
+    def tmp_dir(self):
+        return self._tmp_dir
+
+    def lock(self):
         try:
-            self._lock = zc.lockfile.LockFile(self.lock_file)
-        except zc.lockfile.LockError:
+            super(Lock, self).lock(timedelta(seconds=self.TIMEOUT))
+        except TimeOutError:
             raise LockError(
                 "cannot perform the cmd since DVC is busy and "
                 "locked. Please retry the cmd later."
             )
 
-    def lock(self):
-        """Acquire lock for dvc repo."""
-        try:
-            self._do_lock()
-            return
-        except LockError:
-            time.sleep(self.TIMEOUT)
+    def _set_claimfile(self, pid=None):
+        super(Lock, self)._set_claimfile(pid)
 
-        self._do_lock()
-
-    def unlock(self):
-        """Release lock for dvc repo."""
-        self._lock.close()
-        self._lock = None
-
-    def __enter__(self):
-        self.lock()
-
-    def __exit__(self, typ, value, tbck):
-        self.unlock()
+        if self._tmp_dir is not None:
+            filename = self._claimfile.replace(os.sep, "_")
+            self._claimfile = os.path.join(self._tmp_dir, filename)

dvc/logger.py

@@ -196,6 +196,10 @@ def setup(level=logging.INFO):
                     "level": logging.CRITICAL,
                     "handlers": ["console", "console_errors"],
                 },
+                "flufl.lock": {
+                    "level": logging.CRITICAL,
+                    "handlers": ["console", "console_errors"],
+                },
             },
         }
     )

dvc/repo/__init__.py

@@ -70,7 +70,12 @@ class Repo(object):
 
         self.tree = WorkingTree(self.root_dir)
 
-        self.lock = Lock(self.dvc_dir)
+        self.lock = Lock(
+            os.path.join(self.dvc_dir, "lock"),
+            # This should be longer than operations we are protecting
+            lifetime=60 * 60 * 24 * 30,
+            tmp_dir=os.path.join(self.dvc_dir, "tmp"),
+        )
         # NOTE: storing state and link_state in the repository itself to avoid
         # any possible state corruption in 'shared cache dir' scenario.
         self.state = State(self, self.config.config)
@@ -129,10 +134,11 @@ class Repo(object):
         updater = Updater(self.dvc_dir)
 
         flist = [
-            self.lock.lock_file,
+            self.lock.lockfile,
+            self.lock.tmp_dir,
             self.config.config_local_file,
             updater.updater_file,
-            updater.lock.lock_file,
+            updater.lock.lockfile,
         ] + self.state.files
 
         if self.cache.local.cache_dir.startswith(self.root_dir):

dvc/updater.py

@@ -25,7 +25,9 @@ class Updater(object):  # pragma: no cover
     def __init__(self, dvc_dir):
         self.dvc_dir = dvc_dir
         self.updater_file = os.path.join(dvc_dir, self.UPDATER_FILE)
-        self.lock = Lock(dvc_dir, self.updater_file + ".lock")
+        self.lock = Lock(
+            self.updater_file + ".lock", tmp_dir=os.path.join(dvc_dir, "tmp")
+        )
         self.current = parse_version(__version__).base_version
 
     def _is_outdated_file(self):
@@ -41,7 +43,7 @@ class Updater(object):  # pragma: no cover
                 func()
         except LockError:
             msg = "Failed to acquire '{}' before {} updates"
-            logger.debug(msg.format(self.lock.lock_file, action))
+            logger.debug(msg.format(self.lock.lockfile, action))
 
     def check(self):
         if os.getenv("CI") or os.getenv("DVC_TEST"):

setup.py

@@ -39,7 +39,6 @@ class build_py(_build_py):
 install_requires = [
     "ply>=3.9",  # See https://github.com/pyinstaller/pyinstaller/issues/1945
     "configparser>=3.5.0",
-    "zc.lockfile>=1.2.1",
     "future>=0.16.0",
     "colorama>=0.3.9",
     "configobj>=5.0.6",
@@ -139,7 +138,8 @@ setup(
         "ssh_gssapi": ssh_gssapi,
         "hdfs": hdfs,
         # NOTE: https://github.com/inveniosoftware/troubleshooting/issues/1
-        ":python_version=='2.7'": ["futures", "pathlib2"],
+        ":python_version=='2.7'": ["futures", "pathlib2", "flufl.lock==2.4.1"],
+        ":python_version>='3.0'": ["flufl.lock>=3.2"],
         "tests": tests_requirements,
     },
     keywords="data science, data version control, machine learning",

tests/func/test_lock.py

@@ -1,3 +1,5 @@
+import os
+
 from dvc.lock import LockError
 from dvc.main import main
 from dvc.lock import Lock
@@ -7,15 +9,17 @@ from tests.basic_env import TestDvc
 
 class TestLock(TestDvc):
     def test_with(self):
-        lock = Lock(self.dvc.dvc_dir)
+        lockfile = os.path.join(self.dvc.dvc_dir, "lock")
+        lock = Lock(lockfile)
         with lock:
             with self.assertRaises(LockError):
-                lock2 = Lock(self.dvc.dvc_dir)
+                lock2 = Lock(lockfile)
                 with lock2:
                     self.assertTrue(False)
 
     def test_cli(self):
-        lock = Lock(self.dvc.dvc_dir)
+        lockfile = os.path.join(self.dvc.dvc_dir, "lock")
+        lock = Lock(lockfile)
         with lock:
             ret = main(["add", self.FOO])
             self.assertEqual(ret, 1)