diff --git a/dvc/rwlock.py b/dvc/rwlock.py index 27f4215b2..f186bf2a0 100644 --- a/dvc/rwlock.py +++ b/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 diff --git a/tests/unit/test_rwlock.py b/tests/unit/test_rwlock.py index 6bf7ae043..0a80d7fa5 100644 --- a/tests/unit/test_rwlock.py +++ b/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": {}}"""