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 class IdleBlocker(Blocker):
 100         """An IdleBlocker blocks until a task starts waiting on it, then
 101         immediately triggers. An instance of this class is used internally
 102         when a Task yields None."""
 103         def add_task(self, task):
 104                 """Also calls trigger."""
 105                 Blocker.add_task(self, task)
 106                 self.trigger()
 107
 108 class TimeoutBlocker(Blocker):
 109         """Triggers after a set number of seconds. rox.toplevel_ref/unref
 110         are called to prevent the app quitting while a TimeoutBlocker is
 111         running."""
 112         def __init__(self, timeout):
 113                 """Trigger after 'timeout' seconds (may be a fraction)."""
 114                 Blocker.__init__(self)
 115                 rox.toplevel_ref()
 116                 g.timeout_add(long(timeout * 1000), self._timeout)
 117
 118         def _timeout(self):
 119                 rox.toplevel_unref()
 120                 self.trigger()
 121
 122 _idle_blocker = IdleBlocker()
 123
 124 class Task:
 125         """Create a new Task when you have some long running function to
 126         run in the background, but which needs to do work in 'chunks'.
 127         Example (the first line is needed to enable the 'yield' keyword in
 128         python 2.2):
 129
 130         from __future__ import generators
 131         from rox import tasks
 132         def my_task(start):
 133                 for x in range(start, start + 5):
 134                         print "x =", x
 135                         yield None
 136
 137         tasks.Task(my_task(0))
 138         tasks.Task(my_task(10))
 139
 140         rox.mainloop()
 141
 142         Yielding None gives up control of the processor to another Task,
 143         causing the sequence printed to be interleaved. You can also yield a
 144         Blocker (or a list of Blockers) if you want to wait for some
 145         particular event before resuming (see the Blocker class for details).
 146         """
 147
 148         def __init__(self, iterator, name = None):
 149                 """Call iterator.next() from a glib idle function. This function
 150                 can yield Blocker() objects to suspend processing while waiting
 151                 for events. name is used only for debugging."""
 152                 assert iterator.next, "Object passed is not an iterator!"
 153                 self.next = iterator.next
 154                 self.name = name
 155                 # Block new task on the idle handler...
 156                 _idle_blocker.add_task(self)
 157                 self._rox_blockers = (_idle_blocker,)
 158
 159         def _resume(self):
 160                 # Remove from our blockers' queues
 161                 for blocker in self._rox_blockers:
 162                         del blocker._rox_lib_tasks[self]
 163                 # Resume the task
 164                 try:
 165                         new_blockers = self.next()
 166                 except StopIteration:
 167                         # Task ended
 168                         return
 169                 except Exception:
 170                         # Task crashed
 171                         rox.report_exception()
 172                         return
 173                 if new_blockers is None:
 174                         # Just give up control briefly
 175                         new_blockers = (_idle_blocker,)
 176                 else:
 177                         if isinstance(new_blockers, Blocker):
 178                                 # Wrap a single yielded blocker into a list
 179                                 new_blockers = (new_blockers,)
 180                         # Are we blocking on something that already happened?
 181                         for blocker in new_blockers:
 182                                 if blocker.happened:
 183                                         new_blockers = (_idle_blocker,)
 184                                         break
 185                 # Add to new blockers' queues
 186                 for blocker in new_blockers:
 187                         blocker.add_task(self)
 188                 self._rox_blockers = new_blockers
 189
 190         def __repr__(self):
 191                 if self.name is None:
 192                         return "[Task]"
 193                 return "[Task '%s']" % self.name
 194
 195 # Must append to _run_queue right after calling this!
 196 def _schedule():
 197         assert not _run_queue
 198         rox.toplevel_ref()
 199         g.idle_add(_handle_run_queue)
 200
 201 def _handle_run_queue():
 202         global _idle_blocker
 203         assert _run_queue
 204
 205         next = _run_queue[0]
 206         assert next.happened
 207
 208         if next is _idle_blocker:
 209                 # Since this blocker will never run again, create a
 210                 # new one for future idling.
 211                 _idle_blocker = IdleBlocker()
 212
 213         tasks = next._rox_lib_tasks.keys()
 214         #print "Resume", tasks
 215         for task in tasks:
 216                 # Run 'task'.
 217                 task._resume()
 218
 219         del _run_queue[0]
 220
 221         if _run_queue:
 222                 return True
 223         rox.toplevel_unref()
 224         return False