Timeouts in Python
Timeouts are important for programs that connect to external resources or that otherwise need to bound execution time. Implementing timeouts in Python is straightforward using the threading
module and the Event
class.
import threading
import time
def main():
# For our example, suppose we're executing an external
# call that returns its result after 2 seconds.
# We'll use an Event to signal when the result is ready.
event = threading.Event()
result = []
def worker():
time.sleep(2)
result.append("result 1")
event.set()
# Start the worker thread
thread = threading.Thread(target=worker)
thread.start()
# Here we implement a timeout using the Event's wait method.
# If the event is set within 1 second, we'll get the result.
# Otherwise, we'll hit the timeout.
if event.wait(1):
print(result[0])
else:
print("timeout 1")
# If we allow a longer timeout of 3s, then we'll get the result
event.clear()
result.clear()
def worker2():
time.sleep(2)
result.append("result 2")
event.set()
thread2 = threading.Thread(target=worker2)
thread2.start()
if event.wait(3):
print(result[0])
else:
print("timeout 2")
if __name__ == "__main__":
main()
Running this program shows the first operation timing out and the second succeeding.
$ python timeouts.py
timeout 1
result 2
In this Python version, we use threading.Event
to simulate channels and the select
statement from Go. The Event.wait()
method allows us to implement a timeout mechanism similar to Go’s time.After()
.
We create worker functions that simulate long-running tasks, and we use threads to run these functions concurrently. The Event.wait()
method blocks until either the event is set (indicating the task is complete) or the specified timeout is reached.
This approach provides a way to implement timeouts in Python that is conceptually similar to the Go example, although the syntax and specific mechanisms differ due to the languages’ different concurrency models.