diff --git a/dvc/cache.py b/dvc/cache.py
index 7f0c9fdda..0b8c1ac0f 100644
--- a/dvc/cache.py
+++ b/dvc/cache.py
@@ -1,48 +1,208 @@
 import os
+import json
+import shutil
 
+from dvc.state import State
 from dvc.system import System
+from dvc.logger import Logger
+from dvc.utils import move, remove
+from dvc.lock import Lock
+from dvc.exceptions import DvcException
 
 
 class Cache(object):
     CACHE_DIR = 'cache'
+    CACHE_DIR_LOCK = 'cache.lock'
+    CACHE_TYPES = ['reflink', 'hardlink', 'symlink', 'copy']
+    CACHE_TYPE_MAP = {
+        'copy': shutil.copyfile,
+        'symlink': System.symlink,
+        'hardlink': System.hardlink,
+        'reflink': System.reflink,
+    }
+
+    def __init__(self, dvc_dir, cache_dir=None, cache_type=None):
+        self.cache_type = cache_type
+
+        cache_dir = cache_dir if cache_dir else self.CACHE_DIR
+        if os.path.isabs(cache_dir):
+            self.cache_dir = cache_dir
+        else:
+            self.cache_dir = os.path.abspath(os.path.realpath(os.path.join(dvc_dir, cache_dir)))
 
-    def __init__(self, dvc_dir):
-        self.cache_dir = os.path.abspath(os.path.realpath(os.path.join(dvc_dir, self.CACHE_DIR)))
         if not os.path.exists(self.cache_dir):
             os.mkdir(self.cache_dir)
 
+        self.state = State(self.cache_dir)
+        self.lock = Lock(self.cache_dir, name=self.CACHE_DIR_LOCK)
+
     @staticmethod
-    def init(dvc_dir):
-        cache_dir = os.path.join(dvc_dir, Cache.CACHE_DIR)
-        os.mkdir(cache_dir)
-        return Cache(dvc_dir)
+    def init(dvc_dir, cache_dir=None):
+        return Cache(dvc_dir, cache_dir=None)
 
     def all(self):
-        clist = []
-        for entry in os.listdir(self.cache_dir):
-            subdir = os.path.join(self.cache_dir, entry)
-            if not os.path.isdir(subdir):
-                continue
+        with self.lock:
+            clist = []
+            for entry in os.listdir(self.cache_dir):
+                subdir = os.path.join(self.cache_dir, entry)
+                if not os.path.isdir(subdir):
+                    continue
 
-            for cache in os.listdir(subdir):
-                path = os.path.join(subdir, cache)
-                clist.append(path)
+                for cache in os.listdir(subdir):
+                    path = os.path.join(subdir, cache)
+                    clist.append(path)
 
-        return clist
+            return clist
 
     def get(self, md5):
+        if not md5:
+            return None
+
         return os.path.join(self.cache_dir, md5[0:2], md5[2:])
 
     def path_to_md5(self, path):
         relpath = os.path.relpath(path, self.cache_dir)
         return os.path.dirname(relpath) + os.path.basename(relpath)
 
-    def find_cache(self, files):
-        file_set = set(files)
-        cached = {}
-        for cache_file in self.all():
-            cached_files = list(filter(lambda f: System.samefile(cache_file, f), file_set))
-            cached.update(dict((f, os.path.basename(os.path.dirname(cache_file)) + os.path.basename(cache_file)) for f in cached_files))
-            file_set = file_set - set(cached_files)
-        return cached
+    def _changed(self, md5):
+        cache = self.get(md5)
+        if self.state.changed(cache, md5=md5):
+            if os.path.exists(cache):
+                Logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache)))
+                remove(cache)
+            return True
 
+        return False
+
+    def changed(self, md5):
+        with self.lock:
+            return self._changed(md5)
+
+    def link(self, src, link):
+        dname = os.path.dirname(link)
+        if not os.path.exists(dname):
+            os.makedirs(dname)
+
+        if self.cache_type != None:
+            types = [self.cache_type]
+        else:
+            types = self.CACHE_TYPES
+
+        for typ in types:
+            try:
+                self.CACHE_TYPE_MAP[typ](src, link)
+                return
+            except Exception as exc:
+                msg = 'Cache type \'{}\' is not supported'.format(typ)
+                Logger.debug(msg)
+                if typ == types[-1]:
+                    raise DvcException(msg, cause=exc)
+
+    @staticmethod
+    def load_dir_cache(path):
+        if os.path.isabs(path):
+            relpath = os.path.relpath(path)
+        else:
+            relpath = path
+
+        try:
+            with open(path, 'r') as fd:
+                d = json.load(fd)
+        except Exception as exc:
+            msg = u'Failed to load dir cache \'{}\''
+            Logger.error(msg.format(relpath), exc)
+            return []
+
+        if not isinstance(d, list):
+            msg = u'Dir cache file format error \'{}\': skipping the file'
+            Logger.error(msg.format(relpath))
+            return []
+
+        return d
+
+    @staticmethod
+    def get_dir_cache(path):
+        res = {}
+        d = Cache.load_dir_cache(path)
+
+        for entry in d:
+            res[entry[State.PARAM_RELPATH]] = entry[State.PARAM_MD5]
+
+        return res
+
+    def dir_cache(self, cache):
+        res = {}
+        dir_cache = self.get_dir_cache(cache)
+
+        for relpath, md5 in dir_cache.items():
+            res[relpath] = self.get(md5)
+
+        return res
+
+    @staticmethod
+    def is_dir_cache(cache):
+        return cache.endswith(State.MD5_DIR_SUFFIX)
+
+    def _checkout(self, path, md5):
+        cache = self.get(md5)
+
+        if not cache or not os.path.exists(cache) or self._changed(md5):
+            if cache:
+                Logger.warn(u'\'{}({})\': cache file not found'.format(os.path.relpath(cache),
+                                                                       os.path.relpath(path)))
+            remove(path)
+            return
+
+        if os.path.exists(path):
+            msg = u'Data \'{}\' exists. Removing before checkout'
+            Logger.debug(msg.format(os.path.relpath(path)))
+            remove(path)
+
+        msg = u'Checking out \'{}\' with cache \'{}\''
+        Logger.debug(msg.format(os.path.relpath(path), os.path.relpath(cache)))
+
+        if not self.is_dir_cache(cache):
+            self.link(cache, path)
+            return
+
+        dir_cache = self.dir_cache(cache)
+        for relpath, c in dir_cache.items():
+            p = os.path.join(path, relpath)
+            self.link(c, p)
+
+    def checkout(self, path, md5):
+        with self.lock:
+            return self._checkout(path, md5)
+
+    def _save_file(self, path):
+        md5 = self.state.update(path)
+        cache = self.get(md5)
+        if self._changed(md5):
+            move(path, cache)
+            self.state.update(cache)
+        self._checkout(path, md5)
+
+    def _save_dir(self, path):
+        md5 = self.state.update(path)
+        cache = self.get(md5)
+        dname = os.path.dirname(cache)
+        dir_info = self.state.collect_dir(path)
+
+        for entry in dir_info:
+            relpath = entry[State.PARAM_RELPATH]
+            p = os.path.join(path, relpath)
+
+            self._save_file(p)
+
+        if not os.path.isdir(dname):
+            os.makedirs(dname)
+
+        with open(cache, 'w+') as fd:
+            json.dump(dir_info, fd, sort_keys=True)
+
+    def save(self, path):
+        with self.lock:
+            if os.path.isdir(path):
+                self._save_dir(path)
+            else:
+                self._save_file(path)
diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py
index b23e19728..54dec99dd 100644
--- a/dvc/cloud/base.py
+++ b/dvc/cloud/base.py
@@ -5,7 +5,7 @@ import tempfile
 from dvc.logger import Logger
 from dvc.exceptions import DvcException
 from dvc.config import Config, ConfigError
-from dvc.stage import Output
+from dvc.cache import Cache
 
 
 STATUS_UNKNOWN = 0
@@ -31,9 +31,8 @@ class DataCloudError(DvcException):
 
 
 class CloudSettings(object):
-    def __init__(self, cache=None, state=None, global_storage_path=None, cloud_config=None):
+    def __init__(self, cache=None, global_storage_path=None, cloud_config=None):
         self.cache = cache
-        self.state = state
         self.cloud_config = cloud_config
         self.global_storage_path = global_storage_path
 
@@ -131,7 +130,7 @@ class DataCloudBase(object):
         path, local = arg
         ret = [path]
 
-        if not Output.is_dir_cache(path):
+        if not Cache.is_dir_cache(path):
             return ret
 
         if local:
@@ -147,7 +146,7 @@ class DataCloudBase(object):
             self._pull_key(key, tmp, no_progress_bar=True)
             dir_path = tmp
 
-        for relpath, md5 in Output.get_dir_cache(dir_path).items():
+        for relpath, md5 in Cache.get_dir_cache(dir_path).items():
             cache = self._cloud_settings.cache.get(md5)
             ret.append(cache)
 
@@ -155,7 +154,7 @@ class DataCloudBase(object):
 
     def _cmp_checksum(self, blob, fname):
         md5 = self._cloud_settings.cache.path_to_md5(fname)
-        if self._cloud_settings.state.changed(fname, md5=md5):
+        if self._cloud_settings.cache.state.changed(fname, md5=md5):
             return False
 
         return True
diff --git a/dvc/cloud/data_cloud.py b/dvc/cloud/data_cloud.py
index c88679934..06b23be65 100644
--- a/dvc/cloud/data_cloud.py
+++ b/dvc/cloud/data_cloud.py
@@ -23,10 +23,9 @@ class DataCloud(object):
         'local' : DataCloudLOCAL,
     }
 
-    def __init__(self, cache=None, config=None, state=None):
+    def __init__(self, cache=None, config=None):
         self._cache = cache
         self._config = config
-        self._state = state
 
         remote = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_REMOTE, '')
         if remote == '':
@@ -82,7 +81,6 @@ class DataCloud(object):
             Logger.warn('Using obsoleted config format. Consider updating.')
 
         cloud_settings = CloudSettings(cache=self._cache,
-                                       state=self._state,
                                        global_storage_path=global_storage_path,
                                        cloud_config=cloud_config)
 
diff --git a/dvc/config.py b/dvc/config.py
index b1832c0de..a98454d9a 100644
--- a/dvc/config.py
+++ b/dvc/config.py
@@ -5,6 +5,7 @@ import os
 import schema
 import configobj
 
