code.delx.au - bg-scripts/blob - asyncsched.py

   1 # Copyright 2008 James Bunton <jamesbunton@fastmail.fm>
   2 # Licensed for distribution under the GPL version 2, check COPYING for details
   3 # asyncore.loop() with delayed function calls
   4
   5 import asyncore, select
   6 import heapq
   7 import signal
   8 import time
   9
  10 tasks = []
  11 running = False
  12
  13 class Task(object):
  14     def __init__(self, delay, func, args=[], kwargs={}):
  15         self.time = time.time() + delay
  16         self.func = lambda: func(*args, **kwargs)
  17
  18     def __lt__(self, other):
  19         return self.time < other.time
  20
  21     def __eq__(self, other):
  22         return self.time == other.time
  23
  24     def __call__(self):
  25         f = self.func
  26         self.func = None
  27         if f:
  28             return f()
  29
  30     def cancel(self):
  31         assert self.func is not None
  32         self.func = None
  33
  34 def schedule(delay, func, args=[], kwargs={}):
  35     task = Task(delay, func, args, kwargs)
  36     heapq.heappush(tasks, task)
  37     return task
  38
  39 def loop(timeout=30.0, use_poll=False):
  40     global running
  41     running = True
  42     oldhandler = signal.signal(signal.SIGTERM, exit)
  43
  44     if use_poll:
  45         if hasattr(select, 'poll'):
  46             poll_fun = asyncore.poll3
  47         else:
  48             poll_fun = asyncore.poll2
  49     else:
  50         poll_fun = asyncore.poll
  51
  52     while running:
  53         now = time.time()
  54         while tasks and tasks[0].time < now:
  55             task = heapq.heappop(tasks)
  56             task()
  57
  58         t = timeout
  59         if tasks:
  60             t = max(min(t, tasks[0].time - now), 0)
  61
  62         poll_fun(timeout=t)
  63
  64     signal.signal(signal.SIGTERM, oldhandler)
  65
  66 def exit(*args):
  67     global running
  68     running = False
  69
  70 __all__ = ("schedule", "loop", "exit")
  71