remote: base: modify cache_exists to use exists instead of batch_exists

dvc/remote/azure.py

@@ -1,12 +1,12 @@
 from __future__ import absolute_import
 from __future__ import unicode_literals
 
-import itertools
 import os
 import re
 import logging
 from datetime import datetime, timedelta
 
+
 from dvc.scheme import Schemes
 
 try:
@@ -138,6 +138,10 @@ class RemoteAZURE(RemoteBASE):
             from_info.bucket, from_info.path, to_file, progress_callback=cb
         )
 
+    def exists(self, path_info):
+        paths = self._list_paths(path_info.bucket, path_info.path)
+        return any(path_info.path == path for path in paths)
+
     def open(self, path_info, mode="r", encoding=None):
         get_url = lambda: self._generate_download_url(path_info)  # noqa: E731
         return open_url(get_url, mode=mode, encoding=encoding)

dvc/remote/base.py

@@ -20,6 +20,7 @@ from dvc.exceptions import (
     DvcIgnoreInCollectedDirError,
 )
 from dvc.progress import progress, ProgressCallback
+from dvc.utils import LARGE_DIR_SIZE, tmp_fname, move, relpath
 from dvc.utils import (
     LARGE_DIR_SIZE,
     tmp_fname,
@@ -644,15 +645,10 @@ class RemoteBASE(object):
         """
         progress_callback = ProgressCallback(len(checksums))
 
-        def exists_with_progress(chunks):
-            return self.batch_exists(chunks, callback=progress_callback)
-
-        if self.no_traverse and hasattr(self, "batch_exists"):
+        if self.no_traverse:
             with ThreadPoolExecutor(max_workers=jobs or self.JOBS) as executor:
                 path_infos = [self.checksum_to_path_info(x) for x in checksums]
-                chunks = to_chunks(path_infos, num_chunks=self.JOBS)
-                results = executor.map(exists_with_progress, chunks)
-                in_remote = itertools.chain.from_iterable(results)
+                in_remote = list(executor.map(self.exists, path_infos))
                 ret = list(itertools.compress(checksums, in_remote))
                 progress_callback.finish("")
                 return ret

dvc/remote/gs.py

@@ -1,7 +1,6 @@
 from __future__ import unicode_literals
 
 import logging
-import itertools
 from funcy import cached_property
 
 try:
@@ -89,17 +88,6 @@ class RemoteGS(RemoteBASE):
         paths = set(self._list_paths(path_info.bucket, path_info.path))
         return any(path_info.path == path for path in paths)
 
-    def batch_exists(self, path_infos, callback):
-        paths = []
-
-        for path_info in path_infos:
-            paths.append(self._list_paths(path_info.bucket, path_info.path))
-            callback.update(str(path_info))
-
-        paths = set(itertools.chain.from_iterable(paths))
-
-        return [path_info.path in paths for path_info in path_infos]
-
     def _upload(self, from_file, to_info, **_kwargs):
         bucket = self.gs.bucket(to_info.bucket)
         blob = bucket.blob(to_info.path)

dvc/remote/http.py

@@ -65,15 +65,6 @@ class RemoteHTTP(RemoteBASE):
     def exists(self, path_info):
         return bool(self._request("HEAD", path_info.url))
 
-    def batch_exists(self, path_infos, callback):
-        results = []
-
-        for path_info in path_infos:
-            results.append(self.exists(path_info))
-            callback.update(str(path_info))
-
-        return results
-
     def _content_length(self, url):
         return self._request("HEAD", url).headers.get("Content-Length")
 

dvc/remote/oss.py

@@ -119,3 +119,7 @@ class RemoteOSS(RemoteBASE):
         self.oss_service.get_object_to_file(
             from_info.path, to_file, progress_callback=cb
         )
+
+    def exists(self, path_info):
+        paths = self._list_paths(path_info.bucket, path_info.path)
+        return any(path_info.path == path for path in paths)

dvc/remote/s3.py

@@ -3,7 +3,6 @@ from __future__ import unicode_literals
 import os
 import threading
 import logging
-import itertools
 from funcy import cached_property
 
 try:
@@ -216,16 +215,6 @@ class RemoteS3(RemoteBASE):
         paths = self._list_paths(path_info.bucket, path_info.path)
         return any(path_info.path == path for path in paths)
 
-    def batch_exists(self, path_infos, callback):
-        paths = []
-
-        for path_info in path_infos:
-            paths.append(self._list_paths(path_info.bucket, path_info.path))
-            callback.update(str(path_info))
-
-        paths = set(itertools.chain.from_iterable(paths))
-        return [path_info.path in paths for path_info in path_infos]
-
     def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
         total = os.path.getsize(from_file)
         cb = None if no_progress_bar else Callback(name, total)

dvc/remote/ssh/__init__.py

@@ -3,9 +3,6 @@ from __future__ import unicode_literals
 import os
 import getpass
 import logging
-import itertools
-import errno
-from concurrent.futures import ThreadPoolExecutor
 import threading
 
 try:
@@ -15,7 +12,6 @@ except ImportError:
 
 import dvc.prompt as prompt
 from dvc.config import Config
-from dvc.utils import to_chunks
 from dvc.utils.compat import urlparse, StringIO
 from dvc.remote.base import RemoteBASE
 from dvc.scheme import Schemes
@@ -148,34 +144,6 @@ class RemoteSSH(RemoteBASE):
         with self.ssh(path_info) as ssh:
             return ssh.exists(path_info.path)
 
-    def batch_exists(self, path_infos, callback):
-        def _exists(chunk_and_channel):
-            chunk, channel = chunk_and_channel
-            ret = []
-            for path in chunk:
-                try:
-                    channel.stat(path)
-                    ret.append(True)
-                except IOError as exc:
-                    if exc.errno != errno.ENOENT:
-                        raise
-                    ret.append(False)
-                callback.update(path)
-            return ret
-
-        with self.ssh(path_infos[0]) as ssh:
-            channels = ssh.open_max_sftp_channels()
-            max_workers = len(channels)
-
-            with ThreadPoolExecutor(max_workers=max_workers) as executor:
-                paths = [path_info.path for path_info in path_infos]
-                chunks = to_chunks(paths, num_chunks=max_workers)
-                chunks_and_channels = zip(chunks, channels)
-                outcome = executor.map(_exists, chunks_and_channels)
-                results = list(itertools.chain.from_iterable(outcome))
-
-            return results
-
     def get_file_checksum(self, path_info):
         if path_info.scheme != self.scheme:
             raise NotImplementedError