Getting very precise timeout timing is somewhat troublesome, as the operation of reporting on the timing can affect the timing itself if not carefully written.
Something like this is often best accomplished with two threads in parallel
Making the threads "daemon threads" allows you to end them by quitting the program (otherwise the program will wait for them to be .join()ed)
You can rely on threading.Event() to clearly communicate into a thread and provide a .wait() method, which will either wait for a given timeout or immediately end when the event .is_set()
import threading
import time
def function_to_call(): # replace me with your function
time.sleep(1000)
def timer_fn(timeout, event, delay=5): # 5 seconds between reports
time_end = time.time() + timeout
while not event.is_set():
time_remaining = int(time_end - time.time())
if time_remaining <= 0:
return
print(f"{int(time_remaining)}s remaining")
event.wait((min(delay, time_remaining))) # wait for event
timeout = int(input("minutes: ") or 0) * 60 + int(input("seconds: ") or 0)
E = threading.Event() # Event to kill t2
# making t1 a daemon lets it not prevent shutdown
t1 = threading.Thread(target=function_to_call, daemon=True)
t2 = threading.Thread(target=timer_fn, args=(timeout, E), daemon=True)
# begin both threads
t1.start()
t2.start()
# wait for t1 to exit (but not t2)
t1.join(timeout=timeout)
# t1 may join faster by ending, otherwise reach here after timeout
E.set() # set the Event to quickly end t2 (timeout or t1 returned)
t2.join() # not technically necessary, but cleans up t2
# program quits and t1 is killed if it's still running
Note that the timing display is actually separate from the thread ending, and ending the function early is done here by stopping the program. The function will continue to run if the program is not stopped!
If more advanced task control is needed, consider
- modifying your callable function to repeatedly check another
threading.Event throughout it to see if it's time is up (yet)
- use
multiprocessing, which is much more featureful than threading, though it may need to copy memory into the new process, which can be slow if you have a tremendous amount
- use
subprocess, creating a script just for the purpose of a timed end (here you can kill the subprocess with a simple .kill() or setting a timeout= argument when creating it! (though again you'll find some inefficiency in copying/streaming input into and out of the new process)
- use
os.fork() and [os.kill()](https://docs.python.org/3/library/os.html#os.kill) to split your process (advanced, but usually much more efficient than multiprocessing` due to how memory is (or rather is not) copied)
- if your function can be made asynchronous, which allows multiple tasks to collaborate in the same namespace similar to
threading, but in a more friendly way (though how this behaves is fundamentally different from the other techniques given, it can be very efficient if you have many tasks which don't rely on local resources, such as a webserver or database)
Though further fundamentally different, you may find further or more benefit in designing your task to be a collection of work to iterate over (maybe in a list) and consider a library like tqdm to report on its status