dvc: bump to 0.77.2

remote: make `copy` links atomic for local/hdfs/ssh

dvc/command/checkout.py

@@ -12,6 +12,7 @@ class CmdCheckout(CmdBase):
             targets=self.args.targets,
             with_deps=self.args.with_deps,
             force=self.args.force,
+            relink=self.args.relink,
             recursive=self.args.recursive,
         )
         return 0
@@ -34,6 +35,12 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Do not prompt when removing working directory files.",
     )
+    checkout_parser.add_argument(
+        "--relink",
+        action="store_true",
+        default=False,
+        help="Recreate links or copies from cache to workspace.",
+    )
     checkout_parser.add_argument(
         "-d",
         "--with-deps",

dvc/command/config.py

@@ -29,6 +29,14 @@ class CmdConfig(CmdBaseNoRepo):
                 section, opt, self.args.value, level=self.args.level
             )
 
+        is_write = self.args.unset or self.args.value is not None
+        if is_write and self.args.name == "cache.type":
+            logger.warning(
+                "You have changed the 'cache.type' option. This doesn't update"
+                " any existing workspace file links, but it can be done with:"
+                "\n             dvc checkout --relink"
+            )
+
         return 0
 
 

dvc/command/daemon.py

@@ -17,7 +17,7 @@ class CmdDaemonUpdater(CmdDaemonBase):
 
         root_dir = Repo.find_root()
         dvc_dir = os.path.join(root_dir, Repo.DVC_DIR)
-        config = Config(dvc_dir, verify=False)
+        config = Config(dvc_dir, validate=False)
         hardlink_lock = config.config.get("core", {}).get(
             "hardlink_lock", False
         )

dvc/config.py

@@ -173,6 +173,10 @@ class Config(object):  # pylint: disable=too-many-instance-attributes
     SECTION_AWS_USE_SSL = "use_ssl"
     SECTION_AWS_SSE = "sse"
     SECTION_AWS_ACL = "acl"
+    SECTION_AWS_GRANT_READ = "grant_read"
+    SECTION_AWS_GRANT_READ_ACP = "grant_read_acp"
+    SECTION_AWS_GRANT_WRITE_ACP = "grant_write_acp"
+    SECTION_AWS_GRANT_FULL_CONTROL = "grant_full_control"
 
     # gcp specific options
     SECTION_GCP_CREDENTIALPATH = CREDENTIALPATH
@@ -212,6 +216,10 @@ class Config(object):  # pylint: disable=too-many-instance-attributes
         Optional(SECTION_AWS_USE_SSL, default=True): Bool,
         SECTION_AWS_SSE: str,
         SECTION_AWS_ACL: str,
+        SECTION_AWS_GRANT_READ: str,
+        SECTION_AWS_GRANT_READ_ACP: str,
+        SECTION_AWS_GRANT_WRITE_ACP: str,
+        SECTION_AWS_GRANT_FULL_CONTROL: str,
         SECTION_GCP_PROJECTNAME: str,
         SECTION_CACHE_TYPE: supported_cache_type,
         Optional(SECTION_CACHE_PROTECTED, default=False): Bool,

dvc/exceptions.py

@@ -47,7 +47,8 @@ class OutputDuplicationError(DvcException):
         )
         msg = (
             "file/directory '{}' is specified as an output in more than one "
-            "stage: {}"
+            "stage: {}\n"
+            "This is not allowed. Consider using a different output name."
         ).format(output, "\n    ".join(stages))
         super(OutputDuplicationError, self).__init__(msg)
 

dvc/lock.py

@@ -7,10 +7,9 @@ import time
 from datetime import timedelta
 
 import zc.lockfile
-from funcy.py3 import lkeep
 
 from dvc.exceptions import DvcException
-from dvc.utils import makedirs, format_link
+from dvc.utils import format_link
 from dvc.utils.compat import is_py3, is_py2
 from dvc.progress import Tqdm
 
@@ -93,8 +92,6 @@ if is_py3:
             import socket
 
             self._tmp_dir = tmp_dir
-            if self._tmp_dir is not None:
-                makedirs(self._tmp_dir, exist_ok=True)
 
             # NOTE: this is basically Lock.__init__ copy-paste, except that
             # instead of using `socket.getfqdn()` we use `socket.gethostname()`
@@ -120,10 +117,6 @@ if is_py3:
         def lockfile(self):
             return self._lockfile
 
-        @property
-        def files(self):
-            return lkeep([self._lockfile, self._tmp_dir])
-
         def lock(self):
             try:
                 super(Lock, self).lock(timedelta(seconds=DEFAULT_TIMEOUT))

dvc/output/base.py

@@ -282,7 +282,9 @@ class OutputBase(object):
     def download(self, to):
         self.remote.download(self.path_info, to.path_info)
 
-    def checkout(self, force=False, progress_callback=None, tag=None):
+    def checkout(
+        self, force=False, progress_callback=None, tag=None, relink=False
+    ):
         if not self.use_cache:
             if progress_callback:
                 progress_callback(str(self.path_info), self.get_files_number())
@@ -298,6 +300,7 @@ class OutputBase(object):
             info,
             force=force,
             progress_callback=progress_callback,
