Let's say that you want to start a thread and run some background processing there but also you want to stop the program when you press CTRL+C or set a time out. (if you want to close all threads scroll down to the end)
We have 2 options:
- Just close the thread and don't care what happens
- Wait for the background thread to finish the current task and then return from it
We can achieve option 1 using daemon threads, but keep in mind:
Note
Daemon threads are abruptly stopped at shutdown. Their resources (such as open files, database transactions, etc.) may not be released properly. If you want your threads to stop gracefully, make them non-daemonic and use a suitable signalling mechanism such as an
Event. (from python documentation)
If we manage to close the main-thread all the daemon threads will be forced to close.
Using daemon threads for tasks that can be interrupted
1. Using try/except for KeyboardInterrupt
Notice that we don't use join() , because is blocking, we can't detect when CTRL+C has been pressed when we are blocked. Because of this we use is_alive() to check if the thread finished or not.
#!/usr/bin/env python3 import sys import threading from time import sleep def process(): """Long lasting operation""" sleep(300) print("finished work") def main(argv): t = threading.Thread(target=process) t.daemon = True t.start() try: while t.is_alive(): print("Waiting for background thread to finish") sleep(1) print("Thread finished task, exiting") except KeyboardInterrupt: print("Closing main-thread.This will also close the background thread because is set as daemon.") return 0 if __name__ == "__main__": sys.exit(main(sys.argv))
2. Using signal.signal to exit main thread
In this example we use signal.signal() to close the main-thread, when CTRL+ C is pressed.
SIGINT is the signal for CTRL+C and the signal_handler is the function to be executed when the SIGINT is detected.
A new signal handler can be set only from the main-thread otherwise an exception is raised.
#!/usr/bin/env python3 import signal import sys import threading from time import sleep def process(): """Long lasting operation""" sleep(300) print("finished work") def signal_handler(signal, frame): print("Closing main-thread.This will also close the background thread because is set as daemon.") sys.exit(0) def main(argv): signal.signal(signal.SIGINT, signal_handler) t = threading.Thread(target=process) t.daemon = True t.start() while t.is_alive(): print("Waiting for background thread to finish") sleep(1) print("Thread finished task, exiting") return 0 if __name__ == "__main__": sys.exit(main(sys.argv))
3. Using join() to add a timeout and then close all threads (can't catch CTRL+C)
We can see that join() supports timeout from the signature:
The call to join() is blocking, we can't catch CTRL+C, so we will see another example how to add timeout and catch CTRL+C. You can use this if in your context you don't care about CTRL+C.
Don't use join() on a daemon thread without timeout, by definition daemon threads are infinite loops but nobody stops you to mark a thread as daemon that is not an infinite loop.
#!/usr/bin/env python3 import sys import threading from time import sleep TIMEOUT = 10 def process(): """Long lasting operation""" sleep(12) print("finished work") def main(argv): """ This example has timeout using the join() method. The join() is blocking, meaning that an SIGINT event(CTRL+C) can't be detected. :param argv: :return: """ t = threading.Thread(target=process) t.daemon = True t.start() print(f"Waiting {TIMEOUT} seconds for background thread to finish.") t.join(TIMEOUT) if t.is_alive(): print("Background thread timed out. Closing all threads") return -1 else: print("Background thread finished processing. Closing all threads") return 0 if __name__ == "__main__": sys.exit(main(sys.argv))
4. Using is_alive() for timeout and catching CTRL+C
Using is_alive() and a while loop we can add a timeout to a thread and also catch the CTRL+C.
#!/usr/bin/env python3 import sys import threading from time import sleep TIMEOUT = 10 def process(): """Long lasting operation""" sleep(12) print("finished work") def main(argv): """ This example combines timeout and KeyboardInterrupt. If detected KeyboardInterrupt all threads are closed. The background thread is close automatically because is set as daemon. :param argv: :return: """ t = threading.Thread(target=process) t.daemon = True t.start() print(f"Waiting {TIMEOUT} seconds for background thread to finish.") try: loop_thread(t) except KeyboardInterrupt: print("KeyboardInterrupt detected closing all threads") def loop_thread(t): i = 1 while i <= TIMEOUT: if t.is_alive(): print("Background thread processing, please wait.") sleep(1) i += 1 else: print("Background thread finished processing. Closing all threads") if i == TIMEOUT + 1: print("Timeout occurred.") if __name__ == "__main__": sys.exit(main(sys.argv))
Closing an non daemon thread gracefully (waiting for thread to finish work)
1. Using Event() to stop the thread gracefully
As stated in the python doc, this example uses an Event() which will be set when CTRL+C is pressed. The background thread will check this flag every time it gets a new item to process, if set will not get any new items and just return.
This thread is not marked as daemon meaning that the main-thread can't close if this thread hasn't finished executing.
If we do one time background job we can just have the code in the thread target function, if we want the thread to loop over some items we can use a queue.Queue()
Also notice that we join() on the Queue object , this way we make sure that all work items are done.
#!/usr/bin/env python3 import queue import sys import threading from time import sleep stop = threading.Event() work_queue = queue.Queue() work_queue.put('Item 1') work_queue.put('Item 2') def process(): """Long lasting operation""" while not stop.isSet() and not work_queue.empty(): sleep(5) item = work_queue.get() # process item print(f"Done task: {item}") work_queue.task_done() if stop.isSet(): print("KeyboardInterrupt detected, closing background thread. ") def main(argv): t = threading.Thread(target=process) t.start() try: while t.is_alive(): print("Waiting for background thread to finish") sleep(1) except KeyboardInterrupt: stop.set() print("Closing main-thread.Please wait for background thread to finish the current item.") return 0 work_queue.join() print("Thread finished all tasks, exiting") return 0 if __name__ == "__main__": sys.exit(main(sys.argv))
Forcefully closing all threads os._exit(), even those that are not a daemon thread
If you just want close everything you can use os._exit()
You can see this being used in sqlmap (open source penetration testing tool).
Use with caution, don't just copy paste this everywhere.
activeCount() returns the number of threads but it also includes the main-thread.
#!/usr/bin/env python3 import os import queue import sys import threading from time import sleep work_queue = queue.Queue() work_queue.put('Item 1') work_queue.put('Item 2') def process(): """Long lasting operation""" while not work_queue.empty(): sleep(5) item = work_queue.get() # process item print(f"Done task: {item}") work_queue.task_done() def main(argv): t = threading.Thread(target=process) t.start() try: while t.is_alive(): print("Waiting for background thread to finish") sleep(1) except KeyboardInterrupt: print("Closing main-thread.Please wait for background thread to finish the current item.") if threading.activeCount() > 1: os._exit(getattr(os, "_exitcode", 0)) else: sys.exit(getattr(os, "_exitcode", 0)) work_queue.join() print("Thread finished all tasks, exiting") return 0 if __name__ == "__main__": sys.exit(main(sys.argv))
You could also look into concurrent.futures depending on the situation the code could be easier to write using concurrent.futures.
Consider use of timeout with threads all the time in production, if code shouldn't take more than 10 seconds use timeout, is always better to see an error in a log file that threaded timed out than having the issue hidden for probably hours and then you start debugging, there is no need to lose time and money.
When to use threads? When you wait on I/O. If you do cpu intensive work look into subprocess module
If you want to learn more about threads and concurrency in python you can watch this youtube video by python core developer Raymond Hettinger.
Comments
Post a Comment