Fix Thread-Safe Trigger Synchronization

Problem Statement

In concurrent systems, worker threads often need to wait for a shared initialization step, external event, or shutdown signal before proceeding. A common pattern for this coordination is a One-Shot Trigger (similar to a countdown latch of size 1 or an event flag).

Implement a thread-safe OneShotTrigger class that coordinates multiple waiting threads and one or more firing threads with the following specifications:

• fire(): Transitions the trigger into the "fired" state and unblocks all currently waiting threads. Subsequent calls to fire() should be safe, idempotent, and have no additional effect.
• wait(timeout: float = None) -> bool: Blocks the calling thread until the trigger has been fired. If timeout is provided (in seconds, non-negative), it blocks for at most timeout seconds. Returns True if the trigger was fired (either before the call or during the wait); returns False if the wait timed out before the trigger was fired.
• is_fired() -> bool: Returns True if fire() has been called, False otherwise, without blocking.

Your implementation must be thread-safe, avoid busy-waiting, properly guard against spurious wakeups, and ensure that threads calling wait() after fire() has already occurred return immediately with True.

Interface

class OneShotTrigger:
    def __init__(self):
        pass

    def fire(self) -> None:
        """Fires the trigger and releases all waiting threads."""
        pass

    def wait(self, timeout: float = None) -> bool:
        """Waits for the trigger to be fired or until timeout seconds elapse.
        
        Returns True if fired, False if timed out.
        """
        pass

    def is_fired(self) -> bool:
        """Returns True if trigger is fired, False otherwise."""
        pass

Examples

Example 1: Basic wait and fire

trigger = OneShotTrigger()
results = []

def worker():
    triggered = trigger.wait()
    results.append(triggered)

t = threading.Thread(target=worker)
t.start()

time.sleep(0.05)
trigger.fire()
t.join()

assert results == [True]
assert trigger.is_fired() is True

Example 2: Timeout expiration

trigger = OneShotTrigger()
# No fire() called; wait should time out after 0.1 seconds
result = trigger.wait(timeout=0.1)
assert result is False
assert trigger.is_fired() is False

Constraints & Edge Cases

• Any number of threads may call wait(), fire(), and is_fired() concurrently.
• timeout can be None (wait indefinitely) or a float >= 0.0. If timeout == 0.0, return immediately based on the current state without blocking.
• Ensure condition variable predicate checks handle spurious wakeups correctly.
• Avoid race conditions where a thread registers a wait right as fire() is broadcasting.