diff --git a/dvc/command/base.py b/dvc/command/base.py index ae06ba78d..1cd76bd9f 100644 --- a/dvc/command/base.py +++ b/dvc/command/base.py @@ -1,11 +1,41 @@ import argparse import os +import fasteners from dvc.config import ConfigError +from dvc.exceptions import DvcException from dvc.logger import Logger from dvc.system import System +class CmdBaseError(DvcException): + def __init__(self, msg): + super(CmdBaseError, self).__init__('{}'.format(msg)) + + +class DvcLockerError(CmdBaseError): + def __init__(self, msg): + super(DvcLockerError, self).__init__('DVC locker error: {}'.format(msg)) + + +class DvcLock(object): + def __init__(self, is_locker, git): + self.is_locker = is_locker + self.git = git + self.lock = None + + def __enter__(self): + if self.is_locker: + self.lock = fasteners.InterProcessLock(self.git.lock_file) + if not self.lock.acquire(timeout=5): + raise DvcLockerError('Cannot perform the cmd since DVC is busy and locked. Please retry the cmd later.') + return self.lock + + def __exit__(self, type, value, traceback): + if self.is_locker: + self.lock.release() + + class CmdBase(object): def __init__(self, settings): self._settings = settings diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index 8ce226c22..cc01009f1 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -2,12 +2,10 @@ import base64 import hashlib import os -import fasteners from boto.s3.connection import S3Connection from google.cloud import storage as gc -from dvc.command.base import CmdBase -from dvc.command.init import CmdInit +from dvc.command.base import CmdBase, DvcLock from dvc.logger import Logger from dvc.exceptions import DvcException from dvc.runtime import Runtime @@ -16,7 +14,7 @@ from dvc.system import System class DataSyncError(DvcException): def __init__(self, msg): - DvcException.__init__(self, 'Data sync error: {}'.format(msg)) + super(DataSyncError, self).__init__('Data sync error: {}'.format(msg)) def sizeof_fmt(num, suffix='B'): @@ -51,22 +49,9 @@ class CmdDataSync(CmdBase): nargs='*') def run(self): - if self.is_locker: - lock = fasteners.InterProcessLock(self.git.lock_file) - gotten = lock.acquire(timeout=5) - if not gotten: - Logger.info('Cannot perform the cmd since DVC is busy and locked. Please retry the cmd later.') - return 1 + with DvcLock(self.is_locker, self.git): + self.sync_sanity_check() - good = self.config.sanity_check() - - if not good[0]: - Logger.error('config \'%s\' is not correctly setup. Please fix:' % Runtime.CONFIG) - for e in good[1]: - Logger.error(' ' + e) - return 1 - - try: for target in self.parsed_args.targets: data_item = self.settings.path_factory.data_item(target) if System.islink(target): @@ -75,27 +60,36 @@ class CmdDataSync(CmdBase): self.sync_dir(target) else: raise DataSyncError('File "{}" does not exit'.format(target)) - finally: - if self.is_locker: - lock.release() + pass + + def sync_sanity_check(self): + good = self.config.sanity_check() + if not good[0]: + error = 'config \'%s\' is not correctly setup. Please fix:\n' % Runtime.CONFIG + for e in good[1]: + error += ' \n' + e + raise DataSyncError(error) def sync_dir(self, dir): for file in os.listdir(dir): try: fname = None fname = os.path.join(dir, file) - if os.path.isdir(fname): - self.sync_dir(fname) - elif System.islink(fname): - self.sync_symlink(self.settings.path_factory.data_item(fname)) - else: - raise DvcException('Unsupported file type "{}"'.format(fname)) + self.sync_object(fname) except DataSyncError as ex: Logger.debug(ex) except Exception as ex: Logger.error('Cannot sync file {}: {}'.format(fname, ex)) pass + def sync_object(self, fname): + if os.path.isdir(fname): + self.sync_dir(fname) + elif System.islink(fname): + self.sync_symlink(self.settings.path_factory.data_item(fname)) + else: + raise DvcException('Unsupported file type "{}"'.format(fname)) + def sync_symlink(self, data_item): if os.path.isfile(data_item.resolved_cache.dvc): self.sync_to_cloud(data_item) @@ -157,10 +151,7 @@ class CmdDataSync(CmdBase): return bucket def _sync_from_cloud_gcp(self, item): - """ sync from cloud, gcp version - Params: - TODO - """ + """ sync from cloud, gcp version """ bucket = self._get_bucket_gc() key = self.cache_file_key(item.resolved_cache.dvc) diff --git a/dvc/command/import_bulk.py b/dvc/command/import_bulk.py index 1f8df7b7d..88ea39f2a 100644 --- a/dvc/command/import_bulk.py +++ b/dvc/command/import_bulk.py @@ -1,9 +1,7 @@ import os -import fasteners -from dvc.command.base import CmdBase +from dvc.command.base import CmdBase, DvcLock from dvc.command.import_file import CmdImportFile -from dvc.logger import Logger from dvc.runtime import Runtime @@ -25,14 +23,7 @@ class CmdImportBulk(CmdBase): pass def run(self): - if self.is_locker: - lock = fasteners.InterProcessLock(self.git.lock_file) - gotten = lock.acquire(timeout=5) - if not gotten: - Logger.info('Cannot perform the cmd since DVC is busy and locked. Please retry the cmd later.') - return 1 - - try: + with DvcLock(self.is_locker, self.git): if not self.no_git_actions and not self.git.is_ready_to_go(): return 1 @@ -59,9 +50,6 @@ class CmdImportBulk(CmdBase): cmd.import_and_commit_if_needed(filename, out, self.parsed_args.lock) pass - finally: - if self.is_locker: - lock.release() pass if __name__ == '__main__': diff --git a/dvc/command/import_file.py b/dvc/command/import_file.py index cf18afe4f..d031f0d8c 100644 --- a/dvc/command/import_file.py +++ b/dvc/command/import_file.py @@ -1,10 +1,9 @@ import os from shutil import copyfile import re -import fasteners import requests -from dvc.command.base import CmdBase +from dvc.command.base import CmdBase, DvcLock from dvc.command.data_sync import sizeof_fmt from dvc.logger import Logger from dvc.exceptions import DvcException @@ -30,20 +29,10 @@ class CmdImportFile(CmdBase): pass def run(self): - if self.is_locker: - lock = fasteners.InterProcessLock(self.git.lock_file) - gotten = lock.acquire(timeout=5) - if not gotten: - Logger.info('Cannot perform the cmd since DVC is busy and locked. Please retry the cmd later.') - return 1 - - try: + with DvcLock(self.is_locker, self.git): return self.import_and_commit_if_needed(self.parsed_args.input, self.parsed_args.output, self.parsed_args.lock) - finally: - if self.is_locker: - lock.release() pass def import_and_commit_if_needed(self, input, output, lock=False, check_if_ready=True): diff --git a/dvc/command/lock.py b/dvc/command/lock.py index f3da2d2cb..587245b02 100644 --- a/dvc/command/lock.py +++ b/dvc/command/lock.py @@ -1,5 +1,4 @@ -import fasteners -from dvc.command.base import CmdBase +from dvc.command.base import CmdBase, DvcLock from dvc.logger import Logger from dvc.runtime import Runtime from dvc.state_file import StateFile @@ -19,19 +18,8 @@ class CmdLock(CmdBase): pass def run(self): - if self.is_locker: - lock = fasteners.InterProcessLock(self.git.lock_file) - gotten = lock.acquire(timeout=5) - if not gotten: - self.warning_dvc_is_busy() - return 1 - - try: - + with DvcLock(self.is_locker, self.git): return self.lock_files(self.parsed_args.files, not self.parsed_args.unlock) - finally: - if self.is_locker: - lock.release() def lock_files(self, files, target): cmd = 'lock' if target else 'unlock' diff --git a/dvc/command/remove.py b/dvc/command/remove.py index a9b9b4ec9..2b23bb926 100644 --- a/dvc/command/remove.py +++ b/dvc/command/remove.py @@ -1,9 +1,8 @@ import os -import fasteners from boto.s3.connection import S3Connection -from dvc.command.base import CmdBase +from dvc.command.base import CmdBase, DvcLock from dvc.exceptions import DvcException from dvc.logger import Logger from dvc.runtime import Runtime @@ -30,19 +29,9 @@ class CmdDataRemove(CmdBase): pass def run(self): - if self.is_locker: - lock = fasteners.InterProcessLock(self.git.lock_file) - gotten = lock.acquire(timeout=5) - if not gotten: - Logger.info('[Cmd-Remove] Cannot perform the cmd since DVC is busy and locked. Please retry the cmd later.') - return 1 - - try: + with DvcLock(self.is_locker, self.git): if not self.remove_all_targets(): return 1 - finally: - if self.is_locker: - lock.release() return 0 diff --git a/dvc/command/repro.py b/dvc/command/repro.py index ffe6f720e..92ba6bd0e 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -1,7 +1,7 @@ import os -import fasteners import copy +from dvc.command.base import DvcLock from dvc.command.import_file import CmdImportFile from dvc.command.run import CmdRun from dvc.logger import Logger @@ -51,18 +51,9 @@ class CmdRepro(CmdRun): return [] def run(self): - if self.is_locker: - lock = fasteners.InterProcessLock(self.git.lock_file) - gotten = lock.acquire(timeout=5) - if not gotten: - self.warning_dvc_is_busy() - return 1 - try: + with DvcLock(self.is_locker, self.git): recursive = not self.parsed_args.single_item return self.repro_target(self.parsed_args.target, recursive, self.parsed_args.force) - finally: - if self.is_locker: - lock.release() pass def repro_target(self, target, recursive, force): diff --git a/dvc/command/run.py b/dvc/command/run.py index cb9685ae1..1c3bf5107 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -1,7 +1,6 @@ import os -import fasteners -from dvc.command.base import CmdBase +from dvc.command.base import CmdBase, DvcLock from dvc.exceptions import DvcException from dvc.git_wrapper import GitWrapper from dvc.logger import Logger @@ -53,14 +52,7 @@ class CmdRun(CmdBase): return self._data_items_from_params(self.parsed_args.output, 'Output') def run(self): - if self.is_locker: - lock = fasteners.InterProcessLock(self.git.lock_file) - gotten = lock.acquire(timeout=5) - if not gotten: - self.warning_dvc_is_busy() - return 1 - - try: + with DvcLock(self.is_locker, self.git): data_items_from_args, not_data_items_from_args = self.argv_files_by_type(self.command_args) return self.run_and_commit_if_needed(self.command_args, data_items_from_args, @@ -68,9 +60,7 @@ class CmdRun(CmdBase): self.parsed_args.stdout, self.parsed_args.stderr, self.parsed_args.shell) - finally: - if self.is_locker: - lock.release() + pass def run_and_commit_if_needed(self, command_args, data_items_from_args, not_data_items_from_args, stdout, stderr, shell, check_if_ready=True): diff --git a/dvc/main.py b/dvc/main.py index 1b19675c4..16a32d81e 100644 --- a/dvc/main.py +++ b/dvc/main.py @@ -18,7 +18,7 @@ from dvc.command.import_bulk import CmdImportBulk from dvc.command.lock import CmdLock from dvc.command.test import CmdTest -VERSION = '0.8.3' +VERSION = '0.8.4' def print_usage(): usage = ('',