oc_chat
Loading...
Searching...
No Matches
lock.py File Reference

Go to the source code of this file.

Classes

class  oc_chat.registry.regenerate.lock.LockReclaimEvent
class  oc_chat.registry.regenerate.lock.LockHandle
class  oc_chat.registry.regenerate.lock._LockRecord

Namespaces

namespace  oc_chat
namespace  oc_chat.registry
namespace  oc_chat.registry.regenerate
namespace  oc_chat.registry.regenerate.lock

Functions

str oc_chat.registry.regenerate.lock._default_clock ()
"tuple[str, Optional[str]]" oc_chat.registry.regenerate.lock._probe_posix (int pid)
Optional[str] oc_chat.registry.regenerate.lock._proc_start_time_posix (int pid)
"tuple[str, Optional[str]]" oc_chat.registry.regenerate.lock._probe_windows (int pid)
"tuple[str, Optional[str]]" oc_chat.registry.regenerate.lock._default_probe (int pid)
Optional[_LockRecord] oc_chat.registry.regenerate.lock._read_record (Path lock_path)
None oc_chat.registry.regenerate.lock._atomic_create (Path lock_path, int pid, str start_time, str acquired_at)
LockReclaimEvent oc_chat.registry.regenerate.lock._make_event (Path lock_path, Optional[_LockRecord] record, str reason, *, bool force, Clock clock)
LockReclaimEvent oc_chat.registry.regenerate.lock._decide_reclaim (Path lock_path, Optional[_LockRecord] record, bool force, Probe probe, Clock clock)
LockHandle oc_chat.registry.regenerate.lock.acquire (Path lock_path, *, bool force=False, Optional[Clock] clock=None, Optional[Probe] probe=None)
None oc_chat.registry.regenerate.lock.release (LockHandle handle)

Variables

str oc_chat.registry.regenerate.lock.PROBE_DEAD = "dead"
str oc_chat.registry.regenerate.lock.PROBE_ALIVE = "alive"
str oc_chat.registry.regenerate.lock.PROBE_INCONCLUSIVE = "inconclusive"
str oc_chat.registry.regenerate.lock.REASON_DEAD_PID = "dead_pid"
str oc_chat.registry.regenerate.lock.REASON_START_TIME_MISMATCH = "start_time_mismatch"
str oc_chat.registry.regenerate.lock.REASON_FORCED = "forced_live_or_inconclusive"
 oc_chat.registry.regenerate.lock.Clock = Callable[[], str]
 oc_chat.registry.regenerate.lock.Probe = Callable[[int], "tuple[str, Optional[str]]"]