diff --git a/dvc/analytics.py b/dvc/analytics.py index 6643db768..e4b0dc425 100644 --- a/dvc/analytics.py +++ b/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 diff --git a/dvc/lock.py b/dvc/lock.py index 81356a070..40ecc3093 100644 --- a/dvc/lock.py +++ b/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) diff --git a/dvc/logger.py b/dvc/logger.py index f9fe77b8d..cafedd728 100644 --- a/dvc/logger.py +++ b/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"], + }, }, } ) diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index 2f743a6ff..0502adbcc 100644 --- a/dvc/repo/__init__.py +++ b/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): diff --git a/dvc/updater.py b/dvc/updater.py index 008e37ea7..388d8c94b 100644 --- a/dvc/updater.py +++ b/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"): diff --git a/setup.py b/setup.py index 94c3b5153..c80bb2ac2 100644 --- a/setup.py +++ b/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", diff --git a/tests/func/test_lock.py b/tests/func/test_lock.py index 1c5a0bda2..ad04f22ed 100644 --- a/tests/func/test_lock.py +++ b/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)