Reliable version of os.replace for Windows (wait/retry loop)

mypy/build.py

@@ -901,7 +901,7 @@ def write_cache(id: str, path: str, tree: MypyFile,
             f.write(data_str)
             f.write('\n')
         data_mtime = os.path.getmtime(data_json_tmp)
-        os.replace(data_json_tmp, data_json)
+        util.replace(data_json_tmp, data_json)
         manager.trace("Interface for {} has changed".format(id))
 
     mtime = st.st_mtime
@@ -927,7 +927,7 @@ def write_cache(id: str, path: str, tree: MypyFile,
             json.dump(meta, f, indent=2, sort_keys=True)
         else:
             json.dump(meta, f)
-    os.replace(meta_json_tmp, meta_json)
+    util.replace(meta_json_tmp, meta_json)
 
     return interface_hash
 

mypy/test/testutil.py

@@ -0,0 +1,83 @@
+from typing import Iterator, List, Tuple
+import time
+import os
+import sys
+import tempfile
+from contextlib import contextmanager
+from threading import Thread
+from unittest import TestCase, main, skipUnless
+from mypy import util
+from mypy.build import random_string
+
+
+WIN32 = sys.platform.startswith("win")
+
+
+def lock_file(filename: str, duration: float) -> Thread:
+    """Open a file and keep it open in a background thread for duration sec."""
+    def _lock_file() -> None:
+        with open(filename):
+            time.sleep(duration)
+    t = Thread(target=_lock_file, daemon=True)
+    t.start()
+    return t
+
+
+@skipUnless(WIN32, "only relevant for Windows")
+class WindowsReplace(TestCase):
+    tmpdir = tempfile.TemporaryDirectory(prefix='mypy-test-',
+                                         dir=os.path.abspath('tmp-test-dirs'))
+    # Choose timeout value that would ensure actual wait inside util.replace is close to timeout
+    timeout = 0.0009 * 2 ** 10
+    short_lock = timeout / 4
+    long_lock = timeout * 2
+
+    threads = []  # type: List[Thread]
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        # Need to wait for threads to complete, otherwise we'll get PermissionError
+        # at the end (whether tmpdir goes out of scope or we explicitly call cleanup).
+        for t in cls.threads:
+            t.join()
+        cls.tmpdir.cleanup()
+
+    def prepare_src_dest(self, src_lock_duration: float, dest_lock_duration: float
+                         ) -> Tuple[str, str]:
+        # Create two temporary files with random names.
+        src = os.path.join(self.tmpdir.name, random_string())
+        dest = os.path.join(self.tmpdir.name, random_string())
+
+        for fname in (src, dest):
+            with open(fname, 'w') as f:
+                f.write(fname)
+
+        self.threads.append(lock_file(src, src_lock_duration))
+        self.threads.append(lock_file(dest, dest_lock_duration))
+        return src, dest
+
+    def replace_ok(self, src_lock_duration: float, dest_lock_duration: float,
+                   timeout: float) -> None:
+        src, dest = self.prepare_src_dest(src_lock_duration, dest_lock_duration)
+        util._replace(src, dest, timeout=timeout)
+        self.assertEqual(open(dest).read(), src, 'replace failed')
+
+    def test_everything(self) -> None:
+        # No files locked.
+        self.replace_ok(0, 0, self.timeout)
+        # Make sure we can reproduce the issue with our setup.
+        src, dest = self.prepare_src_dest(self.short_lock, 0)
+        with self.assertRaises(PermissionError):
+            os.replace(src, dest)
+        # Lock files for a time short enough that util.replace won't timeout.
+        self.replace_ok(self.short_lock, 0, self.timeout)
+        self.replace_ok(0, self.short_lock, self.timeout)
+        # Lock files for a time long enough that util.replace times out.
+        with self.assertRaises(PermissionError):
+            self.replace_ok(self.long_lock, 0, self.timeout)
+        with self.assertRaises(PermissionError):
+            self.replace_ok(0, self.long_lock, self.timeout)
+
+
+if __name__ == '__main__':
+    main()

mypy/util.py

@@ -2,8 +2,16 @@
 
 import re
 import subprocess
+import os
+import sys
+import math
+import time
+from itertools import count
 from xml.sax.saxutils import escape
-from typing import TypeVar, List, Tuple, Optional, Sequence, Dict
+from typing import (
+    TypeVar, List, Tuple, Optional, Sequence, Dict, Callable, Iterable, Type, Union,
+    AnyStr, Generic
+)
 
 
 T = TypeVar('T')
@@ -134,3 +142,39 @@ def id(self, o: object) -> int:
             self.id_map[o] = self.next_id
             self.next_id += 1
         return self.id_map[o]
+
+
+if sys.version_info >= (3, 6):
+    PathType = Union[AnyStr, 'os.PathLike[AnyStr]']
+else:
+    # Workaround to allow the use of generic in _replace
+    class _PathType(Generic[AnyStr]): ...
+    PathType = _PathType[AnyStr]
+
+
+def _replace(src: PathType[AnyStr], dest: PathType[AnyStr], timeout: float = 1) -> None:
+    """Replace src with dest using os.replace(src, dest).
+
+    Wait and retry if OSError exception is raised.
+
+    Increase wait time exponentially until total wait of timeout sec;
+    on timeout, give up and reraise the last exception seen.
+    """
+    wait = 0.001
+    while True:
+        try:
+            os.replace(src, dest)
+        except PermissionError:
+            # The actual total wait might be up to 2x larger than timeout parameter
+            if wait > timeout:
+                raise
+        else:
+            return
+        time.sleep(wait)
+        wait *= 2
+
+
+if sys.platform.startswith("win"):
+    replace = _replace
+else:
+    replace = os.replace

runtests.py

@@ -215,6 +215,7 @@ def add_imports(driver: Driver) -> None:
     'testdiff',
     'testfinegrained',
     'testmerge',
+    'testutil',
 ]]