+            relink=relink,
         )
 
     def remove(self, ignore_remove=False):

dvc/path_info.py

@@ -23,7 +23,16 @@ if is_py2:
     fs_encoding = "utf-8"
 
 
-class PathInfo(pathlib.PurePath):
+class _BasePath(object):
+    def overlaps(self, other):
+        if isinstance(other, basestring):
+            other = self.__class__(other)
+        elif self.__class__ != other.__class__:
+            return False
+        return self == other or self.isin(other) or other.isin(self)
+
+
+class PathInfo(pathlib.PurePath, _BasePath):
     # Use __slots__ in PathInfo objects following PurePath implementation.
     # This makes objects smaller and speeds up attribute access.
     # We don't add any fields so it's empty.
@@ -138,7 +147,7 @@ class _URLPathParents(object):
         return "<{}.parents>".format(self.src)
 
 
-class URLInfo(object):
+class URLInfo(_BasePath):
     DEFAULT_PORTS = {"http": 80, "https": 443, "ssh": 22, "hdfs": 0}
 
     def __init__(self, url):

dvc/remote/base.py

@@ -1,4 +1,5 @@
 from __future__ import unicode_literals
+from dvc.utils.compat import basestring, FileNotFoundError, str, urlparse
 
 import itertools
 import json
@@ -14,22 +15,17 @@ from shortuuid import uuid
 
 import dvc.prompt as prompt
 from dvc.config import Config
-from dvc.exceptions import ConfirmRemoveError
-from dvc.exceptions import DvcException
-from dvc.exceptions import DvcIgnoreInCollectedDirError
+from dvc.exceptions import (
+    DvcException,
+    ConfirmRemoveError,
+    DvcIgnoreInCollectedDirError,
+)
 from dvc.ignore import DvcIgnore
-from dvc.path_info import PathInfo
-from dvc.path_info import URLInfo
+from dvc.path_info import PathInfo, URLInfo
 from dvc.progress import Tqdm
 from dvc.remote.slow_link_detection import slow_link_guard
 from dvc.state import StateNoop
-from dvc.utils import makedirs
-from dvc.utils import relpath
-from dvc.utils import tmp_fname
-from dvc.utils.compat import basestring
-from dvc.utils.compat import FileNotFoundError
-from dvc.utils.compat import str
-from dvc.utils.compat import urlparse
+from dvc.utils import makedirs, relpath, tmp_fname
 from dvc.utils.fs import move
 from dvc.utils.http import open_url
 
@@ -49,13 +45,6 @@ STATUS_MAP = {
 }
 
 
