diff --git a/dvc/output/base.py b/dvc/output/base.py
index abbcaf402..c841200b4 100644
--- a/dvc/output/base.py
+++ b/dvc/output/base.py
@@ -137,11 +137,10 @@ class OutputBase(object):
         return {}
 
     def save(self):
-        if not self.use_cache:
-            self.info = self.remote.save_info(self.path_info)
-        else:
-            self.info = getattr(self.repo.cache, self.scheme).save(
-                self.path_info
+        self.info = self.remote.save_info(self.path_info)
+        if self.use_cache:
+            getattr(self.repo.cache, self.scheme).save(
+                self.path_info, self.info
             )
 
     def dumpd(self):
diff --git a/dvc/output/local.py b/dvc/output/local.py
index 1324e1377..ea47f9285 100644
--- a/dvc/output/local.py
+++ b/dvc/output/local.py
@@ -117,4 +117,5 @@ class OutputLOCAL(OutputBase):
             if self.use_cache:
                 self.repo.scm.ignore(self.path)
 
-        self.info = self.repo.cache.local.save(self.path_info)
+        self.info = self.remote.save_info(self.path_info)
+        self.project.cache.local.save(self.path_info, self.info)
diff --git a/dvc/remote/base.py b/dvc/remote/base.py
index b08b83986..5793751e1 100644
--- a/dvc/remote/base.py
+++ b/dvc/remote/base.py
@@ -172,7 +172,7 @@ class RemoteBase(object):
         logger.debug("'{}' hasn't changed.".format(path_info))
         return False
 
-    def save(self, path_info):
+    def save(self, path_info, checksum_info):
         raise RemoteActionNotImplemented("save", self.scheme)
 
     def download(
diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py
index 5d36500cb..287bb66ae 100644
--- a/dvc/remote/hdfs.py
+++ b/dvc/remote/hdfs.py
@@ -88,20 +88,18 @@ class RemoteHDFS(RemoteBase):
 
         return {self.PARAM_CHECKSUM: self.checksum(path_info)}
 
-    def save(self, path_info):
+    def save(self, path_info, checksum_info):
         if path_info["scheme"] != "hdfs":
             raise NotImplementedError
 
         assert path_info.get("path")
 
-        checksum = self.checksum(path_info)
+        checksum = checksum_info[self.PARAM_CHECKSUM]
         dest = path_info.copy()
         dest["path"] = self.checksum_to_path(checksum)
 
         self.copy(path_info, dest)
 
-        return {self.PARAM_CHECKSUM: checksum}
-
     def remove(self, path_info):
         if path_info["scheme"] != "hdfs":
             raise NotImplementedError
diff --git a/dvc/remote/local.py b/dvc/remote/local.py
index 6a5414189..7526bad88 100644
--- a/dvc/remote/local.py
+++ b/dvc/remote/local.py
@@ -370,9 +370,7 @@ class RemoteLOCAL(RemoteBase):
         move(inp, tmp)
         move(tmp, outp)
 
-    def _save_file(self, path_info):
-        path = path_info["path"]
-        md5 = self.state.update(path)
+    def _save_file(self, path, md5):
         assert md5 is not None
 
         cache = self.get(md5)
@@ -393,9 +391,8 @@ class RemoteLOCAL(RemoteBase):
 
         return {self.PARAM_CHECKSUM: md5}
 
-    def _save_dir(self, path_info):
-        path = path_info["path"]
-        md5, dir_info = self.state.update_info(path)
+    def _save_dir(self, path, md5):
+        dir_info = self.load_dir_cache(md5)
         dir_relpath = os.path.relpath(path)
         dir_size = len(dir_info)
         bar = dir_size > LARGE_DIR_SIZE
@@ -429,9 +426,7 @@ class RemoteLOCAL(RemoteBase):
         if bar:
             progress.finish_target(dir_relpath)
 
-        return {self.PARAM_CHECKSUM: md5}
-
-    def save(self, path_info):
+    def save(self, path_info, checksum_info):
         if path_info["scheme"] != "local":
             raise NotImplementedError
 
@@ -442,10 +437,11 @@ class RemoteLOCAL(RemoteBase):
             msg.format(os.path.relpath(path), os.path.relpath(self.cache_dir))
         )
 
+        md5 = checksum_info[self.PARAM_CHECKSUM]
         if os.path.isdir(path):
-            return self._save_dir(path_info)
+            self._save_dir(path, md5)
         else:
-            return self._save_file(path_info)
+            self._save_file(path, md5)
 
     def save_info(self, path_info):
         if path_info["scheme"] != "local":
diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py
index d7c3d4f19..ad8349fa6 100644
--- a/dvc/remote/s3.py
+++ b/dvc/remote/s3.py
@@ -109,18 +109,16 @@ class RemoteS3(RemoteBase):
         source = {"Bucket": from_info["bucket"], "Key": from_info["path"]}
         self.s3.copy(source, to_info["bucket"], to_info["path"])
 
-    def save(self, path_info):
+    def save(self, path_info, checksum_info):
         if path_info["scheme"] != "s3":
             raise NotImplementedError
 
-        etag = self.get_etag(path_info["bucket"], path_info["path"])
+        etag = checksum_info[self.PARAM_CHECKSUM]
         path = self.checksum_to_path(etag)
         to_info = {"scheme": "s3", "bucket": self.bucket, "path": path}
 
         self.copy(path_info, to_info)
 
-        return {self.PARAM_CHECKSUM: etag}
-
     def remove(self, path_info):
         if path_info["scheme"] != "s3":
             raise NotImplementedError
diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py
index 1a2f6ee5e..2b175e45c 100644
--- a/dvc/remote/ssh.py
+++ b/dvc/remote/ssh.py
@@ -244,18 +244,16 @@ class RemoteSSH(RemoteBase):
 
         return {self.PARAM_CHECKSUM: self.md5(path_info)}
 
-    def save(self, path_info):
+    def save(self, path_info, checksum_info):
         if path_info["scheme"] != "ssh":
             raise NotImplementedError
 
-        md5 = self.md5(path_info)
+        md5 = checksum_info[self.PARAM_CHECKSUM]
         dest = path_info.copy()
         dest["path"] = self.checksum_to_path(md5)
 
         self.copy(path_info, dest)
 
-        return {self.PARAM_CHECKSUM: md5}
-
     def remove(self, path_info):
         if path_info["scheme"] != "ssh":
             raise NotImplementedError