82 lines
2.2 KiB
Python
82 lines
2.2 KiB
Python
"""
|
|
Межпроцессная блокировка для POST /api/sync/pull по node_id.
|
|
|
|
threading.Lock в SyncRuntime не действует между процессами (Flask reloader, Gunicorn
|
|
workers): два параллельных pull с одним client_id → два process_pull на SQLite и зависание.
|
|
На POSIX используем flock на отдельном файле; на Windows — не используем (остаётся thread lock).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from typing import BinaryIO, Optional
|
|
|
|
_HAVE_FCNTL = False
|
|
if sys.platform != "win32":
|
|
try:
|
|
import fcntl # noqa: F401
|
|
|
|
_HAVE_FCNTL = True
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
def ipc_pull_supported() -> bool:
|
|
return _HAVE_FCNTL
|
|
|
|
|
|
def _lock_file_path(client_id: str) -> Path:
|
|
cid = (client_id or "").strip()
|
|
h = hashlib.sha256(cid.encode("utf-8")).hexdigest()[:32]
|
|
base = Path(os.getenv("WESP_SYNC_PULL_IPC_LOCK_DIR", tempfile.gettempdir())) / "wesp-sync-pull"
|
|
base.mkdir(parents=True, exist_ok=True)
|
|
return base / f"{h}.lock"
|
|
|
|
|
|
def ipc_pull_acquire(client_id: str, timeout: float) -> Optional[BinaryIO]:
|
|
"""exclusive flock; при таймауте — None. Не вызывать, если не ipc_pull_supported()."""
|
|
import fcntl
|
|
|
|
cid = (client_id or "").strip()
|
|
if not cid:
|
|
return None
|
|
path = _lock_file_path(cid)
|
|
fp = open(path, "a+b")
|
|
deadline = time.monotonic() + max(0.0, timeout)
|
|
try:
|
|
while True:
|
|
try:
|
|
fcntl.flock(fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
return fp
|
|
except BlockingIOError:
|
|
if time.monotonic() >= deadline:
|
|
fp.close()
|
|
return None
|
|
time.sleep(0.05)
|
|
except Exception:
|
|
try:
|
|
fp.close()
|
|
except OSError:
|
|
pass
|
|
raise
|
|
|
|
|
|
def ipc_pull_release(fp: Optional[BinaryIO]) -> None:
|
|
if not fp or not _HAVE_FCNTL:
|
|
return
|
|
import fcntl
|
|
|
|
try:
|
|
fcntl.flock(fp.fileno(), fcntl.LOCK_UN)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
fp.close()
|
|
except OSError:
|
|
pass
|