Repository: iterative/dvc
Pull request #828 (2018-06-30 00:39:55)

RFC/perf: data cloud logic


Patch:
diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py
index 050cbfd12..4f80bf2e4 100644
--- a/dvc/data_cloud.py
+++ b/dvc/data_cloud.py
@@ -4,14 +4,12 @@ from multiprocessing.pool import ThreadPool
 from dvc.logger import Logger
 from dvc.exceptions import DvcException
 from dvc.config import Config, ConfigError
-from dvc.utils import map_progress
 
 from dvc.remote.s3 import RemoteS3
 from dvc.remote.gs import RemoteGS
 from dvc.remote.ssh import RemoteSSH
 from dvc.remote.local import RemoteLOCAL
 from dvc.remote.base import RemoteBase
-from dvc.remote.base import STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED
 
 
 class DataCloud(object):
@@ -83,17 +81,6 @@ class DataCloud(object):
         cloud = cloud_type(self.project, cloud_config)
         return cloud
 
-    def _collect(self, cloud, targets, jobs, local):
-        collected = set()
-        pool = ThreadPool(processes=jobs)
-        args = zip(targets, [local]*len(targets))
-        ret = pool.map(cloud.collect, args)
-
-        for r in ret:
-            collected |= set(r)
-
-        return collected
-
     def _get_cloud(self, remote):
         if remote:
             return self._init_remote(remote)
@@ -104,42 +91,20 @@ class DataCloud(object):
         raise ConfigError("No remote repository specified. Setup default repository " \
                           "with 'dvc config core.remote <name>' or use '-r <name>'.")
 
-    def _filter(self, func, status, targets, jobs, remote):
-        cloud = self._get_cloud(remote)
-
-        filtered = []
-        for t, s in self._status(cloud, targets, jobs):
-            if s == STATUS_MODIFIED or s == status:
-                filtered.append(t)
-
-        return map_progress(getattr(cloud, func), filtered, jobs)
-
     def push(self, targets, jobs=1, remote=None):
         """
         Push data items in a cloud-agnostic way.
         """
-        return self._filter('push', STATUS_NEW, targets, jobs, remote)
+        return self.project.cache.local.push(targets, jobs=jobs, remote=self._get_cloud(remote))
 
     def pull(self, targets, jobs=1, remote=None):
         """
         Pull data items in a cloud-agnostic way.
         """
-        return self._filter('pull', STATUS_DELETED, targets, jobs, remote)
-
-    def _collect_targets(self, cloud, targets, jobs=1):
-        collected = set()
-        collected |= self._collect(cloud, targets, jobs, True)
-        collected |= self._collect(cloud, targets, jobs, False)
-        return list(collected)
-
-    def _status(self, cloud, targets, jobs=1):
-        collected = self._collect_targets(cloud, targets, jobs)
-        return map_progress(cloud.status, collected, jobs)
+        return self.project.cache.local.pull(targets, jobs=jobs, remote=self._get_cloud(remote))
 
     def status(self, targets, jobs=1, remote=None):
         """
         Check status of data items in a cloud-agnostic way.
         """
-        cloud = self._get_cloud(remote)
-
-        return self._status(cloud, targets, jobs)
+        return self.project.cache.local.status(targets, jobs=jobs, remote=self._get_cloud(remote))
diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py
index 815143396..9b887e765 100644
--- a/dvc/dependency/base.py
+++ b/dvc/dependency/base.py
@@ -65,4 +65,4 @@ class DependencyBase(object):
         return {self.PARAM_PATH: self.path}
 
     def download(self, path):
-        self.remote.download(self.path_info, path)
+        self.remote.download([self.path_info], [path])
diff --git a/dvc/project.py b/dvc/project.py
index c99de489a..09106c2d0 100644
--- a/dvc/project.py
+++ b/dvc/project.py
@@ -269,26 +269,13 @@ class Project(object):
 
             for stage in stages:
                 for out in stage.outs:
-                    scheme = out.path_info['scheme']
-
                     if not out.use_cache or not out.info:
                         continue
 
-                    cache[scheme] += [out.info]
-
-                    if scheme != 'local':
-                        continue
-
-                    if self.cache.local.is_dir_cache(out.cache) and os.path.isfile(out.cache):
-                        dir_cache = self.cache.local.load_dir_cache(out.cache)
-                        cache[scheme] += dir_cache
+                    cache[out.path_info['scheme']] += [out.info]
 
         return cache
 
-    def _used_cache_files(self, target=None, all_branches=False):
-        infos = self._used_cache(target=target, all_branches=all_branches)['local']
-        return [self.cache.local.get(info[self.cache.local.PARAM_MD5]) for info in infos]
-
     def gc(self, all_branches=False):
         clist = self._used_cache(target=None, all_branches=all_branches)
         self.cache.local.gc(clist['local'])
@@ -306,10 +293,10 @@ class Project(object):
             self.cache.hdfs.gc(clist['hdfs'])
 
     def push(self, target=None, jobs=1, remote=None, all_branches=False):
-        self.cloud.push(self._used_cache_files(target, all_branches), jobs, remote=remote)
+        self.cloud.push(self._used_cache(target, all_branches)['local'], jobs, remote=remote)
 
     def fetch(self, target=None, jobs=1, remote=None, all_branches=False):
-        self.cloud.pull(self._used_cache_files(target, all_branches), jobs, remote=remote)
+        self.cloud.pull(self._used_cache(target, all_branches)['local'], jobs, remote=remote)
 
     def pull(self, target=None, jobs=1, remote=None, all_branches=False):
         self.fetch(target, jobs, remote=remote, all_branches=all_branches)
@@ -332,19 +319,16 @@ class Project(object):
         import dvc.remote.base as cloud
 
         status = {}
-        for target, ret in self.cloud.status(self._used_cache_files(target), jobs, remote=remote):
-            if ret == cloud.STATUS_UNKNOWN or ret == cloud.STATUS_OK:
+        for md5, ret in self.cloud.status(self._used_cache(target)['local'], jobs, remote=remote):
+            if ret == cloud.STATUS_OK:
                 continue
 
             prefix_map = {
                 cloud.STATUS_DELETED: 'deleted',
-                cloud.STATUS_MODIFIED: 'modified',
                 cloud.STATUS_NEW: 'new',
             }
 
-            path = os.path.relpath(target, self.cache.local.cache_dir)
-
-            status[path] = prefix_map[ret]
+            status[md5] = prefix_map[ret]
 
         return status
 
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
index da8a405fa..94ad18cc2 100644
--- a/dvc/remote/base.py
+++ b/dvc/remote/base.py
@@ -8,19 +8,17 @@ from dvc.logger import Logger
 from dvc.exceptions import DvcException
 
 
-STATUS_UNKNOWN = 0
 STATUS_OK = 1
-STATUS_MODIFIED = 2
 STATUS_NEW = 3
 STATUS_DELETED = 4
 
 
 STATUS_MAP = {
-    # (local_exists, remote_exists, cmp)
-    (True, True, True)  : STATUS_OK,
-    (True, True, False) : STATUS_MODIFIED,
-    (True, False, None) : STATUS_NEW,
-    (False, True, None) : STATUS_DELETED,
+    # (local_exists, remote_exists)
+    (True, True)  : STATUS_OK,
+    (False, False) : STATUS_OK,
+    (True, False) : STATUS_NEW,
+    (False, True) : STATUS_DELETED,
 }
 
 
@@ -74,12 +72,6 @@ class RemoteBase(object):
     def checkout(self, path_info, checksum_info):
         raise NotImplementedError
 
-    def _get_path_info(self, path):
-        raise NotImplementedError
-
-    def _new_path_info(self, path):
-        raise NotImplementedError
-
     def download(self, path_info, path, no_progress_bar=False, name=None):
         raise NotImplementedError
 
@@ -92,52 +84,6 @@ class RemoteBase(object):
     def move(self, path_info):
         raise NotImplementedError
 
-    def cache_file_key(self, fname):
-        """ Key of a file within the bucket """
-        relpath = os.path.relpath(fname, self.project.cache.local.cache_dir)
-        relpath = relpath.replace('\\', '/')
-        return posixpath.join(self.prefix, relpath).strip('/')
-
-    def collect(self, arg):
-        path, local = arg
-        ret = [path]
-        md5 = self.project.cache.local.path_to_md5(path)
-
-        if not self.project.cache.local.is_dir_cache(path):
-            return ret
-
-        if not local and self.project.cache.local.changed(md5):
-            path_info = self._get_path_info(path)
-            if not path_info:
-                Logger.debug("File '{}' does not exist in the cloud".format(path))
-                return ret
-            self.download(path_info, path, no_progress_bar=True)
-
-        if self.project.cache.local.changed(md5):
-            return ret
-
-        for entry in self.project.cache.local.load_dir_cache(path):
-            md5 = entry[self.project.cache.local.PARAM_MD5]
-            relpath = entry[self.project.cache.local.PARAM_RELPATH]
-            cache = self.project.cache.local.get(md5)
-            ret.append(cache)
-
-        return ret
-
-    def push(self, path):
-        path_info = self._get_path_info(path)
-        if path_info:
-            Logger.debug("File '{}' already uploaded to the cloud.".format(path))
-            return None
-
-        path_info = self._new_path_info(path)
-        md5 = self.project.cache.local.path_to_md5(path)
-
-        if self.project.cache.local.changed(md5):
-            return None
-
-        self.upload(path, path_info, name=md5)
-
     def _makedirs(self, fname):
         dname = os.path.dirname(fname)
         try:
@@ -146,32 +92,8 @@ class RemoteBase(object):
             if e.errno != os.errno.EEXIST:
                 raise
 
-    def pull(self, path):
-        """ Generic method for pulling data from the cloud """
-        path_info = self._get_path_info(path)
-        if not path_info:
-            Logger.error("File '{}' does not exist in the cloud".format(path))
-            return None
-
-        md5 = self.project.cache.local.path_to_md5(path)
-
-        self.download(path_info, path, name=md5)
-
-        if self.project.cache.local.changed(md5):
-            return None
+    def md5s_to_path_infos(self, md5s):
+        raise NotImplementedError
 
-    def status(self, path):
-        """
-        Generic method for checking data item status.
-        """
-        md5 = self.project.cache.local.path_to_md5(path)
-        path_info = self._get_path_info(path)
-        remote_exists = path_info != None
-        local_exists = os.path.exists(path) and not self.project.cache.local.changed(md5)
-
-        diff = None
-        if remote_exists and local_exists:
-            md5 = self.project.cache.local.path_to_md5(path)
-            diff = not self.project.cache.local.changed(md5)
-
-        return STATUS_MAP.get((local_exists, remote_exists, diff), STATUS_UNKNOWN)
+    def exists(self, path_infos):
+        raise NotImplementedError
diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py
index 7c09a5981..c6e03e2e7 100644
--- a/dvc/remote/gs.py
+++ b/dvc/remote/gs.py
@@ -97,78 +97,105 @@ class RemoteGS(RemoteBase):
 
         blob.delete()
 
-    def _get_path_info(self, path):
-        key = self.cache_file_key(path)
-        if not self.gs.bucket(self.bucket).get_blob(key):
-            return None
-        return {'scheme': 'gs',
-                'bucket': self.bucket,
-                'key': key}
-
-    def _new_path_info(self, path):
-        key = self.cache_file_key(path)
-        return {'scheme': 'gs',
-                'bucket': self.bucket,
-                'key': key}
+    def md5s_to_path_infos(self, md5s):
+        return [{'scheme': 'gs',
+                 'bucket': self.bucket,
+                 'key': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s]
+
+    def _exists(self, gs, path_info):
+        return gs.bucket(path_info['bucket']).get_blob(path_info['key']) != None
+
+    def exists(self, path_infos):
+        ret = []
+        gs = self.gs
+        for path_info in path_infos:
+            ret.append(self._exists(gs, path_info))
+        return ret
+
+    def upload(self, paths, path_infos, names=None):
+        assert isinstance(paths, list)
+        assert isinstance(path_infos, list)
+        assert len(paths) == len(path_infos)
+        if not names:
+            names = len(paths) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(paths)
+
+        gs = self.gs
+
+        for path, path_info, name in zip(paths, path_infos, names):
+            if path_info['scheme'] != 'gs':
+                raise NotImplementedError
+
+            if not os.path.exists(path) or self._exists(gs, path_info):
+                continue
 
-    def upload(self, path, path_info, name=None):
-        if path_info['scheme'] != 'gs':
-            raise NotImplementedError
+            Logger.debug("Uploading '{}' to '{}/{}'".format(path,
+                                                            path_info['bucket'],
+                                                            path_info['key']))
 
-        Logger.debug("Uploading '{}' to '{}/{}'".format(path,
-                                                        path_info['bucket'],
-                                                        path_info['key']))
+            if not name:
+                name = os.path.basename(path)
 
-        if not name:
-            name = os.path.basename(path)
+            progress.update_target(name, 0, None)
 
-        progress.update_target(name, 0, None)
+            try:
+                gs.bucket(path_info['bucket']).blob(path_info['key']).upload_from_filename(path)
+            except Exception as exc:
+                Logger.error("Failed to upload '{}' to '{}/{}'".format(path,
+                                                                       path_info['bucket'],
+                                                                       path_info['key']), exc)
+                continue
 
-        try:
-            self.gs.bucket(path_info['bucket']).blob(path_info['key']).upload_from_filename(path)
-        except Exception as exc:
-            Logger.error("Failed to upload '{}' to '{}/{}'".format(path,
-                                                                   path_info['bucket'],
-                                                                   path_info['key']), exc)
-            return None
+            progress.finish_target(name)
 
-        progress.finish_target(name)
+    def download(self, path_infos, paths, no_progress_bar=False, names=None):
+        assert isinstance(paths, list)
+        assert isinstance(path_infos, list)
+        assert len(paths) == len(path_infos)
+        if not names:
+            names = len(paths) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(paths)
 
-        return path
+        gs = self.gs
 
-    def download(self, path_info, path, no_progress_bar=False, name=None):
-        if path_info['scheme'] != 'gs':
-            raise NotImplementedError
+        for path, path_info, name in zip(paths, path_infos, names):
+            if path_info['scheme'] != 'gs':
+                raise NotImplementedError
 
-        Logger.debug("Downloading '{}/{}' to '{}'".format(path_info['bucket'],
-                                                          path_info['key'],
-                                                          path))
+            if os.path.exists(path) or not self._exists(gs, path_info):
+                continue
 
-        tmp_file = self.tmp_file(path)
-        if not name:
-            name = os.path.basename(path)
+            Logger.debug("Downloading '{}/{}' to '{}'".format(path_info['bucket'],
+                                                              path_info['key'],
+                                                              path))
 
-        if not no_progress_bar:
-            # percent_cb is not available for download_to_filename, so
-            # lets at least update progress at keypoints(start, finish)
-            progress.update_target(name, 0, None)
+            tmp_file = self.tmp_file(path)
+            if not name:
+                name = os.path.basename(path)
 
-        self._makedirs(path)
+            if not no_progress_bar:
+                # percent_cb is not available for download_to_filename, so
+                # lets at least update progress at keypoints(start, finish)
+                progress.update_target(name, 0, None)
 
-        try:
-            self.gs.bucket(path_info['bucket']).get_blob(path_info['key']).download_to_filename(tmp_file)
-        except Exception as exc:
-            Logger.error("Failed to download '{}/{}' to '{}'".format(path_info['bucket'],
-                                                                     path_info['key'],
-                                                                     path), exc)
-            return None
+            self._makedirs(path)
 
-        os.rename(tmp_file, path)
+            try:
+                gs.bucket(path_info['bucket']).get_blob(path_info['key']).download_to_filename(tmp_file)
+            except Exception as exc:
+                Logger.error("Failed to download '{}/{}' to '{}'".format(path_info['bucket'],
+                                                                         path_info['key'],
+                                                                         path), exc)
+                continue
 
-        if not no_progress_bar:
-            progress.finish_target(name)
+            os.rename(tmp_file, path)
 
-        return path
+            if not no_progress_bar:
+                progress.finish_target(name)
 
     def _path_to_etag(self, path):
         relpath = posixpath.relpath(path, self.prefix)
diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py
index aa8c8634e..e25d69910 100644
--- a/dvc/remote/hdfs.py
+++ b/dvc/remote/hdfs.py
@@ -61,6 +61,13 @@ class RemoteHDFS(RemoteBase):
     def rm(self, path_info):
         self.hadoop_fs('rm {}'.format(path_info['url']), user=path_info['user'])
 
+    def exists(self, path_info):
+        try:
+            self.hadoop_fs('test -e {}'.format(path_info['url']), user=path_info['user'])
+            return True
+        except DvcException:
+            return False
+
     def save_info(self, path_info):
         if path_info['scheme'] != 'hdfs':
             raise NotImplementedError
@@ -108,22 +115,52 @@ class RemoteHDFS(RemoteBase):
 
         self.rm(path_info)
 
-    def upload(self, path, path_info):
-        if path_info['scheme'] != 'hdfs':
-            raise NotImplementedError
+    def md5s_to_path_infos(self, md5s):
+        return [{'scheme': 'hdfs',
+                 'url': posixpath.join(self.url, md5[0:2], md5[2:])} for md5 in md5s]
+
+    def upload(self, paths, path_infos, names=None):
+        assert isinstance(paths, list)
+        assert isinstance(path_infos, list)
+        assert len(paths) == len(path_infos)
+        if not names:
+            names = len(paths) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(paths)
+
+        for path, path_info, name in zip(paths, path_infos, names):
+            if path_info['scheme'] != 'hdfs':
+                raise NotImplementedError
+
+            if not os.path.exists(path) or self.exists(path_info):
+                continue
 
-        self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(path_info['url'])), user=path_info['user'])
-        self.hadoop_fs('copyFromLocal {} {}'.format(path, path_info['url']), user=path_info['user'])
+            self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(path_info['url'])), user=path_info['user'])
+            self.hadoop_fs('copyFromLocal {} {}'.format(path, path_info['url']), user=path_info['user'])
 
-    def download(self, path_info, path):
-        if path_info['scheme'] != 'hdfs':
-            raise NotImplementedError
+    def download(self, path_infos, paths, no_progress_bar=False, names=None):
+        assert isinstance(paths, list)
+        assert isinstance(path_infos, list)
+        assert len(paths) == len(path_infos)
+        if not names:
+            names = len(paths) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(paths)
+
+        for path, path_info, name in zip(paths, path_infos, names):
+            if path_info['scheme'] != 'hdfs':
+                raise NotImplementedError
+
+            if os.path.exists(path) or not self.exists(path_info):
+                continue
 
-        dname = os.path.dirname(path)
-        if not os.path.exists(dname):
-            os.makedirs(dname)
+            dname = os.path.dirname(path)
+            if not os.path.exists(dname):
+                os.makedirs(dname)
 
-        self.hadoop_fs('copyToLocal {} {}'.format(path_info['url'], path), user=path_info['user'])
+            self.hadoop_fs('copyToLocal {} {}'.format(path_info['url'], path), user=path_info['user'])
 
     def _path_to_checksum(self, path):
         relpath = posixpath.relpath(path, self.url)
diff --git a/dvc/remote/local.py b/dvc/remote/local.py
index bcc6fcc95..042c5d60e 100644
--- a/dvc/remote/local.py
+++ b/dvc/remote/local.py
@@ -5,12 +5,14 @@ import shutil
 import filecmp
 
 from dvc.system import System
-from dvc.remote.base import RemoteBase
+from dvc.remote.base import RemoteBase, STATUS_MAP
 from dvc.state import State
 from dvc.logger import Logger
-from dvc.utils import remove, move, copyfile, file_md5
+from dvc.utils import remove, move, copyfile, file_md5, to_chunks
 from dvc.config import Config
 from dvc.exceptions import DvcException
+from dvc.progress import progress
+from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
 
 
 class RemoteLOCAL(RemoteBase):
@@ -83,8 +85,15 @@ class RemoteLOCAL(RemoteBase):
 
         return False
 
-    def link(self, src, link, dump=True):
-        dname = os.path.dirname(link)
+    def link(self, md5, path, dump=True):
+        cache = self.get(md5)
+        if not cache or not os.path.exists(cache) or self.changed(md5):
+            if cache:
+                msg = u'Cache \'{}\' not found. File \'{}\' won\'t be created.'
+                Logger.warn(msg.format(md5, os.path.relpath(path)))
+            return
+
+        dname = os.path.dirname(path)
         if not os.path.exists(dname):
             os.makedirs(dname)
 
@@ -95,8 +104,8 @@ class RemoteLOCAL(RemoteBase):
 
         for typ in types:
             try:
-                self.CACHE_TYPE_MAP[typ](src, link)
-                self.link_state.update(link, dump=dump)
+                self.CACHE_TYPE_MAP[typ](cache, path)
+                self.link_state.update(path, dump=dump)
                 return
             except Exception as exc:
                 msg = 'Cache type \'{}\' is not supported'.format(typ)
@@ -130,11 +139,8 @@ class RemoteLOCAL(RemoteBase):
         md5 = checksum_info.get(self.PARAM_MD5, None)
         cache = self.get(md5)
 
-        if not cache or not os.path.exists(cache) or self.changed(md5):
-            if cache:
-                msg = u'Cache \'{}\' not found. File \'{}\' won\'t be created.'
-                Logger.warn(msg.format(md5, os.path.relpath(path)))
-            remove(path)
+        if not cache:
+            Logger.warn('No cache info for \'{}\'. Skipping checkout.'.format(os.path.relpath(path)))
             return
 
         if os.path.exists(path):
@@ -143,10 +149,10 @@ class RemoteLOCAL(RemoteBase):
             remove(path)
 
         msg = u'Checking out \'{}\' with cache \'{}\''
-        Logger.debug(msg.format(os.path.relpath(path), os.path.relpath(cache)))
+        Logger.debug(msg.format(os.path.relpath(path), md5))
 
         if not self.is_dir_cache(cache):
-            self.link(cache, path, dump=True)
+            self.link(md5, path, dump=True)
             return
 
         # Create dir separately so that dir is created
@@ -157,9 +163,8 @@ class RemoteLOCAL(RemoteBase):
         for entry in self.load_dir_cache(cache):
             md5 = entry[self.PARAM_MD5]
             relpath = entry[self.PARAM_RELPATH]
-            c = self.get(md5)
             p = os.path.join(path, relpath)
-            self.link(c, p, dump=False)
+            self.link(md5, p, dump=False)
         self.link_state.dump()
 
     def _move(self, inp, outp):
@@ -238,66 +243,172 @@ class RemoteLOCAL(RemoteBase):
 
         move(from_info['path'], to_info['path'])
 