-class DataCloudError(DvcException):
-    """ Data Cloud exception """
-
-    def __init__(self, msg):
-        super(DataCloudError, self).__init__("Data sync error: {}".format(msg))
-
-
 class RemoteCmdError(DvcException):
     def __init__(self, remote, cmd, ret, err):
         super(RemoteCmdError, self).__init__(
@@ -847,7 +836,7 @@ class RemoteBASE(object):
         pass
 
     def _checkout_dir(
-        self, path_info, checksum, force, progress_callback=None
+        self, path_info, checksum, force, progress_callback=None, relink=False
     ):
         # Create dir separately so that dir is created
         # even if there are no files in it
@@ -865,9 +854,8 @@ class RemoteBASE(object):
             entry_info = path_info / relative_path
 
             entry_checksum_info = {self.PARAM_CHECKSUM: entry_checksum}
-            if self.changed(entry_info, entry_checksum_info):
-                if self.exists(entry_info):
-                    self.safe_remove(entry_info, force=force)
+            if relink or self.changed(entry_info, entry_checksum_info):
+                self.safe_remove(entry_info, force=force)
                 self.link(entry_cache_info, entry_info)
                 self.state.save(entry_info, entry_checksum)
             if progress_callback:
@@ -889,7 +877,12 @@ class RemoteBASE(object):
             self.safe_remove(path, force)
 
     def checkout(
-        self, path_info, checksum_info, force=False, progress_callback=None
+        self,
+        path_info,
+        checksum_info,
+        force=False,
+        progress_callback=None,
+        relink=False,
     ):
         if path_info.scheme not in ["local", self.scheme]:
             raise NotImplementedError
@@ -905,7 +898,7 @@ class RemoteBASE(object):
             self.safe_remove(path_info, force=force)
             failed = path_info
 
-        elif not self.changed(path_info, checksum_info):
+        elif not relink and not self.changed(path_info, checksum_info):
             msg = "Data '{}' didn't change."
             logger.debug(msg.format(str(path_info)))
             skip = True
@@ -926,18 +919,23 @@ class RemoteBASE(object):
         msg = "Checking out '{}' with cache '{}'."
         logger.debug(msg.format(str(path_info), checksum))
 
-        self._checkout(path_info, checksum, force, progress_callback)
+        self._checkout(path_info, checksum, force, progress_callback, relink)
         return None
 
     def _checkout(
-        self, path_info, checksum, force=False, progress_callback=None
+        self,
+        path_info,
+        checksum,
+        force=False,
+        progress_callback=None,
+        relink=False,
     ):
         if not self.is_dir_checksum(checksum):
             return self._checkout_file(
                 path_info, checksum, force, progress_callback=progress_callback
             )
         return self._checkout_dir(
-            path_info, checksum, force, progress_callback=progress_callback
+            path_info, checksum, force, progress_callback, relink
         )
 
     def get_files_number(self, checksum):

dvc/remote/hdfs.py

@@ -100,8 +100,14 @@ class RemoteHDFS(RemoteBASE):
             # NOTE: this is how `hadoop fs -cp` works too: it copies through
             # your local machine.
             with hdfs.open(from_info.path, "rb") as from_fobj:
-                with hdfs.open(to_info.path, "wb") as to_fobj:
-                    to_fobj.upload(from_fobj)
+                tmp_info = to_info.parent / tmp_fname(to_info.name)
+                try:
+                    with hdfs.open(tmp_info.path, "wb") as tmp_fobj:
+                        tmp_fobj.upload(from_fobj)
+                    hdfs.rename(tmp_info.path, to_info.path)
+                except Exception:
+                    self.remove(tmp_info)
+                    raise
 
     def remove(self, path_info):
         if path_info.scheme != "hdfs":

dvc/remote/local.py

@@ -164,9 +164,14 @@ class RemoteLOCAL(RemoteBASE):
 
         move(from_info, to_info, mode=mode)
 
-    @staticmethod
-    def copy(from_info, to_info):
-        System.copy(from_info, to_info)
+    def copy(self, from_info, to_info):
+        tmp_info = to_info.parent / tmp_fname(to_info.name)
+        try:
+            System.copy(from_info, tmp_info)
+            os.rename(fspath_py35(tmp_info), fspath_py35(to_info))
+        except Exception:
+            self.remove(tmp_info)
+            raise
 
     @staticmethod
     def symlink(from_info, to_info):

dvc/remote/s3.py

@@ -7,7 +7,7 @@ import threading
 
 from funcy import cached_property, wrap_prop
 
-from dvc.config import Config
+from dvc.config import Config, ConfigError
 from dvc.exceptions import DvcException
 from dvc.exceptions import ETagMismatchError
 from dvc.path_info import CloudURLInfo
@@ -53,6 +53,8 @@ class RemoteS3(RemoteBASE):
         if self.acl:
             self.extra_args["ACL"] = self.acl
 
+        self._append_aws_grants_to_extra_args(config)
+
         shared_creds = config.get(Config.SECTION_AWS_CREDENTIALPATH)
         if shared_creds:
             os.environ.setdefault("AWS_SHARED_CREDENTIALS_FILE", shared_creds)
@@ -302,3 +304,34 @@ class RemoteS3(RemoteBASE):
                 continue
 
             yield path_info.replace(path=fname)
+
+    def _append_aws_grants_to_extra_args(self, config):
+        # Keys for extra_args can be one of the following list:
+        # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/customizations/s3.html#boto3.s3.transfer.S3Transfer.ALLOWED_UPLOAD_ARGS
+        """
+          ALLOWED_UPLOAD_ARGS = [
+            'ACL', 'CacheControl', 'ContentDisposition', 'ContentEncoding',
+            'ContentLanguage', 'ContentType', 'Expires', 'GrantFullControl',
+            'GrantRead', 'GrantReadACP', 'GrantWriteACP', 'Metadata',
+            'RequestPayer', 'ServerSideEncryption', 'StorageClass',
+            'SSECustomerAlgorithm', 'SSECustomerKey', 'SSECustomerKeyMD5',
+            'SSEKMSKeyId', 'WebsiteRedirectLocation'
+          ]
+        """
+
+        grants = {
+            Config.SECTION_AWS_GRANT_FULL_CONTROL: "GrantFullControl",
+            Config.SECTION_AWS_GRANT_READ: "GrantRead",
+            Config.SECTION_AWS_GRANT_READ_ACP: "GrantReadACP",
+            Config.SECTION_AWS_GRANT_WRITE_ACP: "GrantWriteACP",
+        }
+
+        for grant_option, extra_args_key in grants.items():
+            if config.get(grant_option, ""):
+                if self.acl:
+                    raise ConfigError(
+                        "`acl` and `grant_*` AWS S3 config options "
+                        "are mutually exclusive"
+                    )
+
+                self.extra_args[extra_args_key] = config.get(grant_option)

dvc/remote/ssh/__init__.py

@@ -178,7 +178,7 @@ class RemoteSSH(RemoteBASE):
             raise NotImplementedError
 
         with self.ssh(from_info) as ssh:
-            ssh.copy(from_info.path, to_info.path)
+            ssh.atomic_copy(from_info.path, to_info.path)
 
     def symlink(self, from_info, to_info):
         if not from_info.scheme == to_info.scheme == self.scheme:

dvc/remote/ssh/connection.py

@@ -195,10 +195,10 @@ class SSHConnection:
         try:
             self.sftp.rename(src, dst)
         except OSError:
-            self._atomic_copy(src, dst)
+            self.atomic_copy(src, dst)
             self.remove(src)
 
-    def _atomic_copy(self, src, dst):
+    def atomic_copy(self, src, dst):
         tmp = tmp_fname(dst)
 
         try:

dvc/repo/__init__.py

@@ -76,6 +76,7 @@ class Repo(object):
         from dvc.repo.metrics import Metrics
         from dvc.scm.tree import WorkingTree
         from dvc.repo.tag import Tag
+        from dvc.utils import makedirs
 
         root_dir = self.find_root(root_dir)
 
@@ -88,6 +89,9 @@ class Repo(object):
 
         self.tree = WorkingTree(self.root_dir)
 
+        self.tmp_dir = os.path.join(self.dvc_dir, "tmp")
+        makedirs(self.tmp_dir, exist_ok=True)
+
         hardlink_lock = self.config.config["core"].get("hardlink_lock", False)
         self.lock = make_lock(
             os.path.join(self.dvc_dir, "lock"),
@@ -95,6 +99,7 @@ class Repo(object):
             hardlink_lock=hardlink_lock,
             friendly=True,
         )
+
         # 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)
@@ -165,9 +170,8 @@ class Repo(object):
 
         flist = (
             [self.config.config_local_file, updater.updater_file]
+            + [self.lock.lockfile, updater.lock.lockfile, self.tmp_dir]
             + self.state.files
-            + self.lock.files
-            + updater.lock.files
         )
 
         if path_isin(self.cache.local.cache_dir, self.root_dir):
@@ -352,11 +356,7 @@ class Repo(object):
                     continue
 
                 for out in outs:
-                    if (
-                        out == dep.path_info
-                        or dep.path_info.isin(out)
-                        or out.isin(dep.path_info)
-                    ):
+                    if out.overlaps(dep.path_info):
                         dep_stage = outs[out].stage
                         dep_node = relpath(dep_stage.path, self.root_dir)
                         G.add_node(dep_node, stage=dep_stage)

dvc/repo/checkout.py

@@ -25,7 +25,12 @@ def get_all_files_numbers(stages):
 
 
 def _checkout(
-    self, targets=None, with_deps=False, force=False, recursive=False
+    self,
+    targets=None,
+    with_deps=False,
+    force=False,
+    relink=False,
+    recursive=False,
 ):
     from dvc.stage import StageFileDoesNotExistError, StageFileBadNameError
 
@@ -55,7 +60,11 @@ def _checkout(
     ) as pbar:
         for stage in stages:
             failed.extend(
-                stage.checkout(force=force, progress_callback=pbar.update_desc)
+                stage.checkout(
+                    force=force,
+                    progress_callback=pbar.update_desc,
+                    relink=relink,
+                )
             )
     if failed:
         raise CheckoutError(failed)

dvc/rwlock.py

@@ -0,0 +1,191 @@
+from __future__ import unicode_literals
+
+import os
+import json
+
+from collections import defaultdict
+from contextlib import contextmanager
+
+from voluptuous import Schema, Invalid
+from funcy.py3 import lmap, lfilter, lmapcat
+
+from .exceptions import DvcException
+from .lock import LockError
+from .utils.compat import (
+    convert_to_unicode,
+    FileNotFoundError,
+    JSONDecodeError,
+    str,
+)
+from .utils.fs import relpath
+
+INFO_SCHEMA = {"pid": int, "cmd": str}
+
+SCHEMA = Schema({"write": {str: INFO_SCHEMA}, "read": {str: [INFO_SCHEMA]}})
+
+
+class RWLockFileCorruptedError(DvcException):
+    def __init__(self, path, cause):
+        super(RWLockFileCorruptedError, self).__init__(
+            "Unable to read RWLock-file '{}'. JSON structure is "
+            "corrupted".format(relpath(path)),
+            cause=cause,
+        )
+
+
+class RWLockFileFormatError(DvcException):
+    def __init__(self, path, cause):
+        super(RWLockFileFormatError, self).__init__(
+            "RWLock-file '{}' format error.".format(relpath(path)), cause=cause
+        )
+
+
+@contextmanager
+def _edit_rwlock(lock_dir):
+    path = os.path.join(lock_dir, "rwlock")
+    try:
+        with open(path, "r") as fobj:
+            lock = json.load(fobj)
+        lock = SCHEMA(convert_to_unicode(lock))
+    except FileNotFoundError:
+        lock = SCHEMA({})
+    except JSONDecodeError as exc:
+        raise RWLockFileCorruptedError(path, cause=exc)
+    except Invalid as exc:
+        raise RWLockFileFormatError(path, cause=exc)
+    lock = defaultdict(dict, lock)
+    lock["read"] = defaultdict(list, lock["read"])
+    lock["write"] = defaultdict(dict, lock["write"])
+    yield lock
+    with open(path, "w+") as fobj:
+        json.dump(lock, fobj)
+
+
+def _infos_to_str(infos):
+    return "\n".join(
+        "  (PID {}): {}".format(info["pid"], info["cmd"]) for info in infos
+    )
+
+
+def _check_no_writers(lock, info, path_infos):
+    for path_info in path_infos:
+        blocking_urls = lfilter(path_info.overlaps, lock["write"])
+        if not blocking_urls:
+            continue
+
+        writers = lmap(lock["write"].get, blocking_urls)
+        writers = lfilter(lambda i: info != i, writers)
+        if not writers:
+            continue
+
+        raise LockError(
+            "'{}' is busy, it is being written to by:\n{}".format(
+                str(path_info), _infos_to_str(writers)
+            )
+        )
+
+
+def _check_no_readers(lock, info, path_infos):
+    for path_info in path_infos:
+        blocking_urls = lfilter(path_info.overlaps, lock["read"])
+        if not blocking_urls:
+            continue
+
+        readers = lmapcat(lock["read"].get, blocking_urls)
+        readers = lfilter(lambda i: info != i, readers)
+        if not readers:
+            continue
+
+        raise LockError(
+            "'{}' is busy, it is being read by:\n{}".format(
+                str(path_info), _infos_to_str(readers)
+            )
+        )
+
+
+def _acquire_read(lock, info, path_infos):
+    changes = []
+
+    for path_info in path_infos:
+        url = path_info.url
+
+        readers = lock["read"][url]
+        if info in readers:
+            continue
+
+        changes.append(url)
+        readers.append(info)
+
+    return changes
+
+
+def _acquire_write(lock, info, path_infos):
+    changes = []
+
+    for path_info in path_infos:
+        url = path_info.url
+
+        if lock["write"][url] == info:
+            continue
+
+        changes.append(url)
+        lock["write"][url] = info
+
+    return changes
+
+
+def _release_write(lock, info, changes):
+    for url in changes:
+        assert "write" in lock
+        assert url in lock["write"]
+        assert lock["write"][url] == info
+        del lock["write"][url]
+        if not lock["write"]:
+            del lock["write"]
+
+
+def _release_read(lock, info, changes):
+    for url in changes:
+        assert "read" in lock
+        assert url in lock["read"]
+        assert info in lock["read"][url]
+        lock["read"][url].remove(info)
+        if not lock["read"][url]:
+            del lock["read"][url]
+        if not lock["read"]:
+            del lock["read"]
+
+
+@contextmanager
+def rwlock(tmp_dir, cmd, read, write):
+    """Create non-thread-safe RWLock for PathInfos.
+
+    Args:
+        tmp_dir (str): existing directory where to create the rwlock file.
+        cmd (str): command that will be working on these PathInfos.
+        read ([PathInfo]): PathInfos that are going to be read.
+        write ([PathInfo]): PathInfos that are going to be written.
+
+    Raises:
+        LockError: raised if PathInfo we want to read is being written to by
+            another command or if PathInfo we want to write is being written
+            to or read from by another command.
+        RWLockFileCorruptedError: raised if rwlock file is not a valid JSON.
+        RWLockFileFormatError: raised if rwlock file is a valid JSON, but
+            has internal format that doesn't pass our schema validation.
+    """
+    info = {"pid": os.getpid(), "cmd": cmd}
+
+    with _edit_rwlock(tmp_dir) as lock:
+        _check_no_writers(lock, info, read + write)
+        _check_no_readers(lock, info, write)
+
+        rchanges = _acquire_read(lock, info, read)
+        wchanges = _acquire_write(lock, info, write)
+
+    try:
+        yield
+    finally:
+        with _edit_rwlock(tmp_dir) as lock:
+            _release_write(lock, info, wchanges)
+            _release_read(lock, info, rchanges)

dvc/stage.py

@@ -7,7 +7,10 @@ import re
 import signal
 import subprocess
 import threading
+
+from functools import wraps
 from itertools import chain
+from funcy import decorator
 
 from voluptuous import Any, Schema, MultipleInvalid
 
@@ -129,6 +132,52 @@ class MissingDataSource(DvcException):
         super(MissingDataSource, self).__init__(msg)
 
 
+@decorator
+def rwlocked(call, read=None, write=None):
+    import sys
+    from dvc.rwlock import rwlock
+    from dvc.dependency.repo import DependencyREPO
+
+    if read is None:
+        read = []
+
+    if write is None:
+        write = []
+
+    stage = call._args[0]
+
+    def _chain(names):
+        return [
+            item.path_info
+            for attr in names
+            for item in getattr(stage, attr)
+            # There is no need to lock DependencyREPO deps, as there is no
+            # corresponding OutputREPO, so we can't even write it.
+            if not isinstance(item, DependencyREPO)
+        ]
+
+    cmd = " ".join(sys.argv)
+
+    with rwlock(stage.repo.tmp_dir, cmd, _chain(read), _chain(write)):
+        return call()
+
+
+def unlocked_repo(f):
+    @wraps(f)
+    def wrapper(stage, *args, **kwargs):
+        stage.repo.state.dump()
+        stage.repo.lock.unlock()
+        stage.repo._reset()
+        try:
+            ret = f(stage, *args, **kwargs)
+        finally:
+            stage.repo.lock.lock()
+            stage.repo.state.load()
+        return ret
+
+    return wrapper
+
+
 class Stage(object):
     STAGE_FILE = "Dvcfile"
     STAGE_FILE_SUFFIX = ".dvc"
@@ -286,6 +335,7 @@ class Stage(object):
             return True
         return False
 
+    @rwlocked(read=["deps", "outs"])
     def changed(self):
         # Short-circuit order: stage md5 is fast, deps are expected to change
         ret = (
@@ -299,6 +349,7 @@ class Stage(object):
 
         return ret
 
+    @rwlocked(write=["outs"])
     def remove_outs(self, ignore_remove=False, force=False):
         """Used mainly for `dvc remove --outs` and :func:`Stage.reproduce`."""
         for out in self.outs:
@@ -316,6 +367,7 @@ class Stage(object):
         for out in self.outs:
             out.unprotect()
 
+    @rwlocked(write=["outs"])
     def remove(self, force=False, remove_outs=True):
         if remove_outs:
             self.remove_outs(ignore_remove=True, force=force)
@@ -323,6 +375,7 @@ class Stage(object):
             self.unprotect_outs()
         os.unlink(self.path)
 
+    @rwlocked(read=["deps"], write=["outs"])
     def reproduce(self, interactive=False, **kwargs):
 
         if not kwargs.get("force", False) and not self.changed():
@@ -755,6 +808,7 @@ class Stage(object):
                 )
             self.save()
 
+    @rwlocked(write=["outs"])
     def commit(self):
         for out in self.outs:
             out.commit()
@@ -801,6 +855,7 @@ class Stage(object):
             if occurrence > 1:
                 raise ArgumentDuplicationError(str(path))
 
+    @unlocked_repo
     def _run(self):
         self._check_missing_deps()
 
@@ -850,6 +905,7 @@ class Stage(object):
         if (p is None) or (p.returncode != 0):
             raise StageCmdFailedError(self)
 
+    @rwlocked(read=["deps"], write=["outs"])
     def run(self, dry=False, no_commit=False, force=False):
         if (self.cmd or self.is_import) and not self.locked and not dry:
             self.remove_outs(ignore_remove=False, force=False)
@@ -874,7 +930,6 @@ class Stage(object):
                     self.outs[0].checkout()
                 else:
                     self.deps[0].download(self.outs[0])
-
         elif self.is_data_source:
             msg = "Verifying data sources in '{}'".format(self.relpath)
             logger.info(msg)
@@ -904,11 +959,15 @@ class Stage(object):
         if paths:
             raise MissingDataSource(paths)
 
-    def checkout(self, force=False, progress_callback=None):
+    @rwlocked(write=["outs"])
+    def checkout(self, force=False, progress_callback=None, relink=False):
         failed_checkouts = []
         for out in self.outs:
             failed = out.checkout(
-                force=force, tag=self.tag, progress_callback=progress_callback
+                force=force,
+                tag=self.tag,
+                progress_callback=progress_callback,
+                relink=relink,
             )
             if failed:
                 failed_checkouts.append(failed)
@@ -923,6 +982,7 @@ class Stage(object):
 
         return ret
 
+    @rwlocked(read=["deps", "outs"])
     def status(self):
         ret = []
 

dvc/utils/compat.py

@@ -123,6 +123,7 @@ if is_py2:
     makedirs = _makedirs
     range = xrange  # noqa: F821
     FileNotFoundError = IOError
+    JSONDecodeError = ValueError
 
     import StringIO
     import io
@@ -164,6 +165,7 @@ elif is_py3:
     import configparser as ConfigParser  # noqa: F401
     from collections.abc import Mapping  # noqa: F401
     from contextlib import ExitStack  # noqa: F401
+    from json.decoder import JSONDecodeError  # noqa: F401
 
     builtin_str = str  # noqa: F821
     str = str  # noqa: F821

dvc/version.py

@@ -7,7 +7,7 @@ import os
 import subprocess
 
 
-_BASE_VERSION = "0.76.0"
+_BASE_VERSION = "0.77.2"
 
 
 def _generate_version(base_version):

tests/func/test_add.py

@@ -566,12 +566,14 @@ def test_should_not_checkout_when_adding_cached_copy(tmp_dir, dvc, mocker):
 def test_should_relink_on_repeated_add(
     link, new_link, link_test_func, tmp_dir, dvc
 ):
+    from dvc.path_info import PathInfo
+
     dvc.config.set("cache", "type", link)
 
     tmp_dir.dvc_gen({"foo": "foo", "bar": "bar"})
 
     os.remove("foo")
-    getattr(dvc.cache.local, link)("bar", "foo")
+    getattr(dvc.cache.local, link)(PathInfo("bar"), PathInfo("foo"))
 
     dvc.cache.local.cache_types = [new_link]
 

tests/func/test_checkout.py

@@ -26,7 +26,6 @@ from dvc.utils.stage import load_stage_file
 from tests.basic_env import TestDvc
 from tests.basic_env import TestDvcGit
 from tests.func.test_repro import TestRepro
-from tests.utils import spy
 
 logger = logging.getLogger("dvc")
 
@@ -485,58 +484,29 @@ def test_checkout_no_checksum(tmp_dir, dvc):
     assert not os.path.exists("file")
 
 
-@pytest.mark.skip
 @pytest.mark.parametrize(
     "link, link_test_func",
     [("hardlink", System.is_hardlink), ("symlink", System.is_symlink)],
 )
-def test_should_relink_on_checkout(link, link_test_func, repo_dir, dvc_repo):
-    dvc_repo.cache.local.cache_types = [link]
+def test_checkout_relink(tmp_dir, dvc, link, link_test_func):
+    dvc.cache.local.cache_types = [link]
 
-    dvc_repo.add(repo_dir.DATA_DIR)
-    dvc_repo.unprotect(repo_dir.DATA_SUB)
+    tmp_dir.dvc_gen({"dir": {"data": "text"}})
+    dvc.unprotect("dir/data")
+    assert not link_test_func("dir/data")
 
-    dvc_repo.checkout([repo_dir.DATA_DIR + Stage.STAGE_FILE_SUFFIX])
-
-    assert link_test_func(repo_dir.DATA_SUB)
+    dvc.checkout(["dir.dvc"], relink=True)
+    assert link_test_func("dir/data")
 
 
-@pytest.mark.skip
 @pytest.mark.parametrize("link", ["hardlink", "symlink", "copy"])
-def test_should_protect_on_checkout(link, dvc_repo, repo_dir):
-    dvc_repo.cache.local.cache_types = [link]
-    dvc_repo.cache.local.protected = True
+def test_checkout_relink_protected(tmp_dir, dvc, link):
+    dvc.cache.local.cache_types = [link]
+    dvc.cache.local.protected = True
 
-    dvc_repo.add(repo_dir.FOO)
-    dvc_repo.unprotect(repo_dir.FOO)
+    tmp_dir.dvc_gen("foo", "foo")
+    dvc.unprotect("foo")
+    assert os.access("foo", os.W_OK)
 
-    dvc_repo.checkout([repo_dir.FOO + Stage.STAGE_FILE_SUFFIX])
-
-    assert not os.access(repo_dir.FOO, os.W_OK)
-
-
-@pytest.mark.skip
-def test_should_relink_only_one_file_in_dir(dvc_repo, repo_dir):
-    dvc_repo.cache.local.cache_types = ["symlink"]
-
-    dvc_repo.add(repo_dir.DATA_DIR)
-    dvc_repo.unprotect(repo_dir.DATA_SUB)
-
-    link_spy = spy(System.symlink)
-    with patch.object(dvc_repo.cache.local, "symlink", link_spy):
-        dvc_repo.checkout([repo_dir.DATA_DIR + Stage.STAGE_FILE_SUFFIX])
-
-    assert link_spy.mock.call_count == 1
-
-
-@pytest.mark.skip
-@pytest.mark.parametrize("link", ["hardlink", "symlink", "copy"])
-def test_should_not_relink_on_unchanged_dependency(link, dvc_repo, repo_dir):
-    dvc_repo.cache.local.cache_types = [link]
-
-    dvc_repo.add(repo_dir.DATA_DIR)
-
-    with patch.object(dvc_repo.cache.local, "link") as mock_link:
-        dvc_repo.checkout([repo_dir.DATA_DIR + Stage.STAGE_FILE_SUFFIX])
-
-    assert not mock_link.called
+    dvc.checkout(["foo.dvc"], relink=True)
+    assert not os.access("foo", os.W_OK)

tests/func/test_run.py

@@ -893,7 +893,7 @@ class TestShouldNotCheckoutUponCorruptedLocalHardlinkCache(TestDvc):
         )
         patch_run = mock.patch.object(stage, "_run", wraps=stage._run)
 
