66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
|
|
|
|
class LockTimeoutError(TimeoutError):
|
|
pass
|
|
|
|
|
|
class FileLock:
|
|
"""Small cross-platform inter-process exclusive lock."""
|
|
|
|
def __init__(self, path: Path, timeout: float = 30):
|
|
self.path = Path(path)
|
|
self.timeout = timeout
|
|
self._file = None
|
|
|
|
def __enter__(self):
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._file = self.path.open("a+b")
|
|
deadline = time.monotonic() + self.timeout
|
|
while True:
|
|
try:
|
|
self._acquire()
|
|
return self
|
|
except (BlockingIOError, OSError):
|
|
if time.monotonic() >= deadline:
|
|
self._file.close()
|
|
self._file = None
|
|
raise LockTimeoutError("Another plugin lifecycle operation is still running.")
|
|
time.sleep(0.1)
|
|
|
|
def _acquire(self) -> None:
|
|
if os.name == "nt":
|
|
import msvcrt
|
|
|
|
self._file.seek(0)
|
|
if self._file.read(1) == b"":
|
|
self._file.write(b"\0")
|
|
self._file.flush()
|
|
self._file.seek(0)
|
|
msvcrt.locking(self._file.fileno(), msvcrt.LK_NBLCK, 1)
|
|
else:
|
|
import fcntl
|
|
|
|
fcntl.flock(self._file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
if self._file is None:
|
|
return
|
|
try:
|
|
if os.name == "nt":
|
|
import msvcrt
|
|
|
|
self._file.seek(0)
|
|
msvcrt.locking(self._file.fileno(), msvcrt.LK_UNLCK, 1)
|
|
else:
|
|
import fcntl
|
|
|
|
fcntl.flock(self._file.fileno(), fcntl.LOCK_UN)
|
|
finally:
|
|
self._file.close()
|
|
self._file = None
|