Skip to content

Instantly share code, notes, and snippets.

@bbarrows
Created June 25, 2012 17:56
Show Gist options
  • Save bbarrows/2990205 to your computer and use it in GitHub Desktop.
Save bbarrows/2990205 to your computer and use it in GitHub Desktop.
LiveThread
import threading, logging, traceback
import sys, time
class LiveThread(threading.Thread):
def __init__(self, function_to_quit_the_process, thread_wait_in_seconds=1,
clean_up_wait_in_seconds=1, group=None, target=None, name="LiveThread",
args=(), kwargs={}, verbose=None, clean_up=None):
threading.Thread.__init__(self, group=group, target=target, name=name,
args=args, kwargs=kwargs, verbose=verbose)
self._function_to_quit_the_process = function_to_quit_the_process
self._should_stop_event = threading.Event()
self._done_running_event = threading.Event()
self._thread_wait_in_seconds = thread_wait_in_seconds
self._clean_up_wait_in_seconds = clean_up_wait_in_seconds
self._thread_name = name
self._target = target
self._args = args
self._kwargs = kwargs
self._clean_up_method = clean_up
self._has_ran = False
self._has_joined = False
def clean_up(self):
if self._clean_up_method:
self._clean_up_method()
def work_to_do(self):
if self._target:
self._target(*self._args, **self._kwargs)
def run(self):
if self._has_ran:
raise RuntimeError("This thread has already been started once.")
self._has_ran = True
try:
while not self._should_stop_event.isSet():
self._should_stop_event.wait(self._thread_wait_in_seconds)
self.work_to_do()
#self._threading_object.run()
self._done_running_event.set()
#this except could just raise or be removed too..
except (KeyboardInterrupt, SystemExit):
traceback_s = ''.join(traceback.format_exception(*sys.exc_info()))
logging.error("The thread: %s has cause a system exit our keyboard exit." % self._thread_name)
logging.error(traceback_s)
self._function_to_quit_the_process()
except:
traceback_s = ''.join(traceback.format_exception(*sys.exc_info()))
logging.error("The thread: %s has died with the following exception:" % self._thread_name)
logging.error(traceback_s)
self._function_to_quit_the_process()
finally:
# Avoid a refcycle if the thread is running a function with
# an argument that has a member that points to the thread.
# This finally is from the threading.Thread.run method
del self._target, self._args, self._kwargs
def join(self, timeout=None):
if not self._has_joined:
self._should_stop_event.set()
self._done_running_event.wait(self._clean_up_wait_in_seconds)
#Here I have given the run command 1 second to shutdown. I can
#do thread specific shutdown work if I want..
#self._threading_object.clean_up()
self.clean_up()
self._has_joined = True
threading.Thread.join(self, timeout)
#Something you want to thread has two required methods:
#run and clean_up
#run is the code looped over
#clean_up is ran after the run code has finished after the thread has been
# told to stop or until some time has elapsed
class SomeAutoUpgraderThing(LiveThread):
def __init__(self, count_til, count_by, function_to_quit_the_process):
LiveThread.__init__(self, function_to_quit_the_process, name="BradsThread")
self.count_til = count_til
self.count_by = count_by
self.current_count = 0
def clean_up(self):
print "Cleanin up my really important thread"
def work_to_do(self):
if self.current_count > self.count_til:
self.current_count = 0
else:
self.current_count = self.current_count + self.count_by
print "I am on %s" % str(self.current_count)
#This sleep emulates the thread doing some work
time.sleep(2)
def tear_down_proc():
print "Im dying gracefully!"
#2 is just a random time here. Just showing
#that we should specifiy a max amoutn of time
#we give the cleanup method to run
a_thing_to_thread.join(2)
another_thread.join(2)
sys.exit(0)
def my_exception_hook(exctype, value, tb):
try:
logging.error("This is the main threads exception handler:")
traceback_s = ''.join(traceback.format_exception(exctype, value, tb))
logging.error(traceback_s)
tear_down_proc()
except Exception, e:
logging.info("An error occurred in the custom traceback handling.")
logging.info(e)
sys.excepthook = my_exception_hook
a_thing_to_thread = SomeAutoUpgraderThing(100, 2, tear_down_proc)
a_thing_to_thread.start()
#The next 10 or so lines show the use of the target method of threading
#Versus extending a LiveThread
def this_is_another_important_thread():
print "Doooing lots of work!!"
time.sleep(1)
def another_clean_up_method():
print "Cleaning up the other thread!!!"
another_thread = LiveThread(tear_down_proc, clean_up=another_clean_up_method, target=this_is_another_important_thread, name="This is another thread")
another_thread.start()
time.sleep(3)
another_thread.join()
#This sleep emulates the process's other threads doign some work while the thread is off working
time.sleep(8)
tear_down_proc()
time.sleep(1)
print "Shoudlnt get here"
@bebrws
Copy link

bebrws commented Jul 1, 2019

I reported this as Abuse if some admin happens to read this comment.

I lost access to my account when I lost my phone, number, 2FA, etc.. and had to create a new account. Still fighting to get my gmail back but I don't think that will happen.

Anyway I was hoping to take down this Gist, I don't want it up anymore for some personal reasons but I can't take it down anymore without my account. Please consider doing so.

Thanks
Brad

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment