Skip to content

Instantly share code, notes, and snippets.

@beugley
Last active March 14, 2022 00:13
Show Gist options
  • Save beugley/e37dd3e36fd8654553df to your computer and use it in GitHub Desktop.
Save beugley/e37dd3e36fd8654553df to your computer and use it in GitHub Desktop.
Python: implement a stoppable thread
#!/usr/bin/env python
import sys
import logging
import threading
from time import sleep
class StoppableThread(threading.Thread):
"""
Implements a thread that can be stopped.
"""
def __init__(self, name, target, args=()):
super(StoppableThread, self).__init__(name=name, target=target, args=args)
self._status = 'running'
def stop_me(self):
if (self._status == 'running'):
self._status = 'stopping'
def stopped(self):
self._status = 'stopped'
def is_running(self):
return (self._status == 'running')
def is_stopping(self):
return (self._status == 'stopping')
def is_stopped(self):
return (self._status == 'stopped')
dThreads = {}
def ThreadFunc(id, arg1, arg2):
thread = dThreads[id]
while (thread.is_running()):
logging.info('Tick...')
sleep(4)
thread.stopped()
def StartThread(id, arg1, arg2):
"""
Starts a thread and adds an entry to the global dThreads dictionary.
"""
logging.info('Starting %s' % id)
dThreads[id] = StoppableThread(name=id, target=ThreadFunc,
args=(id,arg1,arg2))
dThreads[id].start()
def StopThread(id):
"""
Stops a thread and removes its entry from the global dThreads dictionary.
"""
thread = dThreads[id]
if (thread.is_running()):
logging.info('Stopping %s' % id)
thread.stop_me()
thread.join()
logging.info('Stopped %s' % id)
del dThreads[id]
if __name__ == "__main__":
logging.basicConfig(format='%(threadName)s:%(asctime)s:%(levelname)s:%(message)s',
stream=sys.stdout, level=logging.INFO)
StartThread('one', 'argval1', 'argval2')
sleep(2)
StartThread('two', 'argval3', 'argval4')
sleep(10)
StopThread('one')
sleep(10)
StopThread('two')
@diazagasatya
Copy link

love this thankyou!

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