+from dvc.cache import Cache
 from dvc.exceptions import DvcException
 
 
@@ -28,6 +29,15 @@ class Config(object):
     SECTION_CORE_LOGLEVEL_SCHEMA = schema.And(schema.Use(str.lower), lambda l: l in ('info', 'debug', 'warning', 'error'))
     SECTION_CORE_REMOTE = 'remote'
 
+    SECTION_CACHE = 'cache'
+    SECTION_CACHE_DIR = 'dir'
+    SECTION_CACHE_TYPE = 'type'
+    SECTION_CACHE_TYPE_SCHEMA = schema.And(schema.Use(str.lower), lambda t: t in Cache.CACHE_TYPE_MAP.keys())
+    SECTION_CACHE_SCHEMA = {
+        schema.Optional(SECTION_CACHE_DIR, default=Cache.CACHE_DIR): str,
+        schema.Optional(SECTION_CACHE_TYPE, default=None): SECTION_CACHE_TYPE_SCHEMA,
+    }
+
     # backward compatibility
     SECTION_CORE_CLOUD = 'cloud'
     SECTION_CORE_CLOUD_SCHEMA = schema.And(schema.Use(str.lower), lambda c: c in ('aws', 'gcp', 'local', ''))
@@ -86,6 +96,7 @@ class Config(object):
     SCHEMA = {
         schema.Optional(SECTION_CORE, default={}): SECTION_CORE_SCHEMA,
         schema.Optional(schema.Regex(SECTION_REMOTE_REGEX)): SECTION_REMOTE_SCHEMA,
+        schema.Optional(SECTION_CACHE, default={}): SECTION_CACHE_SCHEMA,
 
         # backward compatibility
         schema.Optional(SECTION_AWS, default={}): SECTION_AWS_SCHEMA,
diff --git a/dvc/lock.py b/dvc/lock.py
index 0fbff6160..68a659562 100644
--- a/dvc/lock.py
+++ b/dvc/lock.py
@@ -13,8 +13,8 @@ class Lock(object):
     LOCK_FILE = 'lock'
     TIMEOUT = 5
 
-    def __init__(self, dvc_dir):
-        self.lock_file = os.path.join(dvc_dir, self.LOCK_FILE)
+    def __init__(self, dvc_dir, name=LOCK_FILE):
+        self.lock_file = os.path.join(dvc_dir, name)
         self._lock = None
 
     @staticmethod
diff --git a/dvc/output.py b/dvc/output.py
index c2d032a98..d32a334f1 100644
--- a/dvc/output.py
+++ b/dvc/output.py
@@ -1,15 +1,11 @@
 import os
-import stat
-import json
-import shutil
 import schema
 import posixpath
 import ntpath
 
-from dvc.system import System
-from dvc.utils import file_md5
 from dvc.exceptions import DvcException
 from dvc.logger import Logger
+from dvc.utils import remove
 
 
 class OutputError(DvcException):
@@ -47,7 +43,6 @@ class CmdOutputAlreadyTrackedError(CmdOutputError):
 
 
 class Dependency(object):
-    PARAM_RELPATH = 'relpath'
     PARAM_PATH = 'path'
     PARAM_MD5 = 'md5'
     MD5_DIR_SUFFIX = '.dir'
@@ -59,7 +54,7 @@ class Dependency(object):
 
     def __init__(self, project, path, md5=None):
         self.project = project
-        self.path = os.path.abspath(os.path.realpath(path))
+        self.path = os.path.abspath(os.path.normpath(path))
 
         if not self.path.startswith(self.project.root_dir):
             raise CmdOutputOutsideOfRepoError(self.rel_path)
@@ -96,10 +91,6 @@ class Dependency(object):
             return {self.rel_path: 'changed'}
         return {}
 
-    @staticmethod
-    def is_dir_cache(cache):
-        return cache.endswith(Output.MD5_DIR_SUFFIX)
-
     def save(self):
         if not os.path.exists(self.path):
             raise CmdOutputDoesNotExistError(self.rel_path)
@@ -160,18 +151,11 @@ class Output(Dependency):
 
     @property
     def cache(self):
-        if not self.md5:
+        if not self.use_cache:
             return None
 
         return self.project.cache.get(self.md5)
 
-    @property
-    def rel_cache(self):
-        if not self.cache:
-            return None
-
-        return os.path.relpath(self.cache)
-
     def dumpd(self, cwd):
         ret = super(Output, self).dumpd(cwd)
         ret[Output.PARAM_CACHE] = self.use_cache
@@ -193,151 +177,28 @@ class Output(Dependency):
     def loads_from(cls, project, s_list, use_cache=False, cwd=os.curdir):
         return [cls.loads(project, x, use_cache=use_cache, cwd=cwd) for x in s_list]
 
-    def _changed_cache(self, cache):
-        md5 = self.project.cache.path_to_md5(cache)
-        if self.project.state.changed(cache, md5=md5):
-            self.project.logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache)))
-            os.unlink(cache)
+    def changed(self):
+        if super(Output, self).changed():
+            return True
+
+        if self.use_cache and self.project.cache.changed(self.md5):
             return True
 
         return False
 
