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 class OutputBlocker(Blocker):
 148         """Triggers when os.write(stream) would not block."""
 149         _tag = None
 150         _stream = None
 151         def __init__(self, stream):
 152                 Blocker.__init__(self)
 153                 self._stream = stream
 154
 155         def add_task(self, task):
 156                 Blocker.add_task(self, task)
 157                 if self._tag is None:
 158                         INPUT_WRITE = 0x14 # g.gdk.INPUT_WRITE sometimes wrong!!
 159                         self._tag = g.input_add(self._stream, INPUT_WRITE,
 160                                 lambda src, cond: self.trigger())
 161
 162         def remove_task(self, task):
 163                 Blocker.remove_task(self, task)
 164                 if not self._rox_lib_tasks:
 165                         g.input_remove(self._tag)
 166                         self._tag = None
 167
 168 _idle_blocker = IdleBlocker()
 169
 170 class Task:
 171         """Create a new Task when you have some long running function to
 172         run in the background, but which needs to do work in 'chunks'.
 173         Example (the first line is needed to enable the 'yield' keyword in
 174         python 2.2):
 175
 176         from __future__ import generators
 177         from rox import tasks
 178         def my_task(start):
 179                 for x in range(start, start + 5):
 180                         print "x =", x
 181                         yield None
 182
 183         tasks.Task(my_task(0))
 184         tasks.Task(my_task(10))
 185
 186         rox.mainloop()
 187
 188         Yielding None gives up control of the processor to another Task,
 189         causing the sequence printed to be interleaved. You can also yield a
 190         Blocker (or a list of Blockers) if you want to wait for some
 191         particular event before resuming (see the Blocker class for details).
 192         """
 193
 194         def __init__(self, iterator, name = None):
 195                 """Call iterator.next() from a glib idle function. This function
 196                 can yield Blocker() objects to suspend processing while waiting
 197                 for events. name is used only for debugging."""
 198                 assert iterator.next, "Object passed is not an iterator!"
 199                 self.next = iterator.next
 200                 self.name = name
 201                 # Block new task on the idle handler...
 202                 _idle_blocker.add_task(self)
 203                 self._rox_blockers = (_idle_blocker,)
 204
 205         def _resume(self):
 206                 # Remove from our blockers' queues
 207                 for blocker in self._rox_blockers:
 208                         blocker.remove_task(self)
 209                 # Resume the task
 210                 try:
 211                         new_blockers = self.next()
 212                 except StopIteration:
 213                         # Task ended
 214                         return
 215                 except Exception:
 216                         # Task crashed
 217                         rox.report_exception()
 218                         return
 219                 if new_blockers is None:
 220                         # Just give up control briefly
 221                         new_blockers = (_idle_blocker,)
 222                 else:
 223                         if isinstance(new_blockers, Blocker):
 224                                 # Wrap a single yielded blocker into a list
 225                                 new_blockers = (new_blockers,)
 226                         # Are we blocking on something that already happened?
 227                         for blocker in new_blockers:
 228                                 if blocker.happened:
 229                                         new_blockers = (_idle_blocker,)
 230                                         break
 231                 # Add to new blockers' queues
 232                 for blocker in new_blockers:
 233                         blocker.add_task(self)
 234                 self._rox_blockers = new_blockers
 235
 236         def __repr__(self):
 237                 if self.name is None:
 238                         return "[Task]"
 239                 return "[Task '%s']" % self.name
 240
 241 # Must append to _run_queue right after calling this!
 242 def _schedule():
 243         assert not _run_queue
 244         rox.toplevel_ref()
 245         g.idle_add(_handle_run_queue)
 246
 247 def _handle_run_queue():
 248         global _idle_blocker
 249         assert _run_queue
 250
 251         next = _run_queue[0]
 252         assert next.happened
 253
 254         if next is _idle_blocker:
 255                 # Since this blocker will never run again, create a
 256                 # new one for future idling.
 257                 _idle_blocker = IdleBlocker()
 258
 259         tasks = next._rox_lib_tasks.keys()
 260         #print "Resume", tasks
 261         for task in tasks:
 262                 # Run 'task'.
 263                 task._resume()
 264
 265         del _run_queue[0]
 266
 267         if _run_queue:
 268                 return True
 269         rox.toplevel_unref()
 270         return False