Function _has_deadlocked

Lib/importlib/_bootstrap.py:183–223

Check if 'target_id' is holding the same lock as another thread(s). The search within 'blocking_on' starts with the threads listed in 'candidate_ids'. 'seen_ids' contains any threads that are considered already traversed in the search. Keyword arguments: target_id -- The t

(target_id, *, seen_ids, candidate_ids, blocking_on)

def _has_deadlocked(target_id, *, seen_ids, candidate_ids, blocking_on):
    """Check if 'target_id' is holding the same lock as another thread(s).

    The search within 'blocking_on' starts with the threads listed in
    'candidate_ids'.  'seen_ids' contains any threads that are considered
    already traversed in the search.

    Keyword arguments:
    target_id     -- The thread id to try to reach.
    seen_ids      -- A set of threads that have already been visited.
    candidate_ids -- The thread ids from which to begin.
    blocking_on   -- A dict representing the thread/blocking-on graph.  This may
                     be the same object as the global '_blocking_on' but it is
                     a parameter to reduce the impact that global mutable
                     state has on the result of this function.
    """
    if target_id in candidate_ids:
        # If we have already reached the target_id, we're done - signal that it
        # is reachable.
        return True

    # Otherwise, try to reach the target_id from each of the given candidate_ids.
    for tid in candidate_ids:
        if not (candidate_blocking_on := blocking_on.get(tid)):
            # There are no edges out from this node, skip it.
            continue
        elif tid in seen_ids:
            # bpo 38091: the chain of tid's we encounter here eventually leads
            # to a fixed point or a cycle, but does not reach target_id.
            # This means we would not actually deadlock.  This can happen if
            # other threads are at the beginning of acquire() below.
            return False
        seen_ids.add(tid)

        # Follow the edges out from this thread.
        edges = [lock.owner for lock in candidate_blocking_on]
        if _has_deadlocked(target_id, seen_ids=seen_ids, candidate_ids=edges,
                blocking_on=blocking_on):
            return True

    return False


class _ModuleLock: