rwlock auto solve corrupted lockfile

dvc/rwlock.py

@@ -4,6 +4,7 @@ import os
 from collections import defaultdict
 from contextlib import contextmanager
 
+import psutil
 from voluptuous import Invalid, Optional, Required, Schema
 
 from .exceptions import DvcException
@@ -75,18 +76,50 @@ def _infos_to_str(infos):
 def _check_blockers(tmp_dir, lock, info, *, mode, waiters):
     from .lock import LockError
 
-    for waiter_path in waiters:
-        blockers = [
-            blocker
-            for path, infos in lock[mode].items()
-            if localfs.path.overlaps(waiter_path, path)
-            for blocker in (infos if isinstance(infos, list) else [infos])
-            if blocker != info
-        ]
+    non_existing_pid = set()
 
-        if not blockers:
+    blockers = []
+    to_release = defaultdict(list)
+    for path, infos in lock[mode].items():
+        for waiter_path in waiters:
+            if localfs.path.overlaps(waiter_path, path):
+                break
+        else:
             continue
 
+        infos = infos if isinstance(infos, list) else [infos]
+        for blocker in infos:
+            if blocker == info:
+                continue
+
+            pid = int(blocker["pid"])
+
+            if pid in non_existing_pid:
+                pass
+            elif psutil.pid_exists(pid):
+                blockers.append(blocker)
+                continue
+            else:
+                non_existing_pid.add(pid)
+                cmd = blocker["cmd"]
+                logger.warning(
+                    "Process '%s' with (Pid %s), in RWLock-file '%s'"
+                    " had been killed. Auto remove it from the lock file.",
+                    cmd,
+                    pid,
+                    relpath(path),
+                )
+            to_release[json.dumps(blocker, sort_keys=True)].append(path)
+
+    if to_release:
+        for info_json, path_list in to_release.items():
+            info = json.loads(info_json)
+            if mode == "read":
+                _release_read(lock, info, path_list)
+            elif mode == "write":
+                _release_write(lock, info, path_list)
+
+    if blockers:
         raise LockError(
             f"'{waiter_path}' is busy, it is being blocked by:\n"
             f"{_infos_to_str(blockers)}\n"
@@ -99,6 +132,8 @@ def _check_blockers(tmp_dir, lock, info, *, mode, waiters):
 def _acquire_read(lock, info, paths):
     changes = []
 
+    lock["read"] = lock.get("read", defaultdict(list))
+
     for path in paths:
         readers = lock["read"][path]
         if info in readers:
@@ -113,6 +148,8 @@ def _acquire_read(lock, info, paths):
 def _acquire_write(lock, info, paths):
     changes = []
 
+    lock["write"] = lock.get("write", defaultdict(dict))
+
     for path in paths:
         if lock["write"][path] == info:
             continue

tests/unit/test_rwlock.py

@@ -1,3 +1,4 @@
+import json
 import os
 
 import pytest
@@ -105,3 +106,37 @@ def test_broken_rwlock(tmp_path):
     with pytest.raises(RWLockFileCorruptedError):
         with _edit_rwlock(dir_path, localfs, False):
             pass
+
+
+@pytest.mark.parametrize("return_value", [True, False])
+def test_corrupted_rwlock(tmp_path, mocker, return_value):
+    dir_path = os.fspath(tmp_path)
+    path = tmp_path / "rwlock"
+
+    foo = "foo"
+    bar = "bar"
+    cmd_foo = "cmd_foo"
+    cmd_bar = "cmd_bar"
+    mocker.patch("psutil.pid_exists", return_value=return_value)
+
+    corrupted_rwlock = {
+        "write": {foo: {"pid": 1234, "cmd": cmd_foo}},
+        "read": {
+            foo: [{"pid": 5555, "cmd": cmd_foo}],
+            bar: [
+                {"pid": 6666, "cmd": cmd_bar},
+                {"pid": 7777, "cmd": cmd_bar},
+            ],
+        },
+    }
+
+    path.write_text(json.dumps(corrupted_rwlock), encoding="utf-8")
+
+    if return_value:
+        with pytest.raises(LockError):
+            with rwlock(dir_path, localfs, "cmd_other", [], [foo, bar], False):
+                pass
+    else:
+        with rwlock(dir_path, localfs, "cmd_other", [], [foo, bar], False):
+            pass
+        assert path.read_text() == """{"read": {}}"""