diff --git a/dvc/command/version.py b/dvc/command/version.py
index ad041b4dd..3e2b780f0 100644
--- a/dvc/command/version.py
+++ b/dvc/command/version.py
@@ -5,6 +5,7 @@ import platform
 import argparse
 import logging
 import uuid
+import itertools
 
 try:
     import psutil
@@ -69,9 +70,12 @@ class CmdVersion(CmdBaseNoRepo):
     def get_fs_type(path):
         partition = {
             pathlib.Path(part.mountpoint): (part.fstype, part.device)
-            for part in psutil.disk_partitions()
+            for part in psutil.disk_partitions(all=True)
         }
-        for parent in pathlib.Path(path).parents:
+
+        path = pathlib.Path(path)
+
+        for parent in itertools.chain([path], path.parents):
             if parent in partition:
                 return partition[parent]
         return ("unkown", "none")
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
index 207b94703..73d0710ac 100644
--- a/dvc/remote/base.py
+++ b/dvc/remote/base.py
@@ -12,6 +12,8 @@ from operator import itemgetter
 from multiprocessing import cpu_count
 from functools import partial
 from concurrent.futures import ThreadPoolExecutor
+from copy import copy
+from dvc.remote.slow_link_detection import slow_link_guard
 
 import dvc.prompt as prompt
 from dvc.config import Config
