A simple multi-threaded scheduler that enables tasks to be run at specified intervals.
Python, 79 lines
Download
Copy to clipboard1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | import time
import threading
class Task( threading.Thread ):
def __init__( self, action, loopdelay, initdelay ):
self._action = action
self._loopdelay = loopdelay
self._initdelay = initdelay
self._running = 1
threading.Thread.__init__( self )
def __repr__( self ):
return '%s %s %s' % (
self._action, self._loopdelay, self._initdelay )
def run( self ):
if self._initdelay:
time.sleep( self._initdelay )
self._runtime = time.time()
while self._running:
start = time.time()
self._action()
self._runtime += self._loopdelay
time.sleep( self._runtime - start )
def stop( self ):
self._running = 0
class Scheduler:
def __init__( self ):
self._tasks = []
def __repr__( self ):
rep = ''
for task in self._tasks:
rep += '%s\n' % `task`
return rep
def AddTask( self, action, loopdelay, initdelay = 0 ):
task = Task( action, loopdelay, initdelay )
self._tasks.append( task )
def StartAllTasks( self ):
for task in self._tasks:
task.start()
def StopAllTasks( self ):
for task in self._tasks:
print 'Stopping task', task
task.stop()
task.join()
print 'Stopped'
if __name__ == '__main__':
def timestamp( s ):
print '%.2f : %s' % ( time.time(), s )
def Task1():
timestamp( 'Task1' )
def Task2():
timestamp( '\tTask2' )
def Task3():
timestamp( '\t\tTask3' )
s = Scheduler()
# task loopdelay initdelay
# ---------------------------------------
s.AddTask( Task1, 1.0, 0 )
s.AddTask( Task2, 0.5, 0.25 )
s.AddTask( Task3, 0.1, 0.05 )
print s
s.StartAllTasks()
raw_input()
s.StopAllTasks()
|
Jumps through hoops to get all the threads to stop. Pressing ctrl-C on a Windows platform leaves threads running. Is this required on the Unix platform?
Tags: threads
Fix for OSes with jittery timer. On MacOS X this didn't work, because a jittery timer leads to a negative (self._runtime - start) which is in turn an invalid argument to time.sleep(). Changing the last line in Thread.run() to
fixes this.
Superceded by the sched module? I gather that this recipy is superceded by the sched module from the standard library?