Skip to content

Instantly share code, notes, and snippets.

@coffeewasmyidea
Created October 4, 2023 18:14
Show Gist options
  • Save coffeewasmyidea/116478837fc37aa6263d8b6ba4a41f74 to your computer and use it in GitHub Desktop.
Save coffeewasmyidea/116478837fc37aa6263d8b6ba4a41f74 to your computer and use it in GitHub Desktop.
Python thread with tracking of its execution
import sys
import threading
class enchanted_thread(threading.Thread):
def __init__(self, *args, **keywords):
threading.Thread.__init__(self, *args, **keywords)
self.has_stopped = False
def start(self):
self.__run_backup = self.run
self.run = self.__run
threading.Thread.start(self)
def stop(self):
self.has_stopped = True
def __run(self):
"""
Sets a trace (sys.settrace) on a thread to track its execution.
"""
sys.settrace(self.globaltrace)
self.__run_backup()
self.run = self.__run_backup
def globaltrace(self, frame, event, arg):
"""
This method will be called for every event on the thread, and if the event is a "call" then
it will return a localtrace method to track the subsequent function call.
"""
if event == "call":
return self.localtrace
else:
return None
def localtrace(self, frame, event, arg):
"""
This method is also called for every event on the thread. If the killed flag is set to True
and the event is "line", then a SystemExit exception is raised, which causes the thread to
terminate.
"""
if self.has_stopped:
if event == "line":
raise SystemExit()
return self.localtrace
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment