Add support for float timeouts using alarms (#8737)

src/prefect/_internal/concurrency/timeouts.py

@@ -4,7 +4,6 @@
 
 import contextlib
 import ctypes
-import math
 import signal
 import sys
 import threading
@@ -228,12 +227,14 @@ def raise_alarm_as_timeout(signum, frame):
         )
         raise TimeoutError()
 
+    signal.signal(signal.SIGALRM, raise_alarm_as_timeout)
+    # Use `setitimer` instead of `signal.alarm` for float support; raises a SIGALRM
+    previous = signal.setitimer(signal.ITIMER_REAL, timeout)
     try:
-        signal.signal(signal.SIGALRM, raise_alarm_as_timeout)
-        signal.alarm(math.ceil(timeout))  # alarms do not support floats
         yield ctx
     finally:
-        signal.alarm(0)  # Clear the alarm when the context exits
+        # Clear the alarm when the context exits
+        signal.setitimer(signal.ITIMER_REAL, *previous)
 
 
 @contextlib.contextmanager

tests/_internal/concurrency/test_timeouts.py

@@ -65,13 +65,11 @@ def test_cancel_sync_after_in_main_thread():
     t0 = time.perf_counter()
     with pytest.raises(TimeoutError):
         with cancel_sync_after(0.1) as ctx:
-            # floats are not suppported by alarm timeouts so this will actually timeout
-            # after 1s
-            time.sleep(2)
+            time.sleep(1)
     t1 = time.perf_counter()
 
     assert ctx.cancelled()
-    assert t1 - t0 < 2
+    assert t1 - t0 < 1
 
 
 def test_cancel_sync_after_in_worker_thread():
@@ -109,10 +107,8 @@ def test_cancel_sync_at():
     t0 = time.perf_counter()
     with pytest.raises(TimeoutError):
         with cancel_sync_at(get_deadline(timeout=0.1)) as ctx:
-            # floats are not suppported by alarm timeouts so this will actually timeout
-            # after 1s
-            time.sleep(2)
+            time.sleep(1)
     t1 = time.perf_counter()
 
     assert ctx.cancelled()
-    assert t1 - t0 < 2
+    assert t1 - t0 < 1

tests/_internal/concurrency/test_waiters.py

@@ -74,7 +74,7 @@ def test_sync_waiter_timeout_in_main_thread():
     with WorkerThread(run_once=True) as portal:
 
         def on_worker_thread():
-            callback = Call.new(time.sleep, 2)
+            callback = Call.new(time.sleep, 1)
             call.add_callback(callback)
             return callback
 
@@ -93,8 +93,7 @@ def on_worker_thread():
     with pytest.raises(TimeoutError):
         callback.result()
 
-    # main thread timeouts round up to the nearest second
-    assert t1 - t0 < 2
+    assert t1 - t0 < 1
 
     assert callback.cancelled()
     assert not call.cancelled()