-    def _changed_file(self, path, cache):
-        if os.path.isfile(path) and \
-           os.path.isfile(cache) and \
-           System.samefile(path, cache) and \
-           not self._changed_cache(cache):
-            return False
-
-        return True
-
-    def _changed_dir(self):
-        if os.path.isdir(self.path) and \
-           os.path.isfile(self.cache) and \
-           not self._changed_cache(self.cache):
-            return False
-
-        return True
-
-    def changed(self):
-        if not self.use_cache:
-            ret = super(Output, self).changed()
-        elif not self.cache:
-            ret = True
-        elif self.is_dir_cache(self.cache):
-            ret = self._changed_dir()
-        else:
-            ret = self._changed_file(self.path, self.cache)
-
-        msg = u'Data file or dir \'{}\' with cache \'{}\' {}'
-        self.project.logger.debug(msg.format(self.rel_path, self.rel_cache, self._changed_msg(ret)))
-
-        return ret
-
-    def hardlink(self, src, link):
-        rel_src = os.path.relpath(src)
-        rel_link = os.path.relpath(link)
-        self.project.logger.debug(u'creating hardlink {} -> {}'.format(rel_src, rel_link))
-
-        dname = os.path.dirname(link)
-        if not os.path.exists(dname):
-            os.makedirs(dname)
-
-        System.hardlink(src, link)
-
-    @staticmethod
-    def load_dir_cache(path):
-        if os.path.isabs(path):
-            relpath = os.path.relpath(path)
-        else:
-            relpath = path
-
-        try:
-            with open(path, 'r') as fd:
-                d = json.load(fd)
-        except Exception as exc:
-            msg = u'Failed to load dir cache \'{}\''
-            Logger.error(msg.format(relpath), exc)
-            return []
-
-        if not isinstance(d, list):
-            msg = u'Dir cache file format error \'{}\': skipping the file'
-            Logger.error(msg.format(relpath))
-            return []
-
-        return d
-
-    @staticmethod
-    def get_dir_cache(path):
-        res = {}
-        d = Output.load_dir_cache(path)
-
-        for entry in d:
-            res[entry[Output.PARAM_RELPATH]] = entry[Output.PARAM_MD5]
-
-        return res
-
-    def dir_cache(self):
-        res = {}
-        dir_cache = self.get_dir_cache(self.cache)
-
-        for relpath, md5 in dir_cache.items():
-            res[relpath] = self.project.cache.get(md5)
-
-        return res
-
     def checkout(self):
         if not self.use_cache:
             return
 
         msg = u'Checking out \'{}\' with cache \'{}\''
-        self.project.logger.debug(msg.format(self.rel_path, self.rel_cache))
+        self.project.logger.debug(msg.format(self.rel_path, self.md5))
 
         if not self.changed():
             msg = u'Data file \'{}\' with cache \'{}\' didn\'t change, skipping checkout.'
-            self.project.logger.debug(msg.format(self.rel_path, self.rel_cache))
+            self.project.logger.debug(msg.format(self.rel_path, self.md5))
             return
 
-        if not self.cache or not os.path.exists(self.cache):
-            if self.cache:
-                self.project.logger.warn(u'\'{}({})\': cache file not found'.format(self.rel_cache, self.rel_path))
-            self.remove()
-            return
-
-        if os.path.exists(self.path):
-            msg = u'Data file \'{}\' exists. Removing before checkout'
-            self.project.logger.debug(msg.format(self.rel_path))
-            self.remove()
-
-        if not self.is_dir_cache(self.cache):
-            self.hardlink(self.cache, self.path)
-            return
-
-        for relpath, cache in self.dir_cache().items():
-            path = os.path.join(self.path, relpath)
-            self.hardlink(cache, path)
-
-    def _save_dir(self):
-        dname = os.path.dirname(self.cache)
-        dir_info = self.project.state.collect_dir(self.path)
-
-        for entry in dir_info:
-            md5 = entry[self.PARAM_MD5]
-            relpath = entry[self.PARAM_RELPATH]
-            path = os.path.join(self.path, relpath)
-            cache = self.project.cache.get(md5)
-
-            if os.path.exists(cache):
-                self._remove(path)
-                self.hardlink(cache, path)
-            else:
-                self.hardlink(path, cache)
-
-        if not os.path.isdir(dname):
-            os.makedirs(dname)
-
-        with open(self.cache, 'w+') as fd:
-            json.dump(dir_info, fd, sort_keys=True)
+        self.project.cache.checkout(self.path, self.md5)
 
     def save(self):
         super(Output, self).save()
@@ -345,7 +206,7 @@ class Output(Dependency):
         if not self.use_cache:
             return
 
-        self.project.logger.debug(u'Saving \'{}\' to \'{}\''.format(self.rel_path, self.rel_cache))
+        self.project.logger.debug(u'Saving \'{}\' to \'{}\''.format(self.rel_path, self.md5))
 
         if self.project.scm.is_tracked(self.path):
             raise CmdOutputAlreadyTrackedError(self.rel_path)
@@ -353,29 +214,7 @@ class Output(Dependency):
         if not self.changed():
              return
 
