workaround_cpython_issue115533.py

workaround_cpython_issue115533.py

#!/usr/bin/env python3.12
# LICENSE: Apache 2.0

import itertools
import time
import threading
import sys


def issue115533_workaround_wait_for_non_daemon_threads(
    max_delay_in_secs: float = 0.2,
) -> bool:
    """Wait for non-daemon threads to complete IF the interpreter has the bug.

    ONLY call this from the main Python thread. It is intended to be called
    before exiting to prevent entering interpreter shutdown mode so that the other
    running threads may continue to spawn new threads on their own (see details in
    https://github.com/python/cpython/issues/115533).

       try:
           main(sys.argv)
       finally:
           issue115533_workaround_wait_for_non_daemon_threads()

    This relies on internal private threading implementation details. It is
    also an ugly polling hack.

    Args:
        max_delay_in_secs: controls perceived latency vs polling cpu consumption.

    Returns:
        True if we waited, False if the bug does not apply to this Python.
    """
    if sys.version_info[:2] < (3, 12):  # Adjust to a range once a fix is released.
        return False

    our_tid = threading.get_ident()
    while True:
        live_thread = None
        with threading._active_limbo_lock:
            for live_thread in itertools.chain(
                threading._active.values(),
                threading._limbo.values(),
            ):
                if live_thread.daemon or live_thread.ident == our_tid:
                    continue
                break  # Wait on live_thread post-lock-release below.
            else:
                return True  # No more non-daemon threads!

        if live_thread:
            try:
                live_thread.join(max_delay_in_secs / 2)
            except RuntimeError:
                pass  # Race condition, limbo, etc.
        time.sleep(max_delay_in_secs / 2)


### Demo

if __name__ == "__main__":

    def _print_n_sleep():
        print("+", threading.get_ident())
        time.sleep(threading.get_ident() % 2 + 1.1)
        print("-", threading.get_ident())

    def _spawn_three():
        print("+ _spawn_three")
        time.sleep(1)
        threading.Thread(target=_print_n_sleep).start()
        time.sleep(1)
        threading.Thread(target=_print_n_sleep).start()
        time.sleep(1)
        threading.Thread(target=_print_n_sleep).start()
        print("- _spawn_three")

    try:
        print("Demonstration / non-rigorous test that this 'works'.")
        threading.Thread(target=_spawn_three).start()
        time.sleep(1.1)
    finally:
        print("main complete.")
        if issue115533_workaround_wait_for_non_daemon_threads():
            print("manually waited for threads.")
        else:
            print("bug not present, waiting is done by Python itself.")

You probably do NOT want this code. I overlooked threading.enumerate(). And realistically there is no point in the timeout. See vstinner's much smaller loop on the github issue.