python/rox/tasks.py

   1 """The tasks module provides a simple light-weight alternative to threads.
   2
   3 When you have a long-running job you will want to run it in the background,
   4 while the user does other things. There are four ways to do this:
   5
   6 - Use a new thread for each task.
   7 - Use callbacks from an idle handler.
   8 - Use a recursive mainloop.
   9 - Use this module.
  10
  11 Using threads causes a number of problems. Some builds of pygtk/python don't
  12 support them, they introduce race conditions, often lead to many subtle
  13 bugs, and they require lots of resources (you probably wouldn't want 10,000
  14 threads running at once). In particular, two threads can run at exactly the
  15 same time (perhaps on different processors), so you have to be really careful
  16 that they don't both try to update the same variable at the same time. This
  17 requires lots of messy locking, which is hard to get right.
  18
  19 Callbacks work within a single thread. For example, you open a dialog box and
  20 then tell the system to call one function if it's closed, and another if the
  21 user clicks OK, etc. The function that opened the box then returns, and the
  22 system calls one of the given callback functions later. Callbacks only
  23 execute one at a time, so you don't have to worry about race conditions.
  24 However, they are often very awkward to program with, because you have to
  25 save state somewhere and then pass it to the functions when they're called.
  26
  27 A recursive mainloop only works with nested tasks (you can create a
  28 sub-task, but the main task can't continue until the sub-task has
  29 finished). We use these for, eg, rox.alert() boxes since you don't
  30 normally want to do anything else until the box is closed, but it is not
  31 appropriate for long-running jobs.
  32
  33 Tasks use python's generator API to provide a more pleasant interface to
  34 callbacks. See the Task class (below) for more information.
  35 """
  36
  37 from __future__ import generators
  38
  39 import rox
  40 from rox import g, _
  41
  42 # The list of Blockers whose event has happened, in the order they were
  43 # triggered
  44 _run_queue = []
  45
  46 class Blocker:
  47         """A Blocker object starts life with 'happened = False'. Tasks can
  48         ask to be suspended until 'happened = True'. The value is changed
  49         by a call to trigger().
  50
  51         Example:
  52
  53         kettle_boiled = tasks.Blocker()
  54
  55         def make_tea():
  56                 print "Get cup"
  57                 print "Add tea leaves"
  58                 yield kettle_boiled;
  59                 print "Pour water into cup"
  60                 print "Brew..."
  61                 yield tasks.TimeoutBlocker(120)
  62                 print "Add milk"
  63                 print "Ready!"
  64
  65         tasks.Task(make_tea())
  66
  67         # elsewhere, later...
  68                 print "Kettle boiled!"
  69                 kettle_boiled.trigger()
  70
  71         You can also yield a list of Blockers. Your function will resume
  72         after any one of them is triggered. Use blocker.happened to
  73         find out which one(s). Yielding a Blocker that has already
  74         happened is the same as yielding None (gives any other Tasks a
  75         chance to run, and then continues).
  76         """
  77
  78         def __init__(self):
  79                 self.happened = False   # False until event triggered
  80                 self._rox_lib_tasks = {}        # Tasks waiting on this blocker
  81
  82         def trigger(self):
  83                 """The event has happened. Note that this cannot be undone;
  84                 instead, create a new Blocker to handle the next occurance
  85                 of the event."""
  86                 if self.happened: return        # Already triggered
  87                 self.happened = True
  88                 #assert self not in _run_queue  # XXX: Slow
  89                 if not _run_queue:
  90                         _schedule()
  91                 _run_queue.append(self)
  92
  93         def add_task(self, task):
  94                 """Called by the schedular when a Task yields this
  95                 Blocker. If you override this method, be sure to still
  96                 call this method with Blocker.add_task(self)!"""
  97                 self._rox_lib_tasks[task] = True
  98
  99         def remove_task(self, task):
 100                 """Called by the schedular when a Task that was waiting for
 101                 this blocker is resumed."""
 102                 del self._rox_lib_tasks[task]
 103
 104 class IdleBlocker(Blocker):
 105         """An IdleBlocker blocks until a task starts waiting on it, then
 106         immediately triggers. An instance of this class is used internally
 107         when a Task yields None."""
 108         def add_task(self, task):
 109                 """Also calls trigger."""
 110                 Blocker.add_task(self, task)
 111                 self.trigger()
 112
 113 class TimeoutBlocker(Blocker):
 114         """Triggers after a set number of seconds. rox.toplevel_ref/unref
 115         are called to prevent the app quitting while a TimeoutBlocker is
 116         running."""
 117         def __init__(self, timeout):
 118                 """Trigger after 'timeout' seconds (may be a fraction)."""
 119                 Blocker.__init__(self)
 120                 rox.toplevel_ref()
 121                 g.timeout_add(long(timeout * 1000), self._timeout)
 122
 123         def _timeout(self):
 124                 rox.toplevel_unref()
 125                 self.trigger()
 126
 127 class InputBlocker(Blocker):
 128         """Triggers when os.read(stream) would not block."""
 129         _tag = None
 130         _stream = None
 131         def __init__(self, stream):
 132                 Blocker.__init__(self)
 133                 self._stream = stream
 134
 135         def add_task(self, task):
 136                 Blocker.add_task(self, task)
 137                 if self._tag is None:
 138                         self._tag = g.input_add(self._stream, g.gdk.INPUT_READ,
 139                                 lambda src, cond: self.trigger())
 140
 141         def remove_task(self, task):
 142                 Blocker.remove_task(self, task)
 143                 if not self._rox_lib_tasks:
 144                         g.input_remove(self._tag)
 145                         self._tag = None
 146
 147 _idle_blocker = IdleBlocker()
 148
 149 class Task:
 150         """Create a new Task when you have some long running function to
 151         run in the background, but which needs to do work in 'chunks'.
 152         Example (the first line is needed to enable the 'yield' keyword in
 153         python 2.2):
 154
 155         from __future__ import generators
 156         from rox import tasks
 157         def my_task(start):
 158                 for x in range(start, start + 5):
 159                         print "x =", x
 160                         yield None
 161
 162         tasks.Task(my_task(0))
 163         tasks.Task(my_task(10))
 164
 165         rox.mainloop()
 166
 167         Yielding None gives up control of the processor to another Task,
 168         causing the sequence printed to be interleaved. You can also yield a
 169         Blocker (or a list of Blockers) if you want to wait for some
 170         particular event before resuming (see the Blocker class for details).
 171         """
 172
 173         def __init__(self, iterator, name = None):
 174                 """Call iterator.next() from a glib idle function. This function
 175                 can yield Blocker() objects to suspend processing while waiting
 176                 for events. name is used only for debugging."""
 177                 assert iterator.next, "Object passed is not an iterator!"
 178                 self.next = iterator.next
 179                 self.name = name
 180                 # Block new task on the idle handler...
 181                 _idle_blocker.add_task(self)
 182                 self._rox_blockers = (_idle_blocker,)
 183
 184         def _resume(self):
 185                 # Remove from our blockers' queues
 186                 for blocker in self._rox_blockers:
 187                         blocker.remove_task(self)
 188                 # Resume the task
 189                 try:
 190                         new_blockers = self.next()
 191                 except StopIteration:
 192                         # Task ended
 193                         return
 194                 except Exception:
 195                         # Task crashed
 196                         rox.report_exception()
 197                         return
 198                 if new_blockers is None:
 199                         # Just give up control briefly
 200                         new_blockers = (_idle_blocker,)
 201                 else:
 202                         if isinstance(new_blockers, Blocker):
 203                                 # Wrap a single yielded blocker into a list
 204                                 new_blockers = (new_blockers,)
 205                         # Are we blocking on something that already happened?
 206                         for blocker in new_blockers:
 207                                 if blocker.happened:
 208                                         new_blockers = (_idle_blocker,)
 209                                         break
 210                 # Add to new blockers' queues
 211                 for blocker in new_blockers:
 212                         blocker.add_task(self)
 213                 self._rox_blockers = new_blockers
 214
 215         def __repr__(self):
 216                 if self.name is None:
 217                         return "[Task]"
 218                 return "[Task '%s']" % self.name
 219
 220 # Must append to _run_queue right after calling this!
 221 def _schedule():
 222         assert not _run_queue
 223         rox.toplevel_ref()
 224         g.idle_add(_handle_run_queue)
 225
 226 def _handle_run_queue():
 227         global _idle_blocker
 228         assert _run_queue
 229
 230         next = _run_queue[0]
 231         assert next.happened
 232
 233         if next is _idle_blocker:
 234                 # Since this blocker will never run again, create a
 235                 # new one for future idling.
 236                 _idle_blocker = IdleBlocker()
 237
 238         tasks = next._rox_lib_tasks.keys()
 239         #print "Resume", tasks
 240         for task in tasks:
 241                 # Run 'task'.
 242                 task._resume()
 243
 244         del _run_queue[0]
 245
 246         if _run_queue:
 247                 return True
 248         rox.toplevel_unref()
 249         return False