Skip to content

Instantly share code, notes, and snippets.

@louisswarren

louisswarren/task.py

Last active Mar 20, 2020
Embed
What would you like to do?
Subpar
import heapq
class Task:
def __init__(self, time):
self.time = time
def __lt__(self, other):
return self.time < other.time
def __repr__(self):
return f'Task({self.time})'
def spawned(self):
'''Yield further tasks after run completes.'''
if False:
yield
def run(self):
'''Called at the scheduled time.'''
return
class PrintTask(Task):
def __init__(self, time, msg):
self.time = time
self.msg = msg
def __repr__(self):
return 'PrintTask({}, {})'.format(self.time, repr(self.msg))
def run(self):
print(self.msg)
class RecurrentTask(Task):
def __repr__(self):
return f'RecurrentTask({self.time})'
def spawned(self):
yield RecurrentTask(self.time + 1)
def run_tasks(tasks):
pool = list(tasks)
heapq.heapify(pool)
while pool:
task = heapq.heappop(pool)
task.run()
for new_task in task.spawned():
heapq.heappush(pool, new_task)
yield task
if __name__ == '__main__':
# Schedule some tasks
tasks = [PrintTask(5, 'It is now time=5'),
RecurrentTask(1),
PrintTask(10, 'Final message'),
]
print("Starting with tasks:")
for task in tasks:
print('\t' + str(task))
print()
for i, x in enumerate(run_tasks(tasks)):
print("Ended:", x)
if i > 20:
print("... and so on")
break
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.