@@ -79,6 +81,7 @@ class RemoteBASE(object):
     PARAM_RELPATH = "relpath"
     CHECKSUM_DIR_SUFFIX = ".dir"
     CHECKSUM_JOBS = max(1, min(4, cpu_count() // 2))
+    DEFAULT_CACHE_TYPES = ["copy"]
 
     state = StateNoop()
 
@@ -119,6 +122,15 @@ class RemoteBASE(object):
         self.no_traverse = config.get(Config.SECTION_REMOTE_NO_TRAVERSE, True)
         self._dir_info = {}
 
+        types = config.get(Config.SECTION_CACHE_TYPE, None)
+        if types:
+            if isinstance(types, str):
+                types = [t.strip() for t in types.split(",")]
+            self.cache_types = types
+        else:
+            self.cache_types = copy(self.DEFAULT_CACHE_TYPES)
+        self.cache_type_confirmed = False
+
     def __repr__(self):
         return "{class_name}: '{path_info}'".format(
             class_name=type(self).__name__,
@@ -352,7 +364,47 @@ class RemoteBASE(object):
         return False
 
     def link(self, from_info, to_info):
-        self.copy(from_info, to_info)
+        self._link(from_info, to_info, self.cache_types)
+
+    def _link(self, from_info, to_info, link_types):
+        assert self.isfile(from_info)
+
+        self.makedirs(to_info.parent)
+
+        self._try_links(from_info, to_info, link_types)
+
+    @slow_link_guard
+    def _try_links(self, from_info, to_info, link_types):
+        while link_types:
+            link_method = getattr(self, link_types[0])
+            try:
+                self._do_link(from_info, to_info, link_method)
+                self.cache_type_confirmed = True
+                return
+
+            except DvcException as exc:
+                msg = "Cache type '{}' is not supported: {}"
+                logger.debug(msg.format(link_types[0], str(exc)))
+                del link_types[0]
+
+        raise DvcException("no possible cache types left to try out.")
+
+    def _do_link(self, from_info, to_info, link_method):
+        if self.exists(to_info):
+            raise DvcException("Link '{}' already exists!".format(to_info))
+
+        link_method(from_info, to_info)
+
+        if self.protected:
+            self.protect(to_info)
+
+        msg = "Created {}'{}': {} -> {}".format(
+            "protected " if self.protected else "",
+            self.cache_types[0],
+            from_info,
+            to_info,
+        )
+        logger.debug(msg)
 
     def _save_file(self, path_info, checksum, save_link=True):
         assert checksum
@@ -391,9 +443,15 @@ class RemoteBASE(object):
         return False
 
     def isfile(self, path_info):
-        raise NotImplementedError
+        """Optional: Overwrite only if the remote has a way to distinguish
+        between a directory and a file.
+        """
+        return True
 
     def isdir(self, path_info):
+        """Optional: Overwrite only if the remote has a way to distinguish
+        between a directory and a file.
+        """
         return False
 
     def walk(self, path_info):
@@ -512,6 +570,15 @@ class RemoteBASE(object):
     def copy(self, from_info, to_info):
         raise RemoteActionNotImplemented("copy", self.scheme)
 
+    def symlink(self, from_info, to_info):
+        raise RemoteActionNotImplemented("symlink", self.scheme)
+
+    def hardlink(self, from_info, to_info):
+        raise RemoteActionNotImplemented("hardlink", self.scheme)
+
+    def reflink(self, from_info, to_info):
+        raise RemoteActionNotImplemented("reflink", self.scheme)
+
     def exists(self, path_info):
         raise NotImplementedError
 
@@ -720,7 +787,10 @@ class RemoteBASE(object):
         return True
 
     def makedirs(self, path_info):
-        raise NotImplementedError
+        """Optional: Implement only if the remote needs to create
+        directories before copying/linking/moving data
+        """
+        pass
 
     def _checkout_dir(
         self, path_info, checksum, force, progress_callback=None
diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py
index 43e679dbb..1c1256efd 100644
--- a/dvc/remote/local/__init__.py
+++ b/dvc/remote/local/__init__.py
@@ -1,16 +1,12 @@
 from __future__ import unicode_literals
 
-from copy import copy
-
 from dvc.scheme import Schemes
-from dvc.remote.local.slow_link_detection import slow_link_guard
 from dvc.utils.compat import str, fspath_py35, open
 
 import os
 import stat
 import errno
 from shortuuid import uuid
-import shutil
 import logging
 from functools import partial
 
@@ -51,12 +47,6 @@ class RemoteLOCAL(RemoteBASE):
     UNPACKED_DIR_SUFFIX = ".unpacked"
 
     DEFAULT_CACHE_TYPES = ["reflink", "copy"]
-    CACHE_TYPE_MAP = {
-        "copy": shutil.copyfile,
-        "symlink": System.symlink,
-        "hardlink": System.hardlink,
-        "reflink": System.reflink,
-    }
 
     SHARED_MODE_MAP = {None: (0o644, 0o755), "group": (0o664, 0o775)}
 
@@ -71,15 +61,6 @@ class RemoteLOCAL(RemoteBASE):
             # cache files are set to be read-only for everyone
             self._file_mode = stat.S_IREAD | stat.S_IRGRP | stat.S_IROTH
 
-        types = config.get(Config.SECTION_CACHE_TYPE, None)
-        if types:
-            if isinstance(types, str):
-                types = [t.strip() for t in types.split(",")]
-            self.cache_types = types
-        else:
-            self.cache_types = copy(self.DEFAULT_CACHE_TYPES)
-        self.cache_type_confirmed = False
-
         # A clunky way to detect cache dir
         storagepath = config.get(Config.SECTION_LOCAL_STORAGEPATH, None)
         cache_dir = config.get(Config.SECTION_REMOTE_URL, storagepath)
@@ -126,80 +107,14 @@ class RemoteLOCAL(RemoteBASE):
 
         return self.checksum_to_path_info(md5).url
 
-    def exists(self, path_info):
+    @staticmethod
+    def exists(path_info):
         assert path_info.scheme == "local"
         return os.path.lexists(fspath_py35(path_info))
 
     def makedirs(self, path_info):
         makedirs(path_info, exist_ok=True, mode=self._dir_mode)
 
-    def link(self, from_info, to_info):
-        self._link(from_info, to_info, self.cache_types)
-
-    def _link(self, from_info, to_info, link_types):
-        from_path = from_info.fspath
-        to_path = to_info.fspath
-
-        assert os.path.isfile(from_path)
-
-        dname = os.path.dirname(to_path)
-        if not os.path.exists(dname):
-            os.makedirs(dname)
-
-        # NOTE: just create an empty file for an empty cache
-        if os.path.getsize(from_path) == 0:
-            open(to_path, "w+").close()
-
-            msg = "Created empty file: {} -> {}".format(from_path, to_path)
-            logger.debug(msg)
-            return
-
-        self._try_links(from_info, to_info, link_types)
-
-    @classmethod
-    def _get_link_method(cls, link_type):
-        try:
-            return cls.CACHE_TYPE_MAP[link_type]
-        except KeyError:
-            raise DvcException(
-                "Cache type: '{}' not supported!".format(link_type)
-            )
-
-    def _do_link(self, from_info, to_info, link_method):
-        if self.exists(to_info):
-            raise DvcException("Link '{}' already exists!".format(to_info))
-        else:
-            link_method(from_info.fspath, to_info.fspath)
-
-        if self.protected:
-            self.protect(to_info)
-
-        msg = "Created {}'{}': {} -> {}".format(
-            "protected " if self.protected else "",
-            self.cache_types[0],
-            from_info,
-            to_info,
-        )
-        logger.debug(msg)
-
-    @slow_link_guard
-    def _try_links(self, from_info, to_info, link_types):
-        i = len(link_types)
-        while i > 0:
-            link_method = self._get_link_method(link_types[0])
-            try:
-                self._do_link(from_info, to_info, link_method)
-                self.cache_type_confirmed = True
-                return
-
-            except DvcException as exc:
-                msg = "Cache type '{}' is not supported: {}"
-                logger.debug(msg.format(link_types[0], str(exc)))
-                del link_types[0]
-                i -= 1
-
-        raise DvcException("no possible cache types left to try out.")
-
     def already_cached(self, path_info):
         assert path_info.scheme in ["", "local"]
 
@@ -221,12 +136,18 @@ class RemoteLOCAL(RemoteBASE):
 
         return False
 
-    def isfile(self, path_info):
+    @staticmethod
+    def isfile(path_info):
         return os.path.isfile(fspath_py35(path_info))
 
-    def isdir(self, path_info):
+    @staticmethod
+    def isdir(path_info):
         return os.path.isdir(fspath_py35(path_info))
 
+    @staticmethod
+    def getsize(path_info):
+        return os.path.getsize(fspath_py35(path_info))
+
     def walk(self, path_info):
         return dvc_walk(path_info, self.repo.dvcignore)
 
@@ -253,6 +174,44 @@ class RemoteLOCAL(RemoteBASE):
 
         move(from_info, to_info, mode=mode)
 
+    @staticmethod
+    def copy(from_info, to_info):
+        System.copy(from_info, to_info)
+
+    @staticmethod
+    def symlink(from_info, to_info):
+        System.symlink(from_info, to_info)
+
+    def hardlink(self, from_info, to_info):
+        # If there are a lot of empty files (which happens a lot in datasets),
+        # and the cache type is `hardlink`, we might reach link limits and
+        # will get something like: `too many links error`
+        #
+        # This is because all those empty files will have the same checksum
+        # (i.e. 68b329da9893e34099c7d8ad5cb9c940), therfore, they will be
+        # linked to the same file in the cache.
+        #
+        # From https://en.wikipedia.org/wiki/Hard_link
+        #   * ext4 limits the number of hard links on a file to 65,000
+        #   * Windows with NTFS has a limit of 1024 hard links on a file
+        #
+        # That's why we simply create an empty file rather than a link.
+        if self.getsize(from_info) == 0:
+            self.open(to_info, "w").close()
+
+            logger.debug(
+                "Created empty file: {src} -> {dest}".format(
+                    src=str(from_info), dest=str(to_info)
+                )
+            )
+            return
+
+        System.hardlink(from_info, to_info)
+
+    @staticmethod
+    def reflink(from_info, to_info):
+        System.reflink(from_info, to_info)
+
     def cache_exists(self, checksums, jobs=None, name=None):
         return [
             checksum
@@ -286,8 +245,8 @@ class RemoteLOCAL(RemoteBASE):
             name=name,
         )
 
-    def open(self, path_info, mode="r", encoding=None):
-        assert mode in {"r", "rt", "rb"}
+    @staticmethod
+    def open(path_info, mode="r", encoding=None):
         return open(fspath_py35(path_info), mode=mode, encoding=encoding)
 
     def _group(self, checksum_infos, show_checksums=False):
diff --git a/dvc/remote/local/slow_link_detection.py b/dvc/remote/local/slow_link_detection.py
deleted file mode 100644
index 7bd3bb143..000000000
--- a/dvc/remote/local/slow_link_detection.py
+++ /dev/null
@@ -1,62 +0,0 @@
-import logging
-import time
-
-import colorama
-from dvc.config import Config
-
-logger = logging.getLogger(__name__)
-
-
-class SlowLinkDetectorDecorator(object):
-    LINKING_TIMEOUT_SECONDS = 10.0
-    was_displayed = False
-
-    @classmethod
-    def should_display(cls):
-        if not cls.was_displayed:
-            cls.was_displayed = True
-            return True
-        return False
-
-    def __init__(self, method):
-        self.method = method
-
-    def __call__(self, *args, **kwargs):
-        start = time.time()
-        result = self.method(*args, **kwargs)
-        execution_time_seconds = time.time() - start
-
-        if (
-            execution_time_seconds >= self.LINKING_TIMEOUT_SECONDS
-            and self.should_display()
-        ):
-            msg = (
-                "You can cut execution time considerably by using a different"
-                " `cache.type` configuration.\nSee "
-                "{blue}https://dvc.org/doc/commands-reference/config#cache{"
-                "reset} for more information.\n"
-                "To disable this message, run:\n"
-                "'dvc config cache.slow_link_warning false'".format(
-                    blue=colorama.Fore.BLUE, reset=colorama.Fore.RESET
-                )
-            )
-            logger.warning(msg)
-
-        return result
-
-
-def slow_link_guard(method):
-    def call(remote_local, *args, **kwargs):
-        cache_config = remote_local.repo.config.config.get(
-            Config.SECTION_CACHE, {}
-        )
-        should_warn = cache_config.get(
-            Config.SECTION_CACHE_SLOW_LINK_WARNING, True
-        ) and not cache_config.get(Config.SECTION_CACHE_TYPE, None)
-
-        if should_warn:
-            decorated = SlowLinkDetectorDecorator(method)
-            return decorated(remote_local, *args, **kwargs)
-        return method(remote_local, *args, **kwargs)
-
-    return call
diff --git a/dvc/remote/slow_link_detection.py b/dvc/remote/slow_link_detection.py
new file mode 100644
index 000000000..0dd2285d6
--- /dev/null
+++ b/dvc/remote/slow_link_detection.py
@@ -0,0 +1,50 @@
+import logging
+import time
+import sys
+from functools import wraps
+
+import colorama
+from dvc.config import Config
+
+
+logger = logging.getLogger(__name__)
+this = sys.modules[__name__]
+
+this.timeout_seconds = 10.0
+this.already_displayed = False
+this.message = (
+    "You can cut execution time considerably by using a different "
+    "`cache.type` configuration.\n"
+    "See {blue}https://dvc.org/doc/commands-reference/config#cache{reset} "
+    "for more information.\n"
+    "To disable this message, run:\n"
+    "'dvc config cache.slow_link_warning false'".format(
+        blue=colorama.Fore.BLUE, reset=colorama.Fore.RESET
+    )
+)
+
+
+def slow_link_guard(f):
+    @wraps(f)
+    def wrapper(remote, *args, **kwargs):
+        if this.already_displayed:
+            return f(remote, *args, **kwargs)
+
+        config = remote.repo.config.config.get(Config.SECTION_CACHE, {})
+        cache_type = config.get(Config.SECTION_CACHE_TYPE)
+        should_warn = config.get(Config.SECTION_CACHE_SLOW_LINK_WARNING, True)
+
+        if not should_warn or cache_type:
+            return f(remote, *args, **kwargs)
+
+        start = time.time()
+        result = f(remote, *args, **kwargs)
+        delta = time.time() - start
+
+        if delta >= this.timeout_seconds:
+            logger.warning(this.message)
+            this.already_displayed = True
+
+        return result
+
+    return wrapper
diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py
index 74280d7bb..233f133ee 100644
--- a/dvc/remote/ssh/__init__.py
+++ b/dvc/remote/ssh/__init__.py
@@ -47,9 +47,10 @@ class RemoteSSH(RemoteBASE):
     # We use conservative setting of 4 instead to not exhaust max sessions.
     CHECKSUM_JOBS = 4
 
+    DEFAULT_CACHE_TYPES = ["copy"]
+
     def __init__(self, repo, config):
         super(RemoteSSH, self).__init__(repo, config)
-
         url = config.get(Config.SECTION_REMOTE_URL)
         if url:
             parsed = urlparse(url)
@@ -164,12 +165,54 @@ class RemoteSSH(RemoteBASE):
         with self.ssh(path_info) as ssh:
             return ssh.isdir(path_info.path)
 
+    def isfile(self, path_info):
+        with self.ssh(path_info) as ssh:
+            return ssh.isfile(path_info.path)
+
+    def getsize(self, path_info):
+        with self.ssh(path_info) as ssh:
+            return ssh.getsize(path_info.path)
+
     def copy(self, from_info, to_info):
+        if not from_info.scheme == to_info.scheme == self.scheme:
+            raise NotImplementedError
+
+        with self.ssh(from_info) as ssh:
+            ssh.copy(from_info.path, to_info.path)
+
+    def symlink(self, from_info, to_info):
+        if not from_info.scheme == to_info.scheme == self.scheme:
+            raise NotImplementedError
+
+        with self.ssh(from_info) as ssh:
+            ssh.symlink(from_info.path, to_info.path)
+
+    def hardlink(self, from_info, to_info):
+        if not from_info.scheme == to_info.scheme == self.scheme:
+            raise NotImplementedError
+
+        # See dvc/remote/local/__init__.py - hardlink()
+        if self.getsize(from_info) == 0:
+
+            with self.ssh(to_info) as ssh:
+                ssh.sftp.open(to_info.path, "w").close()
+
+            logger.debug(
+                "Created empty file: {src} -> {dest}".format(
+                    src=str(from_info), dest=str(to_info)
+                )
+            )
+            return
+
+        with self.ssh(from_info) as ssh:
+            ssh.hardlink(from_info.path, to_info.path)
+
+    def reflink(self, from_info, to_info):
         if from_info.scheme != self.scheme or to_info.scheme != self.scheme:
             raise NotImplementedError
 
         with self.ssh(from_info) as ssh:
-            ssh.cp(from_info.path, to_info.path)
+            ssh.reflink(from_info.path, to_info.path)
 
     def remove(self, path_info):
         if path_info.scheme != self.scheme:
@@ -207,12 +250,12 @@ class RemoteSSH(RemoteBASE):
 
     @contextmanager
     def open(self, path_info, mode="r", encoding=None):
-        assert mode in {"r", "rt", "rb"}
+        assert mode in {"r", "rt", "rb", "wb"}
 
         with self.ssh(path_info) as ssh, closing(
-            ssh.sftp.file(path_info.path, mode="r")
+            ssh.sftp.open(path_info.path, mode)
         ) as fd:
-            if mode == "rb":
+            if "b" in mode:
                 yield fd
             else:
                 yield io.TextIOWrapper(fd, encoding=encoding)
diff --git a/dvc/remote/ssh/connection.py b/dvc/remote/ssh/connection.py
index 43b380ada..4e1e52a43 100644
--- a/dvc/remote/ssh/connection.py
+++ b/dvc/remote/ssh/connection.py
@@ -60,7 +60,13 @@ class SSHConnection:
 
     def st_mode(self, path):
         with ignore_file_not_found():
-            return self.sftp.stat(path).st_mode
+            return self.sftp.lstat(path).st_mode
+
+        return 0
+
+    def getsize(self, path):
+        with ignore_file_not_found():
+            return self.sftp.lstat(path).st_size
 
         return 0
 
@@ -267,8 +273,7 @@ class SSHConnection:
         assert len(md5) == 32
         return md5
 
-    def cp(self, src, dest):
-        self.makedirs(posixpath.dirname(dest))
+    def copy(self, src, dest):
         self.execute("cp {} {}".format(src, dest))
 
     def open_max_sftp_channels(self):
@@ -282,3 +287,23 @@ class SSHConnection:
                         raise
                     break
         return self._sftp_channels
+
+    def open(self, *args, **kwargs):
+        return self.sftp.open(*args, **kwargs)
+
+    def symlink(self, src, dest):
+        self.sftp.symlink(src, dest)
+
+    def reflink(self, src, dest):
+        if self.uname == "Linux":
+            return self.execute("cp --reflink {} {}".format(src, dest))
+
+        if self.uname == "Darwin":
+            return self.execute("cp -c {} {}".format(src, dest))
+
+        raise DvcException(
+            "'{}' is not supported as a SSH remote".format(self.uname)
+        )
+
+    def hardlink(self, src, dest):
+        self.execute("ln {} {}".format(src, dest))
diff --git a/dvc/system.py b/dvc/system.py
index b445ec01a..c7c7cf8e4 100644
--- a/dvc/system.py
+++ b/dvc/system.py
@@ -4,6 +4,7 @@ from dvc.utils.compat import str, open, fspath
 
 import os
 import errno
+import shutil
 
 
 class System(object):
@@ -11,6 +12,11 @@ class System(object):
     def is_unix():
         return os.name != "nt"
 
+    @staticmethod
+    def copy(src, dest):
+        src, dest = fspath(src), fspath(dest)
+        return shutil.copyfile(src, dest)
+
     @staticmethod
     def hardlink(source, link_name):
         import ctypes
diff --git a/setup.py b/setup.py
index 83db8b3fb..d718bc8c9 100644
--- a/setup.py
+++ b/setup.py
@@ -113,10 +113,11 @@ tests_requirements = [
     "Pygments",  # required by collective.checkdocs,
     "collective.checkdocs",
     "flake8",
+    "psutil",
     "flake8-docstrings",
     "pydocstyle<4.0",
     "jaraco.windows==3.9.2",
-    "mock-ssh-server>=0.5.0",
+    "mock-ssh-server>=0.6.0",
     "moto>=1.3.14.dev55",
     "rangehttpserver==1.2.0",
 ]
diff --git a/tests/func/test_add.py b/tests/func/test_add.py
index 4566351f3..82e474d05 100644
--- a/tests/func/test_add.py
+++ b/tests/func/test_add.py
@@ -565,12 +565,12 @@ def test_should_not_checkout_when_adding_cached_copy(repo_dir, dvc_repo):
 
     shutil.copy(repo_dir.BAR, repo_dir.FOO)
 
-    copy_spy = spy(shutil.copyfile)
+    copy_spy = spy(dvc_repo.cache.local.copy)
 
-    RemoteLOCAL.CACHE_TYPE_MAP["copy"] = copy_spy
-    dvc_repo.add(repo_dir.FOO)
+    with patch.object(dvc_repo.cache.local, "copy", copy_spy):
+        dvc_repo.add(repo_dir.FOO)
 
-    assert copy_spy.mock.call_count == 0
+        assert copy_spy.mock.call_count == 0
 
 
 @pytest.mark.parametrize(
@@ -585,13 +585,13 @@ def test_should_not_checkout_when_adding_cached_copy(repo_dir, dvc_repo):
 def test_should_relink_on_repeated_add(
     link, new_link, link_test_func, repo_dir, dvc_repo
 ):
-    dvc_repo.cache.local.cache_types = [link]
+    dvc_repo.config.set("cache", "type", link)
 
     dvc_repo.add(repo_dir.FOO)
     dvc_repo.add(repo_dir.BAR)
 
     os.remove(repo_dir.FOO)
-    RemoteLOCAL.CACHE_TYPE_MAP[link](repo_dir.BAR, repo_dir.FOO)
+    getattr(dvc_repo.cache.local, link)(repo_dir.BAR, repo_dir.FOO)
 
     dvc_repo.cache.local.cache_types = [new_link]
 
@@ -615,10 +615,11 @@ def test_should_relink_single_file_in_dir(link, link_func, dvc_repo, repo_dir):
     dvc_repo.unprotect(repo_dir.DATA_SUB)
 
     link_spy = spy(link_func)
-    RemoteLOCAL.CACHE_TYPE_MAP[link] = link_spy
-    dvc_repo.add(repo_dir.DATA_DIR)
 
-    assert link_spy.mock.call_count == 1
+    with patch.object(dvc_repo.cache.local, link, link_spy):
+        dvc_repo.add(repo_dir.DATA_DIR)
+
+        assert link_spy.mock.call_count == 1
 
 
 @pytest.mark.parametrize("link", ["hardlink", "symlink", "copy"])
diff --git a/tests/func/test_repro.py b/tests/func/test_repro.py
index f1a006ede..970c3448d 100644
--- a/tests/func/test_repro.py
+++ b/tests/func/test_repro.py
@@ -808,6 +808,10 @@ class TestReproExternalBase(TestDvc):
     def cache_scheme(self):
         return self.scheme
 
+    @property
+    def cache_type(self):
+        return "hardlink"
+
     @property
     def scheme(self):
         return None
@@ -862,7 +866,7 @@ class TestReproExternalBase(TestDvc):
         self.assertEqual(ret, 0)
         ret = main(["remote", "add", "myrepo", cache])
         self.assertEqual(ret, 0)
-        ret = main(["remote", "modify", "myrepo", "type", "hardlink"])
+        ret = main(["remote", "modify", "myrepo", "type", self.cache_type])
         self.assertEqual(ret, 0)
 
         remote_name = "myremote"
@@ -873,7 +877,7 @@ class TestReproExternalBase(TestDvc):
 
         ret = main(["remote", "add", remote_name, remote])
         self.assertEqual(ret, 0)
-        ret = main(["remote", "modify", remote_name, "type", "hardlink"])
+        ret = main(["remote", "modify", remote_name, "type", self.cache_type])
         self.assertEqual(ret, 0)
 
         self.dvc = DvcRepo(".")
@@ -993,6 +997,10 @@ class TestReproExternalHDFS(TestReproExternalBase):
     def scheme(self):
         return "hdfs"
 
+    @property
+    def cache_type(self):
+        return "copy"
+
     @property
     def bucket(self):
         return "{}@127.0.0.1".format(getpass.getuser())
diff --git a/tests/unit/remote/ssh/test_connection.py b/tests/unit/remote/ssh/test_connection.py
index bdd1a98b4..f0c12ec21 100644
--- a/tests/unit/remote/ssh/test_connection.py
+++ b/tests/unit/remote/ssh/test_connection.py
@@ -2,6 +2,14 @@ from __future__ import unicode_literals
 
 import os
 import posixpath
+import platform
+import pytest
+import filecmp
+import tempfile
+
+from dvc.system import System
+from dvc.command.version import CmdVersion
+
 
 here = os.path.abspath(os.path.dirname(__file__))
 
@@ -64,3 +72,42 @@ def test_walk(tmp_path, ssh):
             paths.add(posixpath.join(root, entry))
 
     assert paths == expected
+
+
+@pytest.mark.skipif(
+    CmdVersion.get_fs_type(tempfile.gettempdir())[0]
+    not in ["xfs", "apfs", "btrfs"],
+    reason="Reflinks only work in specified file systems",
+)
+def test_reflink(repo_dir, ssh):
+    ssh.reflink("foo", "link")
+    assert filecmp.cmp("foo", "link")
+    assert not System.is_symlink("link")
+    assert not System.is_hardlink("link")
+
+
+@pytest.mark.skipif(
+    platform.system() == "Windows",
+    reason="sftp symlink is not supported on Windows",
+)
+def test_symlink(repo_dir, ssh):
+    ssh.symlink("foo", "link")
+    assert System.is_symlink("link")
+
+
+@pytest.mark.skipif(
+    platform.system() == "Windows",
+    reason="hardlink is temporarily not supported on Windows",
+)
+def test_hardlink(repo_dir, ssh):
+    ssh.hardlink("foo", "link")
+    assert System.is_hardlink("link")
+
+
+@pytest.mark.skipif(
+    platform.system() == "Windows",
+    reason="copy is temporarily not supported on Windows",
+)
+def test_copy(repo_dir, ssh):
+    ssh.copy("foo", "link")
+    assert filecmp.cmp("foo", "link")
diff --git a/tests/unit/remote/ssh/test_ssh.py b/tests/unit/remote/ssh/test_ssh.py
index 0dee1ac3e..6e6e101aa 100644
--- a/tests/unit/remote/ssh/test_ssh.py
+++ b/tests/unit/remote/ssh/test_ssh.py
@@ -7,7 +7,10 @@ from unittest import TestCase
 from mock import patch, mock_open
 import pytest
 
+from tests.func.test_data_cloud import get_ssh_url_mocked
+
 from dvc.remote.ssh import RemoteSSH
+from dvc.system import System
 
 
 class TestRemoteSSH(TestCase):
@@ -182,3 +185,30 @@ def test_ssh_gss_auth(mock_file, mock_exists, config, expected_gss_auth):
     mock_exists.assert_called_with(RemoteSSH.ssh_config_filename())
     mock_file.assert_called_with(RemoteSSH.ssh_config_filename())
     assert remote.gss_auth == expected_gss_auth
+
+
+def test_hardlink_optimization(repo_dir, ssh_server):
+    port = ssh_server.test_creds["port"]
+    user = ssh_server.test_creds["username"]
+
+    config = {
+        "url": get_ssh_url_mocked(user, port),
+        "port": port,
+        "user": user,
+        "keyfile": ssh_server.test_creds["key_filename"],
+    }
+    remote = RemoteSSH(None, config)
+
+    from_info = remote.path_info / "empty"
+    to_info = remote.path_info / "link"
+
+    with remote.open(from_info, "wb"):
+        pass
+
+    if os.name == "nt":
+        link_path = "c:" + to_info.path.replace("/", "\\")
+    else:
+        link_path = to_info.path
+
+    remote.hardlink(from_info, to_info)
+    assert not System.is_hardlink(link_path)
diff --git a/tests/unit/remote/test_slow_link_detection.py b/tests/unit/remote/test_slow_link_detection.py
index 3b981e69e..141b4ff95 100644
--- a/tests/unit/remote/test_slow_link_detection.py
+++ b/tests/unit/remote/test_slow_link_detection.py
@@ -1,114 +1,47 @@
-import logging
-
-import colorama
 import pytest
+import mock
+
+import dvc.remote.slow_link_detection
+from dvc.remote.slow_link_detection import slow_link_guard
 from dvc.config import Config
-from dvc.remote.local.slow_link_detection import (
-    slow_link_guard,
-    SlowLinkDetectorDecorator,
-)
-from mock import Mock, patch
 
 
-@patch(
-    "dvc.remote.local.slow_link_detection.SlowLinkDetectorDecorator",
-    autospec=True,
-)
-class TestSlowLinkGuard(object):
-    @pytest.fixture(autouse=True)
-    def setUp(self):
-        self.method = Mock()
-        self.remote_local_mock = Mock()
-        self.config_mock = Mock()
-        self.remote_local_mock.repo.config.config.get.return_value = (
-            self.config_mock
-        )
-
-    def _cache_config(self, slow_link_warning, cache_type):
-        def config_side_effect(section_name, _):
-            if section_name == Config.SECTION_CACHE_SLOW_LINK_WARNING:
-                return slow_link_warning
-            elif section_name == Config.SECTION_CACHE_TYPE:
-                return cache_type
-
-        self.config_mock.get.side_effect = config_side_effect
-
-    def test_should_decorate_on_slow_link_warning_and_no_cache_type(
-        self, SlowLinkDetectorDecoratorClassMock
-    ):
-        self._cache_config(slow_link_warning=True, cache_type=None)
-
-        slow_link_guard(self.method)(self.remote_local_mock)
-
-        assert 1 == SlowLinkDetectorDecoratorClassMock.call_count
-        assert 0 == self.method.call_count
-
-    @pytest.mark.parametrize(
-        "slow_link_warning, cache_type",
-        [(True, "any_cache_type"), (False, None)],
-    )
-    def test_should_not_decorate(
-        self, SlowLinkDetectorDecoratorClassMock, slow_link_warning, cache_type
-    ):
-        self._cache_config(
-            slow_link_warning=slow_link_warning, cache_type=cache_type
-        )
-
-        slow_link_guard(self.method)(self.remote_local_mock)
-
-        assert 0 == SlowLinkDetectorDecoratorClassMock.call_count
-        assert 1 == self.method.call_count
+@pytest.fixture(autouse=True)
+def timeout_immediately(monkeypatch):
+    monkeypatch.setattr(dvc.remote.slow_link_detection, "timeout_seconds", 0.0)
 
 
-class TestSlowLinkDetectorDecorator(object):
-    @pytest.fixture(autouse=True)
-    def setUp(self):
-        self.args_mock = Mock()
-        self.method_result_mock = Mock()
+@pytest.fixture
+def make_remote():
+    def _make_remote(cache_type=None, should_warn=True):
+        remote = mock.Mock()
+        remote.repo.config.config.get.return_value = {
+            Config.SECTION_CACHE_TYPE: cache_type,
+            Config.SECTION_CACHE_SLOW_LINK_WARNING: should_warn,
+        }
+        return remote
 
-        def method_effect(arg):
-            assert self.args_mock == arg
-            return self.method_result_mock
+    return _make_remote
 
-        self.method_mock = Mock(side_effect=method_effect)
 
-    @patch.object(SlowLinkDetectorDecorator, "LINKING_TIMEOUT_SECONDS", 0.0)
-    def test_should_log_only_once(self, caplog):
+def test_show_warning_once(caplog, make_remote):
+    remote = make_remote()
+    slow_link_guard(lambda x: None)(remote)
+    slow_link_guard(lambda x: None)(remote)
 
-        with caplog.at_level(logging.WARNING, logger="dvc"):
-            test_instance = SlowLinkDetectorDecorator(self.method_mock)
-            result = test_instance(self.args_mock)
-            assert self.method_result_mock == result
+    assert caplog.records[0].message == dvc.remote.slow_link_detection.message
+    assert len(caplog.records) == 1
 
-            # slow link guard keeps creating new SlowLinkDetectorDecorators
-            # each time it is called, so we need to make sure that we are
-            # preventing every new instance of this class from displaying the
-            # message
-            test_instance2 = SlowLinkDetectorDecorator(self.method_mock)
-            result = test_instance2(self.args_mock)
-            assert self.method_result_mock == result
 
-        msg = (
-            "You can cut execution time considerably by using a different"
-            " `cache.type` configuration.\nSee "
-            "{blue}https://dvc.org/doc/commands-reference/config#cache{"
-            "reset} for more information.\n"
-            "To disable this message, run:\n"
-            "'dvc config cache.slow_link_warning false'".format(
-                blue=colorama.Fore.BLUE, reset=colorama.Fore.RESET
-            )
-        )
-        assert 1 == len(caplog.messages)
-        assert msg == caplog.messages[0]
+def test_dont_warn_when_cache_type_is_set(caplog, make_remote):
+    remote = make_remote(cache_type="copy")
+    slow_link_guard(lambda x: None)(remote)
 
-    @patch.object(
-        SlowLinkDetectorDecorator, "LINKING_TIMEOUT_SECONDS", 99999.0
-    )
-    def test_should_not_log(self, caplog):
-        test_instance = SlowLinkDetectorDecorator(self.method_mock)
+    assert len(caplog.records) == 0
 
-        with caplog.at_level(logging.WARNING, logger="dvc"):
-            result = test_instance(self.args_mock)
 
-        assert self.method_result_mock == result
-        assert 0 == len(caplog.messages)
+def test_dont_warn_when_warning_is_disabled(caplog, make_remote):
+    remote = make_remote(should_warn=False)
+    slow_link_guard(lambda x: None)(remote)
+
+    assert len(caplog.records) == 0