Function _pin_fork_objects

src/filelock/_api.py:1629–1649

()

def _pin_fork_objects() -> None:  # pragma: needs fork
    _ensure_current_process()
    thread_id = get_ident()
    with _FORK_STATE.gate:
        _FORK_STATE.fork_owner_depths[thread_id] = _FORK_STATE.fork_owner_depths.get(thread_id, 0) + 1
        _FORK_STATE.admission_closed = True
        while _FORK_STATE.active_transitions > len(_FORK_STATE.transitions.get(thread_id, ())):
            _FORK_STATE.gate.wait()
        transition_owners = tuple(_FORK_STATE.transitions.get(thread_id, {}).values())
    _verify_unverified_descriptors()
    owners = {id(owner): owner for owner in transition_owners if owner is not None}
    provisional_descriptor_tokens = tuple(
        starmap(
            _snapshot_descriptor_for_fork,
            (descriptor for owner in owners.values() for descriptor in owner._descriptors_for_fork()),  # ruff:ignore[private-member-access]  # snapshots each owner's own fork descriptors
        )
    )
    with _FORK_STATE.registry_lock:
        _FORK_STATE.provisional_descriptor_tokens.setdefault(thread_id, []).append(provisional_descriptor_tokens)
        _FORK_STATE.pinned_objects.setdefault(thread_id, []).append(tuple(_FORK_OBJECTS.values()))
        _FORK_STATE.pinned_classes.setdefault(thread_id, []).append(tuple(_FORK_CLASSES.values()))


def _resume_parent_after_fork() -> None:  # pragma: needs fork