-    def cache_file_key(self, path):
-        relpath = os.path.relpath(os.path.abspath(path), self.project.cache.local.cache_dir)
-        return os.path.join(self.prefix, relpath)
-
-    def _get_path_info(self, path):
-        key = self.cache_file_key(path)
-        if not os.path.exists(key) or not os.path.isfile(key):
-            return None
-        return {'scheme': 'local',
-                'path': key}
+    def md5s_to_path_infos(self, md5s):
+        return [{'scheme': 'local',
+                 'path': os.path.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s]
+
+    def exists(self, path_infos):
+        ret = []
+        for path_info in path_infos:
+            ret.append(os.path.exists(path_info['path']))
+        return ret
+
+    def upload(self, paths, path_infos, names=None):
+        assert isinstance(paths, list)
+        assert isinstance(path_infos, list)
+        assert len(paths) == len(path_infos)
+        if not names:
+            names = len(paths) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(paths)
 
-    def _new_path_info(self, path):
-        key = self.cache_file_key(path)
-        return {'scheme': 'local',
-                'path': key}
+        for path, path_info, name in zip(paths, path_infos, names):
+            if path_info['scheme'] != 'local':
+                raise NotImplementedError
 
-    def upload(self, path, path_info, name=None):
-        if path_info['scheme'] != 'local':
-            raise NotImplementedError
+            if not os.path.exists(path) or os.path.exists(path_info['path']):
+                continue
 
-        Logger.debug("Uploading '{}' to '{}'".format(path, path_info['path']))
+            Logger.debug("Uploading '{}' to '{}'".format(path, path_info['path']))
 
-        if not name:
-            name = os.path.basename(path)
+            if not name:
+                name = os.path.basename(path)
 
-        self._makedirs(path_info['path'])
+            self._makedirs(path_info['path'])
 
-        try:
-            copyfile(path, path_info['path'], name=name)
-        except Exception as exc:
-            Logger.error("Failed to upload '{}' tp '{}'".format(path, path_info['path']))
-            return None
+            try:
+                copyfile(path, path_info['path'], name=name)
+            except Exception as exc:
+                Logger.error("Failed to upload '{}' tp '{}'".format(path, path_info['path']))
+
+    def download(self, path_infos, paths, no_progress_bar=False, names=None):
+        assert isinstance(paths, list)
+        assert isinstance(path_infos, list)
+        assert len(paths) == len(path_infos)
+        if not names:
+            names = len(paths) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(paths)
 
-        return path
+        for path, path_info, name in zip(paths, path_infos, names):
+            if path_info['scheme'] != 'local':
+                raise NotImplementedError
 
-    def download(self, path_info, path, no_progress_bar=False, name=None):
-        if path_info['scheme'] != 'local':
-            raise NotImplementedError
+            if os.path.exists(path) or not os.path.exists(path_info['path']):
+                continue
 
-        Logger.debug("Downloading '{}' to '{}'".format(path_info['path'], path))
+            Logger.debug("Downloading '{}' to '{}'".format(path_info['path'], path))
 
-        if not name:
-            name = os.path.basename(path)
+            if not name:
+                name = os.path.basename(path)
 
-        self._makedirs(path)
-        tmp_file = self.tmp_file(path)
-        try:
-            copyfile(path_info['path'], tmp_file, no_progress_bar=no_progress_bar, name=name)
-        except Exception as exc:
-            Logger.error("Failed to download '{}' to '{}'".format(path_info['path'], path), exc)
-            return None
+            self._makedirs(path)
+            tmp_file = self.tmp_file(path)
+            try:
+                copyfile(path_info['path'], tmp_file, no_progress_bar=no_progress_bar, name=name)
+            except Exception as exc:
+                Logger.error("Failed to download '{}' to '{}'".format(path_info['path'], path), exc)
+                return
 
-        os.rename(tmp_file, path)
+            os.rename(tmp_file, path)
 
-        return path
+    def _collect(self, checksum_infos):
+        missing = []
+        collected = []
+        for info in checksum_infos:
+            md5 = info[self.PARAM_MD5]
+            cache = self.get(md5)
+            if not self.is_dir_cache(info[self.PARAM_MD5]):
+                continue
+            if not os.path.exists(cache):
+                missing.append(info)
+            collected.extend(self.load_dir_cache(cache))
+        collected.extend(checksum_infos)
+        return collected, missing
 
     def gc(self, checksum_infos):
-        used_md5s = [info[self.PARAM_MD5] for info in checksum_infos]
+        used_md5s = [info[self.PARAM_MD5] for info in self._collect(checksum_infos)[0]]
 
         for md5 in self.all():
             if md5 in used_md5s:
                 continue
             remove(self.get(md5))
+
+    def status(self, checksum_infos, remote, jobs=1):
+        checksum_infos = self._collect(checksum_infos)[0]
+        md5s = [info[self.PARAM_MD5] for info in checksum_infos]
+        chunks = to_chunks(md5s, jobs)
+
+        def worker(chunk):
+            local_exists = [os.path.exists(self.get(md5)) for md5 in chunk]
+            path_infos = remote.md5s_to_path_infos(chunk)
+            remote_exists = remote.exists(path_infos)
+            return [(md5, STATUS_MAP[(l,r)]) for md5, l,r in zip(chunk, local_exists, remote_exists)]
+
+        futures = []
+        with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
+            for chunk in chunks:
+                resp = executor.submit(worker, chunk)
+                futures.append(resp)
+
+        status = []
+        for future in futures:
+            status.extend(future.result())
+
+        return status
+
+    def _do_pull(self, checksum_infos, remote, jobs=1, no_progress_bar=False):
+        md5s = [info[self.PARAM_MD5] for info in checksum_infos]
+        cache = [self.get(md5) for md5 in md5s]
+        path_infos = remote.md5s_to_path_infos(md5s)
+
+        assert len(path_infos) == len(cache) == len(md5s)
+
+        chunks = list(zip(to_chunks(path_infos, jobs),
+                          to_chunks(cache, jobs),
+                          to_chunks(md5s, jobs)))
+
+        progress.set_n_total(len(md5s))
+
+        if len(chunks) == 0:
+            return
+
+        with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
+            for path_infos, paths, md5s in chunks:
+                executor.submit(remote.download, path_infos,
+                                                 paths,
+                                                 names=md5s,
+                                                 no_progress_bar=no_progress_bar)
+
+    def pull(self, checksum_infos, remote, jobs=1):
+        # NOTE: try fetching missing dir info
+        checksum_infos, missing = self._collect(checksum_infos)
+        if len(missing) > 0:
+            self._do_pull(missing, remote, jobs, no_progress_bar=True)
+            checksum_infos += self._collect(missing)[0]
+
+        self._do_pull(checksum_infos, remote, jobs)
+
+    def push(self, checksum_infos, remote, jobs=1):
+        md5s = [info[self.PARAM_MD5] for info in self._collect(checksum_infos)[0]]
+        # NOTE: verifying that our cache is not corrupted
+        md5s = list(filter(lambda md5: not self.changed(md5), md5s))
+        cache = [self.get(md5) for md5 in md5s]
+        path_infos = remote.md5s_to_path_infos(md5s)
+
+        assert len(path_infos) == len(cache) == len(md5s)
+
+        chunks = list(zip(to_chunks(path_infos, jobs),
+                          to_chunks(cache, jobs),
+                          to_chunks(md5s, jobs)))
+
+        progress.set_n_total(len(md5s))
+
+        if len(chunks) == 0:
+            return
+
+        with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
+            for path_infos, paths, md5s in chunks:
+                executor.submit(remote.upload, paths, path_infos, names=md5s)
diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py
index ee481ffe7..88c516bcc 100644
--- a/dvc/remote/s3.py
+++ b/dvc/remote/s3.py
@@ -112,80 +112,112 @@ class RemoteS3(RemoteBase):
         except Exception:
             pass
 
-    def _get_path_info(self, path):
-        key = self.cache_file_key(path)
+    def md5s_to_path_infos(self, md5s):
+        return [{'scheme': self.scheme,
+                 'bucket': self.bucket,
+                 'key': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s]
+
+    def _exists(self, s3, path_info):
         try:
-            self.s3.Object(self.bucket, key).get()
-            return {'scheme': self.scheme,
-                    'bucket': self.bucket,
-                    'key': key}
+            s3.Object(path_info['bucket'], path_info['key']).get()
+            return True
         except Exception:
-            return None
+            return False
+
+    def exists(self, path_infos):
+        ret = []
+        session = boto3.session.Session()
+        s3 = session.resource('s3')
+        for path_info in path_infos:
+            ret.append(self._exists(s3, path_info))
+        return ret
+
+    def upload(self, paths, path_infos, names=None):
+        assert isinstance(paths, list)
+        assert isinstance(path_infos, list)
+        assert len(paths) == len(path_infos)
+        if not names:
+            names = len(paths) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(paths)
 
-    def _new_path_info(self, path):
-        key = self.cache_file_key(path)
-        return {'scheme': self.scheme,
-                'bucket': self.bucket,
-                'key': key}
+        session = boto3.session.Session()
+        s3 = session.resource('s3')
 
-    def upload(self, path, path_info, name=None):
-        if path_info['scheme'] != 's3':
-            raise NotImplementedError
+        for path, path_info, name in zip(paths, path_infos, names):
+            if path_info['scheme'] != 's3':
+                raise NotImplementedError
 
-        Logger.debug("Uploading '{}' to '{}/{}'".format(path,
-                                                        path_info['bucket'],
-                                                        path_info['key']))
+            if not os.path.exists(path) or self._exists(s3, path_info):
+                continue
 
-        if not name:
-            name = os.path.basename(path)
+            Logger.debug("Uploading '{}' to '{}/{}'".format(path,
+                                                            path_info['bucket'],
+                                                            path_info['key']))
 
-        total = os.path.getsize(path)
-        cb = Callback(name, total)
+            if not name:
+                name = os.path.basename(path)
 
-        try:
-            self.s3.Object(path_info['bucket'], path_info['key']).upload_file(path, Callback=cb)
-        except Exception as exc:
-            Logger.error("Failed to upload '{}'".format(path), exc)
-            return None
+            total = os.path.getsize(path)
+            cb = Callback(name, total)
+
+            try:
+                s3.Object(path_info['bucket'], path_info['key']).upload_file(path, Callback=cb)
+            except Exception as exc:
+                Logger.error("Failed to upload '{}'".format(path), exc)
+                continue
 
-        progress.finish_target(name)
+            progress.finish_target(name)
 
-        return path
+    def download(self, path_infos, fnames, no_progress_bar=False, names=None):
+        assert isinstance(fnames, list)
+        assert isinstance(path_infos, list)
+        assert len(fnames) == len(path_infos)
+        if not names:
+            names = len(fnames) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(fnames)
 
-    def download(self, path_info, fname, no_progress_bar=False, name=None):
-        if path_info['scheme'] != 's3':
-            raise NotImplementedError
+        session = boto3.session.Session()
+        s3 = session.resource('s3')
 
-        Logger.debug("Downloading '{}/{}' to '{}'".format(path_info['bucket'],
-                                                          path_info['key'],
-                                                          fname))
+        for fname, path_info, name in zip(fnames, path_infos, names):
+            if path_info['scheme'] != 's3':
+                raise NotImplementedError
 
-        tmp_file = self.tmp_file(fname)
-        if not name:
-            name = os.path.basename(fname)
+            if os.path.exists(fname) or not self._exists(s3, path_info):
+                continue
 
-        if no_progress_bar:
-            cb = None
-        else:
-            total = self.s3.Object(bucket_name=path_info['bucket'],
-                                   key=path_info['key']).content_length
-            cb = Callback(name, total)
+            Logger.debug("Downloading '{}/{}' to '{}'".format(path_info['bucket'],
+                                                              path_info['key'],
+                                                              fname))
 
-        self._makedirs(fname)
+            tmp_file = self.tmp_file(fname)
+            if not name:
+                name = os.path.basename(fname)
 
-        try:
-            self.s3.Object(path_info['bucket'], path_info['key']).download_file(tmp_file, Callback=cb)
-        except Exception as exc:
-            Logger.error("Failed to download '{}/{}'".format(path_info['bucket'],
-                                                             path_info['key']), exc)
-            return None
+            if no_progress_bar:
+                cb = None
+            else:
+                total = s3.Object(bucket_name=path_info['bucket'],
+                                  key=path_info['key']).content_length
+                cb = Callback(name, total)
 
-        os.rename(tmp_file, fname)
+            self._makedirs(fname)
 
-        if not no_progress_bar:
-            progress.finish_target(name)
+            try:
+                s3.Object(path_info['bucket'], path_info['key']).download_file(tmp_file, Callback=cb)
+            except Exception as exc:
+                Logger.error("Failed to download '{}/{}'".format(path_info['bucket'],
+                                                                 path_info['key']), exc)
+                return
+
+            os.rename(tmp_file, fname)
 
-        return fname
+            if not no_progress_bar:
+                progress.finish_target(name)
 
     def _path_to_etag(self, path):
         relpath = posixpath.relpath(path, self.prefix)
diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py
index 4203f9c8b..0d81e2689 100644
--- a/dvc/remote/ssh.py
+++ b/dvc/remote/ssh.py
@@ -57,9 +57,11 @@ class RemoteSSH(RemoteBase):
             self.user = config.get(Config.SECTION_REMOTE_USER, getpass.getuser())
         self.prefix = self.group('path')
 
-    def cache_file_key(self, path):
-        relpath = os.path.relpath(os.path.abspath(path), self.project.cache.local.cache_dir)
-        return posixpath.join(self.prefix, relpath.replace('\\', '/'))
+    def md5s_to_path_infos(self, md5s):
+        return [{'scheme': 'ssh',
+                 'host': self.host,
+                 'user': self.user,
+                 'path': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s]
 
     def ssh(self, host=None, user=None):
         Logger.debug("Establishing ssh connection with '{}' as user '{}'".format(host, user))
@@ -73,54 +75,21 @@ class RemoteSSH(RemoteBase):
 
         return ssh
 
-    def _isfile_remote(self, path_info):
+    def _exists(self, ssh, path_info):
         try:
-            ssh = self.ssh(path_info['host'], path_info['user'])
             ssh.open_sftp().open(path_info['path'], 'r')
-            ssh.close()
             return True
         except IOError as exc:
             if exc.errno != errno.ENOENT:
                 raise
         return False
 
-    def _get_path_info(self, path):
-        key = self.cache_file_key(path)
-        ret = {'scheme': 'ssh',
-               'host': self.host,
-               'user': self.user,
-               'path': key}
-        if self._isfile_remote(ret):
-            return ret
-        return None
-
-    def _do_makedirs_remote(self, sftp, dname):
-        try:
-            sftp.chdir(dname)
-        except IOError as exc:
-            if exc.errno != errno.ENOENT:
-                raise
-
-            parent = posixpath.dirname(dname)
-            if len(parent):
-                self._do_makedirs_remote(sftp, parent)
-
-            sftp.mkdir(dname)
-
-    def _makedirs_remote(self, path_info):
-        dname = posixpath.dirname(path_info['path'])
-        ssh = self.ssh(path_info['host'], path_info['user'])
-        sftp = ssh.open_sftp()
-        self._do_makedirs_remote(sftp, dname)
+    def exists(self, path_infos):
+        ret = []
+        ssh = self.ssh(path_infos[0]['host'], path_infos[0]['user'])
+        for path_info in path_infos:
+            ret.append(self._exists(ssh, path_info))
         ssh.close()
-
-    def _new_path_info(self, path):
-        key = self.cache_file_key(path)
-        ret = {'scheme': 'ssh',
-               'host': self.host,
-               'user': self.user,
-               'path': key}
-        self._makedirs_remote(ret)
         return ret
 
     def _exec(self, ssh, cmd):
@@ -198,57 +167,88 @@ class RemoteSSH(RemoteBase):
         ssh.open_sftp().remove(path_info['path'])
         ssh.close()
 
-    def download(self, path_info, path, no_progress_bar=False, name=None):
-        if path_info['scheme'] != 'ssh':
-            raise NotImplementedError
+    def download(self, path_infos, paths, no_progress_bar=False, names=None):
+        assert isinstance(paths, list)
+        assert isinstance(path_infos, list)
+        assert len(paths) == len(path_infos)
+        if not names:
+            names = len(paths) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(paths)
 
-        Logger.debug("Downloading '{}/{}' to '{}'".format(path_info['host'],
-                                                          path_info['path'],
-                                                          path))
-        if not name:
-            name = os.path.basename(path)
+        ssh = self.ssh(host=path_infos[0]['host'], user=path_infos[0]['user'])
 
-        self._makedirs(path)
-        tmp_file = self.tmp_file(path)
-        try:
-            ssh = self.ssh(host=path_info['host'], user=path_info['user'])
-            ssh.open_sftp().get(path_info['path'], tmp_file, callback=create_cb(name))
-            ssh.close()
-        except Exception as exc:
-            Logger.error("Failed to download '{}/{}' to '{}'".format(path_info['host'],
-                                                                     path_info['path'],
-                                                                     path), exc)
-            return None
+        for path, path_info, name in zip(paths, path_infos, names):
+            if path_info['scheme'] != 'ssh':
+                raise NotImplementedError
 
-        os.rename(tmp_file, path)
-        progress.finish_target(name)
+            if os.path.exists(path) or not self._exists(ssh, path_info):
+                continue
 
-        return path
+            Logger.debug("Downloading '{}/{}' to '{}'".format(path_info['host'],
+                                                              path_info['path'],
+                                                              path))
+            if not name:
+                name = os.path.basename(path)
+
+            self._makedirs(path)
+            tmp_file = self.tmp_file(path)
+            try:
+                ssh.open_sftp().get(path_info['path'], tmp_file, callback=create_cb(name))
+            except Exception as exc:
+                Logger.error("Failed to download '{}/{}' to '{}'".format(path_info['host'],
+                                                                         path_info['path'],
+                                                                         path), exc)
+                continue
 
-    def upload(self, path, path_info, name=None):
-        if path_info['scheme'] != 'ssh':
-            raise NotImplementedError
+            os.rename(tmp_file, path)
+            progress.finish_target(name)
 
-        Logger.debug("Uploading '{}' to '{}/{}'".format(path,
-                                                        path_info['host'],
-                                                        path_info['path']))
+        ssh.close()
 
-        if not name:
-            name = os.path.basename(path)
+    def upload(self, paths, path_infos, names=None):
+        assert isinstance(paths, list)
+        assert isinstance(path_infos, list)
+        assert len(paths) == len(path_infos)
+        if not names:
+            names = len(paths) * [None]
+        else:
+            assert isinstance(names, list)
+            assert len(names) == len(paths)
+
+        ssh = self.ssh(host=path_infos[0]['host'], user=path_infos[0]['user'])
+        sftp = ssh.open_sftp()
 
-        try:
-            ssh = self.ssh(host=path_info['host'], user=path_info['user'])
-            ssh.open_sftp().put(path, path_info['path'], callback=create_cb(name))
-            ssh.close()
-        except Exception as exc:
-            Logger.error("Failed to upload '{}' to '{}/{}'".format(path,
-                                                                   path_info['host'],
-                                                                   path_info['path']), exc)
-            return None
-
-        progress.finish_target(name)
-
-        return path
+        for path, path_info, name in zip(paths, path_infos, names):
+            if path_info['scheme'] != 'ssh':
+                raise NotImplementedError
+
+            if not os.path.exists(path) or self._exists(ssh, path_info):
+                continue
+
+            Logger.debug("Uploading '{}' to '{}/{}'".format(path,
+                                                            path_info['host'],
+                                                            path_info['path']))
+
+            if not name:
+                name = os.path.basename(path)
+
+            dname = posixpath.dirname(path_info['path'])
+            self._exec(ssh, 'mkdir -p {}'.format(dname))
+
+            try:
+                sftp.put(path, path_info['path'], callback=create_cb(name))
+            except Exception as exc:
+                Logger.error("Failed to upload '{}' to '{}/{}'".format(path,
+                                                                       path_info['host'],
+                                                                       path_info['path']), exc)
+                continue
+
+            progress.finish_target(name)
+
+        sftp.close()
+        ssh.close()
 
     def _path_to_md5(self, path):
         relpath = posixpath.relpath(path, self.prefix)
diff --git a/dvc/utils.py b/dvc/utils.py
index a01280909..509cd101c 100644
--- a/dvc/utils.py
+++ b/dvc/utils.py
@@ -4,6 +4,7 @@ Helpers for other modules.
 import os
 import re
 import sys
+import math
 import json
 import stat
 import shutil
@@ -64,23 +65,6 @@ def dict_md5(d):
     return bytes_md5(byts)
 
 
-def cached_property(func):
-    '''A decorator for caching properties in classes.'''
-    def get(self):
-        '''Try obtaining cache'''
-        try:
-            return self._property_cache[func]
-        except AttributeError:
-            self._property_cache = {}
-            ret = self._property_cache[func] = func(self)
-            return ret
-        except KeyError:
-            ret = self._property_cache[func] = func(self)
-            return ret
-
-    return property(get)
-
-
 def copyfile(src, dest, no_progress_bar=False, name=None):
     '''Copy file with progress bar'''
     copied = 0
@@ -110,32 +94,6 @@ def copyfile(src, dest, no_progress_bar=False, name=None):
     fdest.close()
 
 
-def wrap(func, t):
-    try:
-        return func(t)
-    except Exception as exc:
-        Logger.error('Error', exc)
-        raise
-
-
-def map_progress(func, targets, n_threads):
-    """
-    Process targets in multi-threaded mode with progress bar
-    """
-    progress.set_n_total(len(targets))
-    pool = ThreadPool(processes=n_threads)
-    ret = []
-
-    wrapper = lambda t: wrap(func, t)
-
-    try:
-        ret = pool.map(wrapper, targets)
-    except Exception as exc:
-        raise
-
-    return list(zip(targets, ret))
-
-
 def move(src, dst):
     dname = os.path.dirname(dst)
     if not os.path.exists(dname):
@@ -153,3 +111,15 @@ def remove(path):
         os.unlink(path)
     else:
         shutil.rmtree(path)
+
+
+def to_chunks(l, jobs):
+    n = int(math.ceil(len(l) / jobs))
+
+    if len(l) == 1:
+        return [l]
+
+    if n == 0:
+        n = 1
+
+    return [l[x:x+n] for x in range(0, len(l), n)]


Test patch:
diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py
index 9f111a414..2c1c513f9 100644
--- a/tests/test_data_cloud.py
+++ b/tests/test_data_cloud.py
@@ -10,7 +10,7 @@ import platform
 from dvc.main import main
 from dvc.config import Config, ConfigError
 from dvc.data_cloud import DataCloud, RemoteS3, RemoteGS, RemoteLOCAL, RemoteSSH
-from dvc.remote.base import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED
+from dvc.remote.base import STATUS_OK, STATUS_NEW, STATUS_DELETED
 
 from tests.basic_env import TestDvc
 
@@ -130,10 +130,6 @@ class TestDataCloudBase(TestDvc):
     def _get_url(self):
         return None
 
-    @property
-    def cloud_class(self):
-        return None
-
     def _setup_cloud(self):
         if not self._should_test():
             return
@@ -142,76 +138,72 @@ class TestDataCloudBase(TestDvc):
 
         config = TEST_CONFIG
         config[TEST_SECTION][Config.SECTION_REMOTE_URL] = repo
-        self.cloud = self._get_cloud_class()(self.dvc, config[TEST_SECTION])
+        self.cloud = DataCloud(self.dvc, config)
+
+        self.assertIsInstance(self.cloud._cloud, self._get_cloud_class())
 
     def _test_cloud(self):
         self._setup_cloud()
 
         stage = self.dvc.add(self.FOO)
         cache = stage.outs[0].cache
+        info = stage.outs[0].info
+        md5 = info['md5']
 
         stage_dir = self.dvc.add(self.DATA_DIR)
         cache_dir = stage_dir.outs[0].cache
+        info_dir = stage_dir.outs[0].info
+        md5_dir = info_dir['md5']
 
         # Check status
         sleep()
-        status = self.cloud.status(cache)
-        self.assertEqual(status, STATUS_NEW)
+        status = self.cloud.status([info])
+        self.assertEqual([(md5, STATUS_NEW)], status)
 
-        status_dir = self.cloud.status(cache_dir)
-        self.assertEqual(status_dir, STATUS_NEW)
+        status_dir = self.cloud.status([info_dir])
+        self.assertTrue((md5_dir, STATUS_NEW) in status_dir)
 
         # Push and check status
-        self.cloud.push(cache)
+        self.cloud.push([info])
         self.assertTrue(os.path.exists(cache))
         self.assertTrue(os.path.isfile(cache))
 
-        self.cloud.push(cache_dir)
+        self.cloud.push([info_dir])
         self.assertTrue(os.path.isfile(cache_dir))
 
         sleep()
-        status = self.cloud.status(cache)
-        self.assertEqual(status, STATUS_OK)
+        status = self.cloud.status([info])
+        self.assertEqual([(md5, STATUS_OK)], status)
 
-        status_dir = self.cloud.status(cache_dir)
-        self.assertEqual(status_dir, STATUS_OK)
-
-        # Modify and check status
-        sleep()
-        with open(cache, 'a') as fd:
-            fd.write('addon')
-        status = self.cloud.status(cache)
-        # NOTE: this is what corrupted cache looks like and it should
-        # be removed automatically, thus causing STATUS_DELETED instead
-        # of STATUS_MODIFIED.
-        self.assertEqual(status, STATUS_DELETED)
+        status_dir = self.cloud.status([info_dir])
+        self.assertTrue((md5_dir, STATUS_OK) in status_dir)
 
         # Remove and check status
         sleep()
         shutil.rmtree(self.dvc.cache.local.cache_dir)
 
-        status = self.cloud.status(cache)
-        self.assertEqual(status, STATUS_DELETED)
+        status = self.cloud.status([info])
+        self.assertEqual([(md5, STATUS_DELETED)], status)
 
-        status_dir = self.cloud.status(cache_dir)
-        self.assertEqual(status_dir, STATUS_DELETED)
+        status_dir = self.cloud.status([info_dir])
+        self.assertEqual([(md5_dir, STATUS_DELETED)], status_dir)
 
         # Pull and check status
-        self.cloud.pull(cache)
+        self.cloud.pull([info])
         self.assertTrue(os.path.exists(cache))
         self.assertTrue(os.path.isfile(cache))
         with open(cache, 'r') as fd:
             self.assertEqual(fd.read(), self.FOO_CONTENTS)
 
-        self.cloud.pull(cache_dir)
+        self.cloud.pull([info_dir])
         self.assertTrue(os.path.isfile(cache_dir))
 
         sleep()
-        status = self.cloud.status(cache)
-        self.assertEqual(status, STATUS_OK)
+        status = self.cloud.status([info])
+        self.assertEqual([(md5, STATUS_OK)], status)
 
-        status_dir = self.cloud.status(cache_dir)
-        self.assertEqual(status_dir, STATUS_OK)
+        status_dir = self.cloud.status([info_dir])
+        self.assertTrue((md5_dir, STATUS_OK) in status_dir)
 
     def test(self):
         if self._should_test():
@@ -322,23 +314,21 @@ class TestDataCloudCLIBase(TestDvc):
     def _test(self):
         pass
 
-    def _test_compat(self):
-        pass
-
     def test(self):
         if self._should_test():
             self._test()
-            self._test_compat()
 
 
-class TestRemoteLOCALCLI(TestDataCloudCLIBase):
-    def _test_compat(self):
+class TestCompatRemoteLOCALCLI(TestDataCloudCLIBase):
+    def _test(self):
         storagepath = get_local_storagepath()
         self.main(['config', 'core.cloud', 'local'])
         self.main(['config', 'local.storagepath', storagepath])
 
         self._test_cloud()
 
+
+class TestRemoteLOCALCLI(TestDataCloudCLIBase):
     def _test(self):
         url = get_local_url()
 
@@ -351,10 +341,6 @@ class TestRemoteSSHCLI(TestDataCloudCLIBase):
     def _should_test(self):
         return _should_test_ssh()
 
-    def _test_compat(self):
-        # We do not support ssh in legacy mode
-        pass
-
     def _test(self):
         url = get_ssh_url()
 
@@ -363,17 +349,22 @@ class TestRemoteSSHCLI(TestDataCloudCLIBase):
         self._test_cloud(TEST_REMOTE)
 
 
-class TestRemoteS3CLI(TestDataCloudCLIBase):
+class TestCompatRemoteS3CLI(TestDataCloudCLIBase):
     def _should_test(self):
         return _should_test_aws()
 
-    def _test_compat(self):
+    def _test(self):
         storagepath = get_aws_storagepath()
         self.main(['config', 'core.cloud', 'aws'])
         self.main(['config', 'aws.storagepath', storagepath])
 
         self._test_cloud()
 
+
+class TestRemoteS3CLI(TestDataCloudCLIBase):
+    def _should_test(self):
+        return _should_test_aws()
+
     def _test(self):
         url = get_aws_url()
 
@@ -382,17 +373,22 @@ class TestRemoteS3CLI(TestDataCloudCLIBase):
         self._test_cloud(TEST_REMOTE)
 
 
-class TestRemoteGSCLI(TestDataCloudCLIBase):
+class TestCompatRemoteGSCLI(TestDataCloudCLIBase):
     def _should_test(self):
         return _should_test_gcp()
 
-    def _test_compat(self):
+    def _test(self):
         storagepath = get_gcp_storagepath()
         self.main(['config', 'core.cloud', 'gcp'])
         self.main(['config', 'gcp.storagepath', storagepath])
 
         self._test_cloud()
 
+
+class TestRemoteGSCLI(TestDataCloudCLIBase):
+    def _should_test(self):
+        return _should_test_gcp()
+
     def _test(self):
         url = get_gcp_url()
 
@@ -408,7 +404,7 @@ class TestDataCloudErrorCLI(TestDvc):
 
     def test_error(self):
         f = 'non-existing-file'
-        self.main_fail(['status', f])
+        self.main_fail(['status', '-c', f])
         self.main_fail(['push', f])
         self.main_fail(['pull', f])
         self.main_fail(['fetch', f])
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 4b4527474..045b803d0 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -27,17 +27,6 @@ class TestUtils(TestDvc):
         os.remove(src)
         os.remove(dest)
 
-    def test_map_progress(self):
-        def f(target):
-            with open(target, 'w+') as o:
-                o.write(target)
-
-        targets = ['map{}'.format(i) for i in range(1, 10)]
-        n_threads = [1, 10, 20]
-
-        for n in n_threads:
-            utils.map_progress(f, targets, n)
-
     def test_file_md5_crlf(self):
         with open('cr', 'wb+') as fd:
             fd.write(b'a\nb\nc')