-        with self.dvc.state:
+        with self.dvc.lock, self.dvc.state:
             with patch_checkout as mock_checkout:
                 with patch_run as mock_run:
                     stage.run()

tests/unit/command/test_checkout.py

@@ -0,0 +1,21 @@
+from dvc.cli import parse_args
+from dvc.command.checkout import CmdCheckout
+
+
+def test_checkout(tmp_dir, dvc, mocker):
+    cli_args = parse_args(
+        ["checkout", "foo.dvc", "bar.dvc", "--relink", "--with-deps"]
+    )
+    assert cli_args.func == CmdCheckout
+
+    cmd = cli_args.func(cli_args)
+    m = mocker.patch("dvc.repo.Repo.checkout")
+
+    assert cmd.run() == 0
+    m.assert_called_once_with(
+        targets=["foo.dvc", "bar.dvc"],
+        force=False,
+        recursive=False,
+        relink=True,
+        with_deps=True,
+    )

tests/unit/remote/test_s3.py

@@ -0,0 +1,56 @@
+import pytest
+
+from dvc.config import ConfigError
+from dvc.remote.s3 import RemoteS3
+
+
+bucket_name = "bucket-name"
+prefix = "some/prefix"
+url = "s3://{}/{}".format(bucket_name, prefix)
+
+
+@pytest.fixture(autouse=True)
+def grants():
+    return {
+        "grant_read": "id=read-permission-id,id=other-read-permission-id",
+        "grant_read_acp": "id=read-acp-permission-id",
+        "grant_write_acp": "id=write-acp-permission-id",
+        "grant_full_control": "id=full-control-permission-id",
+    }
+
+
+def test_init():
+    config = {"url": url}
+    remote = RemoteS3(None, config)
+
+    assert remote.path_info == url
+
+
+def test_grants():
+    config = {
+        "url": url,
+        "grant_read": "id=read-permission-id,id=other-read-permission-id",
+        "grant_read_acp": "id=read-acp-permission-id",
+        "grant_write_acp": "id=write-acp-permission-id",
+        "grant_full_control": "id=full-control-permission-id",
+    }
+    remote = RemoteS3(None, config)
+
+    assert (
+        remote.extra_args["GrantRead"]
+        == "id=read-permission-id,id=other-read-permission-id"
+    )
+    assert remote.extra_args["GrantReadACP"] == "id=read-acp-permission-id"
+    assert remote.extra_args["GrantWriteACP"] == "id=write-acp-permission-id"
+    assert (
+        remote.extra_args["GrantFullControl"]
+        == "id=full-control-permission-id"
+    )
+
+
+def test_grants_mutually_exclusive_acl_error(grants):
+    for grant_option, grant_value in grants.items():
+        config = {"url": url, "acl": "public-read", grant_option: grant_value}
+
+        with pytest.raises(ConfigError):
+            RemoteS3(None, config)

