space_lock.py

"""Lock exclusivo para um unico treino por GPU no Space (evita varios train.py em paralelo)."""

from __future__ import annotations

import fcntl
import os
from collections.abc import Generator
from contextlib import contextmanager


def _lock_path() -> str:
    return os.environ.get("EDA_TRAINING_LOCK_PATH", "/tmp/eda_training_single.lock")


@contextmanager
def exclusive_training_lock() -> Generator[None, None, None]:
    path = _lock_path()
    fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o644)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX)
        yield
    finally:
        fcntl.flock(fd, fcntl.LOCK_UN)
        os.close(fd)