43 lines
1.1 KiB
Python
43 lines
1.1 KiB
Python
"""Cross-process locks for launcher-prefix mutations."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from datetime import datetime
|
|
import fcntl
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Iterator
|
|
|
|
|
|
class OperationBusyError(RuntimeError):
|
|
"""Raised when another launcher-prefix operation is still active."""
|
|
|
|
|
|
@contextmanager
|
|
def launcher_operation_lock(lock_path: Path, operation: str) -> Iterator[None]:
|
|
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with lock_path.open("a+", encoding="utf-8") as handle:
|
|
try:
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError as exc:
|
|
handle.seek(0)
|
|
owner = handle.read().strip() or "unknown operation"
|
|
raise OperationBusyError(
|
|
f"launcher files are busy ({owner}); close the updater and try again"
|
|
) from exc
|
|
handle.seek(0)
|
|
handle.truncate()
|
|
handle.write(
|
|
f"{operation}, pid={os.getpid()}, "
|
|
f"started={datetime.now().astimezone().isoformat()}\n"
|
|
)
|
|
handle.flush()
|
|
try:
|
|
yield
|
|
finally:
|
|
handle.seek(0)
|
|
handle.truncate()
|
|
handle.flush()
|
|
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|