-        if os.path.exists(self.cache):
-            # This means that we already have cache for this data.
-            # We remove data and link it to existing cache to save
-            # some space.
-            msg = u'Cache \'{}\' already exists, performing checkout for \'{}\''
-            self.project.logger.debug(msg.format(self.rel_cache, self.rel_path))
-            self.checkout()
-            return
-
-        if os.path.isdir(self.path):
-            self._save_dir()
-        else:
-            self.hardlink(self.path, self.cache)
-
-    def _remove(self, path):
-        if not os.path.exists(path):
-            return
-
-        self.project.logger.debug(u'Removing \'{}\''.format(os.path.relpath(path)))
-        if os.path.isfile(path):
-            os.unlink(path)
-        else:
-            shutil.rmtree(path)
+        self.project.cache.save(self.path)
 
     def remove(self):
-        self._remove(self.path)
+        remove(self.path)
diff --git a/dvc/project.py b/dvc/project.py
index de3acae58..2c8e521c0 100644
--- a/dvc/project.py
+++ b/dvc/project.py
@@ -9,7 +9,7 @@ import dvc.cloud.base as cloud
 
 from dvc.logger import Logger
 from dvc.exceptions import DvcException
-from dvc.stage import Stage, Output
+from dvc.stage import Stage
 from dvc.config import Config
 from dvc.state import State
 from dvc.lock import Lock
@@ -43,13 +43,14 @@ class Project(object):
         self.root_dir = os.path.abspath(os.path.realpath(root_dir))
         self.dvc_dir = os.path.join(self.root_dir, self.DVC_DIR)
 
+        self.config = Config(self.dvc_dir)
         self.scm = SCM(self.root_dir)
         self.lock = Lock(self.dvc_dir)
-        self.cache = Cache(self.dvc_dir)
-        self.state = State(self.root_dir, self.dvc_dir)
-        self.config = Config(self.dvc_dir)
+        self.cache = Cache(self.dvc_dir, cache_dir=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_DIR, None),
+                                         cache_type=self.config._config[Config.SECTION_CACHE].get(Config.SECTION_CACHE_TYPE, None))
+        self.state = State(self.dvc_dir)
         self.logger = Logger(self.config._config[Config.SECTION_CORE].get(Config.SECTION_CORE_LOGLEVEL, None))
-        self.cloud = DataCloud(cache=self.cache, state=self.state, config=self.config._config)
+        self.cloud = DataCloud(cache=self.cache, config=self.config._config)
 
     @staticmethod
     def init(root_dir=os.curdir, no_scm=False):
@@ -77,7 +78,7 @@ class Project(object):
 
         config = Config.init(dvc_dir)
         cache = Cache.init(dvc_dir)