tests/unit/test_rwlock.py

@@ -0,0 +1,98 @@
+from __future__ import unicode_literals
+
+import os
+
+import pytest
+
+from dvc.rwlock import (
+    rwlock,
+    _edit_rwlock,
+    RWLockFileFormatError,
+    RWLockFileCorruptedError,
+)
+from dvc.lock import LockError
+from dvc.utils.compat import fspath
+from dvc.path_info import PathInfo
+
+
+def test_rwlock(tmp_path):
+    path = fspath(tmp_path)
+    foo = PathInfo("foo")
+
+    with rwlock(path, "cmd1", [foo], []):
+        with pytest.raises(LockError):
+            with rwlock(path, "cmd2", [], [foo]):
+                pass
+
+    with rwlock(path, "cmd1", [], [foo]):
+        with pytest.raises(LockError):
+            with rwlock(path, "cmd2", [foo], []):
+                pass
+
+    with rwlock(path, "cmd1", [], [foo]):
+        with pytest.raises(LockError):
+            with rwlock(path, "cmd2", [], [foo]):
+                pass
+
+
+def test_rwlock_reentrant(tmp_path):
+    path = fspath(tmp_path)
+    foo = PathInfo("foo")
+
+    with rwlock(path, "cmd1", [], [foo]):
+        with rwlock(path, "cmd1", [], [foo]):
+            pass
+        with _edit_rwlock(path) as lock:
+            assert lock == {
+                "read": {},
+                "write": {"foo": {"cmd": "cmd1", "pid": os.getpid()}},
+            }
+
+    with rwlock(path, "cmd", [foo], []):
+        with rwlock(path, "cmd", [foo], []):
+            pass
+        with _edit_rwlock(path) as lock:
+            assert lock == {
+                "read": {"foo": [{"cmd": "cmd", "pid": os.getpid()}]},
+                "write": {},
+            }
+
+
+def test_rwlock_subdirs(tmp_path):
+    path = fspath(tmp_path)
+    foo = PathInfo("foo")
+    subfoo = PathInfo("foo/subfoo")
+
+    with rwlock(path, "cmd1", [foo], []):
+        with pytest.raises(LockError):
+            with rwlock(path, "cmd2", [], [subfoo]):
+                pass
+
+    with rwlock(path, "cmd1", [], [subfoo]):
+        with pytest.raises(LockError):
+            with rwlock(path, "cmd2", [foo], []):
+                pass
+
+    with rwlock(path, "cmd1", [], [subfoo]):
+        with pytest.raises(LockError):
+            with rwlock(path, "cmd2", [], [foo]):
+                pass
+
+    with rwlock(path, "cmd1", [subfoo], []):
+        with rwlock(path, "cmd2", [foo], []):
+            pass
+
+
+def test_broken_rwlock(tmp_path):
+    dir_path = fspath(tmp_path)
+    path = tmp_path / "rwlock"
+
+    path.write_text('{"broken": "format"}', encoding="utf-8")
+    with pytest.raises(RWLockFileFormatError):
+        with _edit_rwlock(dir_path):
+            pass
+
+    path.write_text("{broken json", encoding="utf-8")
+    with pytest.raises(RWLockFileCorruptedError):
+        with _edit_rwlock(dir_path):
+            pass

tests/unit/test_stage.py

@@ -93,16 +93,14 @@ def test_stage_update(mocker):
     not isinstance(threading.current_thread(), threading._MainThread),
     reason="Not running in the main thread.",
 )
-def test_stage_run_ignore_sigint(mocker):
-    stage = Stage(None, "path")
-
+def test_stage_run_ignore_sigint(dvc_repo, mocker):
     proc = mocker.Mock()
     communicate = mocker.Mock()
     proc.configure_mock(returncode=0, communicate=communicate)
     popen = mocker.patch.object(subprocess, "Popen", return_value=proc)
     signal_mock = mocker.patch("signal.signal")
 
-    stage._run()
+    dvc_repo.run(cmd="path")
 
     assert popen.called_once()
     assert communicate.called_once_with()
@@ -110,8 +108,8 @@ def test_stage_run_ignore_sigint(mocker):
     assert signal.getsignal(signal.SIGINT) == signal.default_int_handler
 
 
-def test_always_changed():
-    stage = Stage(None, "path", always_changed=True)
+def test_always_changed(dvc_repo):
+    stage = Stage(dvc_repo, "path", always_changed=True)
     stage.save()
     assert stage.changed()
     assert stage.status()["path"] == ["always changed"]