-        state = State.init(root_dir, dvc_dir)
+        state = State.init(dvc_dir)
         lock = Lock(dvc_dir)
 
         scm.ignore_list([cache.cache_dir,
@@ -211,8 +212,8 @@ class Project(object):
                 if not out.use_cache or not out.cache:
                     continue
                 cache_set |= set([out.cache])
-                if out.is_dir_cache(out.cache) and os.path.isfile(out.cache):
-                    dir_cache = out.dir_cache()
+                if self.cache.is_dir_cache(out.cache) and os.path.isfile(out.cache):
+                    dir_cache = self.cache.dir_cache(out.cache)
                     cache_set |= set(dir_cache.values())
 
         return list(cache_set)
diff --git a/dvc/state.py b/dvc/state.py
index 4d32bc1d6..13a29c99c 100644
--- a/dvc/state.py
+++ b/dvc/state.py
@@ -3,8 +3,8 @@ import json
 import nanotime
 import threading
 
+from dvc.lock import Lock
 from dvc.system import System
-from dvc.output import Output
 from dvc.utils import file_md5, dict_md5
 from dvc.exceptions import DvcException
 
@@ -40,17 +40,22 @@ class StateDuplicateError(DvcException):
 
 class State(object):
     STATE_FILE = 'state'
+    STATE_LOCK_FILE = 'state.lock'
 
-    def __init__(self, root_dir, dvc_dir):
-        self.root_dir = root_dir
+    PARAM_RELPATH = 'relpath'
+    PARAM_MD5 = 'md5'
+    MD5_DIR_SUFFIX = '.dir'
+
+    def __init__(self, dvc_dir):
         self.dvc_dir = dvc_dir
         self.state_file = os.path.join(dvc_dir, self.STATE_FILE)
         self._db = self.load()
         self._lock = threading.Lock()
+        self._lock_file = Lock(dvc_dir, self.STATE_LOCK_FILE)
 
     @staticmethod
-    def init(root_dir, dvc_dir):
-        return State(root_dir, dvc_dir)
+    def init(dvc_dir):
+        return State(dvc_dir)
 
     def collect_dir(self, dname):
         dir_info = []
@@ -61,7 +66,7 @@ class State(object):
                 relpath = os.path.relpath(path, dname)
 
                 md5 = self.update(path, dump=False)
-                dir_info.append({Output.PARAM_RELPATH: relpath, Output.PARAM_MD5: md5})
+                dir_info.append({self.PARAM_RELPATH: relpath, self.PARAM_MD5: md5})
 
         self.dump()
 
@@ -70,7 +75,7 @@ class State(object):
     def compute_md5(self, path):
         if os.path.isdir(path):
             dir_info = self.collect_dir(path)
-            return dict_md5(dir_info) + Output.MD5_DIR_SUFFIX
+            return dict_md5(dir_info) + self.MD5_DIR_SUFFIX
         else:
             return file_md5(path)[0]
 
@@ -117,16 +122,19 @@ class State(object):
         d = state.dumpd()
 
         with self._lock:
-            self._db[inode] = d
+            with self._lock_file:
+                self._db[inode] = d
 
-            if dump:
-                self.dump()
+                if dump:
+                    self.dump()
 
         return md5
 
     def _get(self, inode, mtime):
         with self._lock:
-            d = self._db.get(inode, None)
+            with self._lock_file:
+                d = self._db.get(inode, None)
+
         if not d:
             return None
 
diff --git a/dvc/system.py b/dvc/system.py
index 7a00dfa31..622e287b6 100644
--- a/dvc/system.py
+++ b/dvc/system.py
@@ -1,8 +1,6 @@
-import ctypes
 import os
-import re
-import subprocess
-from builtins import str
+import ctypes
+import reflink
 
 if os.name == 'nt':
     import ntfsutils.hardlink as winlink
@@ -10,9 +8,6 @@ if os.name == 'nt':
 
 
 class System(object):
-    SYMLINK_OUTPUT = '<SYMLINK>'
-    LONG_PATH_BUFFER_SIZE = 1024
-
     @staticmethod
     def is_unix():
         return os.name != 'nt'
@@ -24,6 +19,26 @@ class System(object):
 
         return winlink.create(source, link_name)
 
+    @staticmethod
+    def symlink(source, link_name):
+        if System.is_unix():
+            return os.symlink(source, link_name)
+
+        flags = 0
+        if source is not None and os.path.isdir(source):
+            flags = 1
+
+        func = ctypes.windll.kernel32.CreateSymbolicLinkW
+        func.argtypes = (ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.c_uint32)
+        func.restype = ctypes.c_ubyte
+
+        if func(link_name, source, flags) == 0:
+            raise ctypes.WinError()
+
+    @staticmethod
+    def reflink(source, link_name):
+        return reflink.reflink(source, link_name)
+
     @staticmethod
     def inode(path):
         if System.is_unix():
@@ -32,13 +47,3 @@ class System(object):
         # getdirinfo from ntfsutils works on both files and dirs
         info = getdirinfo(path)
         return hash((info.dwVolumeSerialNumber, info.nFileIndexHigh, info.nFileIndexLow))
-
-    @staticmethod
-    def samefile(path1, path2):
-        if not os.path.exists(path1) or not os.path.exists(path2):
-            return False
-
-        if System.is_unix():
-            return os.path.samefile(path1, path2)
-
-        return winlink.samefile(path1, path2)
diff --git a/dvc/utils.py b/dvc/utils.py
index 7995c856b..86f8fdd4e 100644
--- a/dvc/utils.py
+++ b/dvc/utils.py
@@ -131,3 +131,22 @@ def map_progress(func, targets, n_threads):
         raise
 
     return list(zip(targets, ret))
+
+
+def move(src, dst):
+    dname = os.path.dirname(dst)
+    if not os.path.exists(dname):
+        os.makedirs(dname)
+
+    shutil.move(src, dst)
+
+
+def remove(path):
+    if not os.path.exists(path):
+        return
+
+    Logger.debug(u'Removing \'{}\''.format(os.path.relpath(path)))
+    if os.path.isfile(path):
+        os.unlink(path)
+    else:
+        shutil.rmtree(path)
diff --git a/requirements.txt b/requirements.txt
index 3e8fb4911..2dbfd783e 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -19,3 +19,4 @@ schema>=0.6.7
 paramiko>=2.4.1
 binaryornot==0.4.4
 jsonpath-rw==1.4.0
+reflink==0.2.0
diff --git a/scripts/build_windows.cmd b/scripts/build_windows.cmd
index 1e035d530..8b1502850 100644
--- a/scripts/build_windows.cmd
+++ b/scripts/build_windows.cmd
@@ -21,6 +21,9 @@ call pip install pyinstaller || goto :error
 echo ====== Building dvc binary... ======
 call pyinstaller --onefile --additional-hooks-dir scripts\hooks dvc/__main__.py --name dvc --specpath build
 
+echo ====== Copying additional files... ======
+copy scripts\innosetup\addSymLinkPermissions.ps1 dist\ || goto :error
+
 echo ====== Building dvc installer... ======
 set PYTHONPATH=%cd%
 call python scripts\innosetup\config_gen.py || goto :error
diff --git a/scripts/innosetup/addSymLinkPermissions.ps1 b/scripts/innosetup/addSymLinkPermissions.ps1
new file mode 100644
index 000000000..ea44eba6b
--- /dev/null
+++ b/scripts/innosetup/addSymLinkPermissions.ps1
@@ -0,0 +1,76 @@
+param (
+	[string]$mytype = 'system'
+)
+
+function addSymLinkPermissions($accountToAdd){
+    Write-Host "Checking SymLink permissions.."
+    $sidstr = $null
+    if ( "$accountToAdd" -eq "Everyone" ) {
+    	$sidstr = "S-1-1-0"
+    } else {
+        try {
+            $ntprincipal = new-object System.Security.Principal.NTAccount "$accountToAdd"
+            $sid = $ntprincipal.Translate([System.Security.Principal.SecurityIdentifier])
+            $sidstr = $sid.Value.ToString()
+        } catch {
+            $sidstr = $null
+        }
+    }
+    Write-Host "Account: $($accountToAdd)" -ForegroundColor DarkCyan
+    if( [string]::IsNullOrEmpty($sidstr) ) {
+        Write-Host "Account not found!" -ForegroundColor Red
+        exit -1
+    }
+    Write-Host "Account SID: $($sidstr)" -ForegroundColor DarkCyan
+    $tmp = [System.IO.Path]::GetTempFileName()
+    Write-Host "Export current Local Security Policy" -ForegroundColor DarkCyan
+    secedit.exe /export /cfg "$($tmp)" 
+    $c = Get-Content -Path $tmp 
+    $currentSetting = ""
+    foreach($s in $c) {
+        if( $s -like "SECreateSymbolicLinkPrivilege*") {
+            $x = $s.split("=",[System.StringSplitOptions]::RemoveEmptyEntries)
+            $currentSetting = $x[1].Trim()
+        }
+    }
+    if( $currentSetting -notlike "*$($sidstr)*" ) {
+        Write-Host "Need to add permissions to SymLink" -ForegroundColor Yellow
+        
+        Write-Host "Modify Setting ""Create SymLink""" -ForegroundColor DarkCyan
+
+        if( [string]::IsNullOrEmpty($currentSetting) ) {
+            $currentSetting = "*$($sidstr)"
+        } else {
+            $currentSetting = "*$($sidstr),$($currentSetting)"
+        }
+        Write-Host "$currentSetting"
+    $outfile = @"
+[Unicode]
+Unicode=yes
+[Version]
+signature="`$CHICAGO`$"
+Revision=1
+[Privilege Rights]
+SECreateSymbolicLinkPrivilege = $($currentSetting)
+"@
+    $tmp2 = [System.IO.Path]::GetTempFileName()
+        Write-Host "Import new settings to Local Security Policy" -ForegroundColor DarkCyan
+        $outfile | Set-Content -Path $tmp2 -Encoding Unicode -Force
+        Push-Location (Split-Path $tmp2)
+        try {
+            secedit.exe /configure /db "secedit.sdb" /cfg "$($tmp2)" /areas USER_RIGHTS 
+        } finally { 
+            Pop-Location
+        }
+    } else {
+        Write-Host "NO ACTIONS REQUIRED! Account already in ""Create SymLink""" -ForegroundColor DarkCyan
+        Write-Host "Account $accountToAdd already has permissions to SymLink" -ForegroundColor Green
+        return $true;
+    }
+}
+
+if ( "$mytype" -eq "user" ) {
+    addSymLinkPermissions $(Get-WMIObject -class Win32_ComputerSystem | select username).username
+} else {
+    addSymLinkPermissions Everyone
+}
diff --git a/scripts/innosetup/addsymlink.iss b/scripts/innosetup/addsymlink.iss
new file mode 100644
index 000000000..33fa407af
--- /dev/null
+++ b/scripts/innosetup/addsymlink.iss
@@ -0,0 +1,14 @@
+procedure AddSymLink();
+var
+    ErrorCode: Integer;
+    SRCdir: String;
+begin
+    SRCdir := ExpandConstant('{app}');
+    if isUninstaller() = false then begin
+        if not ShellExec('', 'powershell.exe', '-noninteractive -windowstyle hidden -executionpolicy bypass -File "' + SRCdir + '\addSymLinkPermissions.ps1" -mytype ' + SymLinkType, '', SW_HIDE, ewWaitUntilTerminated, ErrorCode) then begin
+            MsgBox('Failed to automatically grant SeCreateSymbolicLinkPrivilege. Please download Polsedit(www.southsoftware.com/polsedit.zip). Launch polseditx32.exe or polseeditx64.exe (depending on your Windows version), navigate to "Security Settings" -> "User Rights Assignment", add the account(s) to the list named "Create symbolic links", logoff and login back into your account. More info at https://github.com/git-for-windows/git/wiki/Symbolic-Links.', mbInformation, MB_OK);
+        end else begin
+            MsgBox('Automatically added SeCreateSymbolicLinkPrivilege. Please logoff and login back into your account in order for the change to take effect.', mbInformation, MB_OK);
+        end;
+    end;
+end;
diff --git a/scripts/innosetup/setup.iss b/scripts/innosetup/setup.iss
index d6b169407..35b3fae1a 100644
--- a/scripts/innosetup/setup.iss
+++ b/scripts/innosetup/setup.iss
@@ -35,14 +35,18 @@ Source: "{#MyAppDir}\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs c
 [Tasks]
 Name: modifypath; Description: Adds dvc's application directory to environmental path; Flags: checkablealone;
 Name: modifypath\system; Description: Adds dvc's application directory to enviromental path for all users;
+Name: addsymlinkpermissions; Description: Add permission for creating symbolic links; Flags: checkablealone;
+Name: addsymlinkpermissions\system; Description: Add permissions for creating symbolic links for all users;
 
 [Code]
 const
 	ModPathName = 'modifypath';
 	ModPathPath = '{app}';
+	SymLinkName = 'addsymlinkpermissions';
 
 var
 	ModPathType: String;
+	SymLinkType: String;
 
 function GetDefaultDirName(Dummy: string): string;
 begin
@@ -54,6 +58,7 @@ begin
 end;
 
 #include "modpath.iss"
+#include "addsymlink.iss"
 
 procedure CurStepChanged(CurStep: TSetupStep);
 begin
@@ -64,7 +69,15 @@ begin
 			ModPathType := 'user';
 		end;
 
+		if IsTaskSelected(SymLinkName + '\system') then begin
+			SymLinkType := 'system';
+		end else begin
+			SymLinkType := 'user';
+		end;
+
 		if IsTaskSelected(ModPathName) then
 			ModPath();
+		if IsTaskSelected(SymLinkName) then
+			AddSymLink();
 	end;
 end;
diff --git a/setup.py b/setup.py
index 1417ee58d..3d5fa9e87 100644
--- a/setup.py
+++ b/setup.py
@@ -25,6 +25,7 @@ install_requires = [
     "paramiko>=2.4.1",
     "binaryornot==0.4.4",
     "jsonpath-rw==1.4.0",
+    "reflink==0.2.0",
 ]
 
 setup(
diff --git a/tests/test_cache.py b/tests/test_cache.py
index 5520b28b9..5811a5c76 100644
--- a/tests/test_cache.py
+++ b/tests/test_cache.py
@@ -2,7 +2,6 @@ import os
 
 from dvc.cache import Cache
 from dvc.system import System
-from dvc.output import Output
 
 from tests.basic_env import TestDvc
 
@@ -27,23 +26,6 @@ class TestCache(TestDvc):
         cache = Cache(self.dvc.dvc_dir).get(self.cache1_md5)
         self.assertEquals(cache, self.cache1)
 
-    def test_find_cache(self):
-        fname1 = os.path.basename(self.cache1)
-        fname1_md5 = self.cache1_md5
-        fname2 = os.path.basename(self.cache2)
-        fname2_md5 = self.cache2_md5
-        fname3 = 'non_existing'
-
-        System.hardlink(self.cache1, fname1)
-        System.hardlink(self.cache2, fname2)
-
-        cache = Cache(self.dvc.dvc_dir).find_cache([fname1, fname2, fname3])
-
-        expected = {fname1: fname1_md5, fname2: fname2_md5}
-
-        self.assertEqual(len(cache), 2)
-        self.assertEqual(cache, expected)
-
 
 class TestCacheLoadBadDirCache(TestDvc):
     def _do_test(self, ret):
@@ -53,8 +35,8 @@ class TestCacheLoadBadDirCache(TestDvc):
     def test(self):
         fname = 'not-json'
         self.create(fname, '<clearly>not,json')
-        self._do_test(Output.load_dir_cache(fname))
+        self._do_test(Cache.load_dir_cache(fname))
 
         fname = 'not-list'
         self.create(fname, '{"a": "b"}')
-        self._do_test(Output.load_dir_cache(fname))
+        self._do_test(Cache.load_dir_cache(fname))
diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py
index 171cd43a1..8a831c3b5 100644
--- a/tests/test_data_cloud.py
+++ b/tests/test_data_cloud.py
@@ -102,7 +102,7 @@ def sleep():
 
 class TestDataCloud(TestDvc):
     def _test_cloud(self, config, cl):
-        cloud = DataCloud(cache=self.dvc.cache, state=self.dvc.state, config=config)
+        cloud = DataCloud(cache=self.dvc.cache, config=config)
         self.assertIsInstance(cloud._cloud, cl)
 
     def test(self):
@@ -116,7 +116,7 @@ class TestDataCloud(TestDvc):
         with self.assertRaises(ConfigError) as cx:
             config = TEST_CONFIG
             config[TEST_SECTION][Config.SECTION_REMOTE_URL] = 'notsupportedscheme://a/b'
-            DataCloud(cache=self.dvc.cache, state=self.dvc.state, config=config)
+            DataCloud(cache=self.dvc.cache, config=config)
 
 
 class TestDataCloudBase(TestDvc):
@@ -139,7 +139,6 @@ class TestDataCloudBase(TestDvc):
         config = TEST_CONFIG
         config[TEST_SECTION][Config.SECTION_REMOTE_URL] = repo
         cloud_settings = CloudSettings(cache=self.dvc.cache,
-                                       state=self.dvc.state,
                                        cloud_config=config[TEST_SECTION])
         self.cloud = self._get_cloud_class()(cloud_settings)
 
diff --git a/tests/test_state.py b/tests/test_state.py
index 20e87cbfa..10ec64005 100644
--- a/tests/test_state.py
+++ b/tests/test_state.py
@@ -16,7 +16,7 @@ class TestState(TestDvc):
         mtime = os.path.getmtime(path)
         inode = System.inode(path)
 
-        state = State(self.dvc.root_dir, self.dvc.dvc_dir)
+        state = State(self.dvc.dvc_dir)
 
         entry_md5 = state.update(path)
         self.assertEqual(entry_md5, md5)
@@ -41,6 +41,6 @@ class TestState(TestDvc):
 
         # Reload state db to make sure that it stays the same
         old_db = state._db
-        state = State(self.dvc.root_dir, self.dvc.dvc_dir)
+        state = State(self.dvc.dvc_dir)
         new_db = state._db
         self.assertEqual(old_db, new_db)