zeroinstall/support/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: they introduce race conditions,
  12 often lead to many subtle bugs, and they require lots of resources (you
  13 probably wouldn't want 10,000 threads running at once). In particular, two
  14 threads can run at exactly the same time (perhaps on different processors), so
  15 you have to be really careful that they don't both try to update the same
  16 variables at the same time. This requires lots of messy locking, which is hard
  17 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), so they are not appropriate for long-running jobs.
  30
  31 Tasks use Python's generator API to provide a more pleasant interface to
  32 callbacks. See the Task class (below) for more information.
  33 """
  34
  35 # Copyright (C) 2009, Thomas Leonard
  36 # See the README file for details, or visit http://0install.net.
  37
  38 from zeroinstall import _
  39 import sys
  40 from logging import info, warn, debug
  41 import gobject
  42
  43 # The list of Blockers whose event has happened, in the order they were
  44 # triggered
  45 _run_queue = []
  46
  47 def check(blockers, reporter = None):
  48         """See if any of the blockers have pending exceptions.
  49         @param reporter: invoke this function on each error
  50         If reporter is None, raise the first and log the rest."""
  51         ex = None
  52         if isinstance(blockers, Blocker):
  53                 blockers = (blockers,)
  54         for b in blockers:
  55                 if b.exception:
  56                         b.exception_read = True
  57                         if reporter:
  58                                 try:
  59                                         reporter(*b.exception)
  60                                 except:
  61                                         warn("Failure reporting error! Error was: %s", repr(b.exception[0]))
  62                                         raise
  63                         elif ex is None:
  64                                 ex = b.exception
  65                         else:
  66                                 warn(_("Multiple exceptions waiting; skipping %s"), b.exception[0])
  67         if ex:
  68                 raise ex[0], None, ex[1]
  69
  70 class Blocker:
  71         """A Blocker object starts life with 'happened = False'. Tasks can
  72         ask to be suspended until 'happened = True'. The value is changed
  73         by a call to trigger().
  74
  75         Example:
  76
  77         >>> kettle_boiled = tasks.Blocker()
  78         >>> def make_tea():
  79                 print "Get cup"
  80                 print "Add tea leaves"
  81                 yield kettle_boiled
  82                 print "Pour water into cup"
  83                 print "Brew..."
  84                 yield tasks.TimeoutBlocker(120, "Brewing")
  85                 print "Add milk"
  86                 print "Ready!"
  87         >>> tasks.Task(make_tea())
  88
  89         Then elsewhere, later::
  90
  91                 print "Kettle boiled!"
  92                 kettle_boiled.trigger()
  93
  94         You can also yield a list of Blockers. Your function will resume
  95         after any one of them is triggered. Use blocker.happened to
  96         find out which one(s). Yielding a Blocker that has already
  97         happened is the same as yielding None (gives any other Tasks a
  98         chance to run, and then continues).
  99         """
 100
 101         exception = None
 102
 103         def __init__(self, name):
 104                 self.happened = False           # False until event triggered
 105                 self._zero_lib_tasks = set()    # Tasks waiting on this blocker
 106                 self.name = name
 107
 108         def trigger(self, exception = None):
 109                 """The event has happened. Note that this cannot be undone;
 110                 instead, create a new Blocker to handle the next occurance
 111                 of the event.
 112                 @param exception: exception to raise in waiting tasks
 113                 @type exception: (Exception, traceback)"""
 114                 if self.happened: return        # Already triggered
 115                 self.happened = True
 116                 self.exception = exception
 117                 self.exception_read = False
 118                 #assert self not in _run_queue  # Slow
 119                 if not _run_queue:
 120                         _schedule()
 121                 _run_queue.append(self)
 122
 123                 if exception:
 124                         assert isinstance(exception, tuple), exception
 125                         if not self._zero_lib_tasks:
 126                                 info(_("Exception from '%s', but nothing is waiting for it"), self)
 127                         import traceback
 128                         debug(''.join(traceback.format_exception(*sys.exc_info())))
 129
 130         def __del__(self):
 131                 if self.exception and not self.exception_read:
 132                         warn(_("Blocker %(blocker)s garbage collected without having it's exception read: %(exception)s"), {'blocker': self, 'exception': self.exception})
 133
 134         def add_task(self, task):
 135                 """Called by the schedular when a Task yields this
 136                 Blocker. If you override this method, be sure to still
 137                 call this method with Blocker.add_task(self)!"""
 138                 assert task not in self._zero_lib_tasks, "Blocking on a single task twice: %s (%s)" % (task, self)
 139                 self._zero_lib_tasks.add(task)
 140
 141         def remove_task(self, task):
 142                 """Called by the schedular when a Task that was waiting for
 143                 this blocker is resumed."""
 144                 self._zero_lib_tasks.remove(task)
 145
 146         def __repr__(self):
 147                 return "<Blocker:%s>" % self
 148
 149         def __str__(self):
 150                 return self.name
 151
 152 class IdleBlocker(Blocker):
 153         """An IdleBlocker blocks until a task starts waiting on it, then
 154         immediately triggers. An instance of this class is used internally
 155         when a Task yields None."""
 156         def add_task(self, task):
 157                 """Also calls trigger."""
 158                 Blocker.add_task(self, task)
 159                 self.trigger()
 160
 161 class TimeoutBlocker(Blocker):
 162         """Triggers after a set number of seconds."""
 163         def __init__(self, timeout, name):
 164                 """Trigger after 'timeout' seconds (may be a fraction)."""
 165                 Blocker.__init__(self, name)
 166                 gobject.timeout_add(int(timeout * 1000), self._timeout)
 167
 168         def _timeout(self):
 169                 self.trigger()
 170
 171 def _io_callback(src, cond, blocker):
 172         blocker.trigger()
 173         return False
 174
 175 class InputBlocker(Blocker):
 176         """Triggers when os.read(stream) would not block."""
 177         _tag = None
 178         _stream = None
 179         def __init__(self, stream, name):
 180                 Blocker.__init__(self, name)
 181                 self._stream = stream
 182
 183         def add_task(self, task):
 184                 Blocker.add_task(self, task)
 185                 if self._tag is None:
 186                         self._tag = gobject.io_add_watch(self._stream, gobject.IO_IN | gobject.IO_HUP,
 187                                 _io_callback, self)
 188
 189         def remove_task(self, task):
 190                 Blocker.remove_task(self, task)
 191                 if not self._zero_lib_tasks:
 192                         gobject.source_remove(self._tag)
 193                         self._tag = None
 194
 195 class OutputBlocker(Blocker):
 196         """Triggers when os.write(stream) would not block."""
 197         _tag = None
 198         _stream = None
 199         def __init__(self, stream, name):
 200                 Blocker.__init__(self, name)
 201                 self._stream = stream
 202
 203         def add_task(self, task):
 204                 Blocker.add_task(self, task)
 205                 if self._tag is None:
 206                         self._tag = gobject.io_add_watch(self._stream, gobject.IO_OUT | gobject.IO_HUP,
 207                                 _io_callback, self)
 208
 209         def remove_task(self, task):
 210                 Blocker.remove_task(self, task)
 211                 if not self._zero_lib_tasks:
 212                         gobject.source_remove(self._tag)
 213                         self._tag = None
 214
 215 _idle_blocker = IdleBlocker("(idle)")
 216
 217 class Task:
 218         """Create a new Task when you have some long running function to
 219         run in the background, but which needs to do work in 'chunks'.
 220         Example:
 221
 222         >>> from zeroinstall import tasks
 223         >>> def my_task(start):
 224                 for x in range(start, start + 5):
 225                         print "x =", x
 226                         yield None
 227
 228         >>> tasks.Task(my_task(0))
 229         >>> tasks.Task(my_task(10))
 230         >>> mainloop()
 231
 232         Yielding None gives up control of the processor to another Task,
 233         causing the sequence printed to be interleaved. You can also yield a
 234         Blocker (or a list of Blockers) if you want to wait for some
 235         particular event before resuming (see the Blocker class for details).
 236         """
 237
 238         def __init__(self, iterator, name):
 239                 """Call iterator.next() from a glib idle function. This function
 240                 can yield Blocker() objects to suspend processing while waiting
 241                 for events. name is used only for debugging."""
 242                 assert iterator.next, "Object passed is not an iterator!"
 243                 self.iterator = iterator
 244                 self.finished = Blocker(name)
 245                 # Block new task on the idle handler...
 246                 _idle_blocker.add_task(self)
 247                 self._zero_blockers = (_idle_blocker,)
 248                 info(_("Scheduling new task: %s"), self)
 249
 250         def _resume(self):
 251                 # Remove from our blockers' queues
 252                 for blocker in self._zero_blockers:
 253                         blocker.remove_task(self)
 254                 # Resume the task
 255                 try:
 256                         new_blockers = self.iterator.next()
 257                 except StopIteration:
 258                         # Task ended
 259                         self.finished.trigger()
 260                         return
 261                 except SystemExit:
 262                         raise
 263                 except (Exception, KeyboardInterrupt) as ex:
 264                         # Task crashed
 265                         info(_("Exception from '%(name)s': %(exception)s"), {'name': self.finished.name, 'exception': ex})
 266                         #import traceback
 267                         #debug(''.join(traceback.format_exception(*sys.exc_info())))
 268                         tb = sys.exc_info()[2]
 269                         self.finished.trigger(exception = (ex, tb))
 270                         return
 271                 if new_blockers is None:
 272                         # Just give up control briefly
 273                         new_blockers = (_idle_blocker,)
 274                 else:
 275                         if isinstance(new_blockers, Blocker):
 276                                 # Wrap a single yielded blocker into a list
 277                                 new_blockers = (new_blockers,)
 278                         # Are we blocking on something that already happened?
 279                         for blocker in new_blockers:
 280                                 assert hasattr(blocker, 'happened'), "Not a Blocker: %s from %s" % (blocker, self)
 281                                 if blocker.happened:
 282                                         new_blockers = (_idle_blocker,)
 283                                         info(_("Task '%(task)s' waiting on ready blocker %(blocker)s!"), {'task': self, 'blocker': blocker})
 284                                         break
 285                         else:
 286                                 info(_("Task '%(task)s' stopping and waiting for '%(new_blockers)s'"), {'task': self, 'new_blockers': new_blockers})
 287                 # Add to new blockers' queues
 288                 for blocker in new_blockers:
 289                         blocker.add_task(self)
 290                 self._zero_blockers = new_blockers
 291
 292         def __repr__(self):
 293                 return "Task(%s)" % self.finished.name
 294
 295         def __str__(self):
 296                 return self.finished.name
 297
 298 # Must append to _run_queue right after calling this!
 299 def _schedule():
 300         assert not _run_queue
 301         gobject.idle_add(_handle_run_queue)
 302
 303 def _handle_run_queue():
 304         global _idle_blocker
 305         assert _run_queue
 306
 307         next = _run_queue[0]
 308         assert next.happened
 309
 310         if next is _idle_blocker:
 311                 # Since this blocker will never run again, create a
 312                 # new one for future idling.
 313                 _idle_blocker = IdleBlocker("(idle)")
 314         elif next._zero_lib_tasks:
 315                 info(_("Running %(task)s due to triggering of '%(next)s'"), {'task': next._zero_lib_tasks, 'next': next})
 316         else:
 317                 info(_("Running %s"), next)
 318
 319         tasks = frozenset(next._zero_lib_tasks)
 320         if tasks:
 321                 next.noticed = True
 322
 323         for task in tasks:
 324                 # Run 'task'.
 325                 task._resume()
 326
 327         del _run_queue[0]
 328
 329         if _run_queue:
 330                 return True
 331         return False
 332
 333 def named_async(name):
 334         """Decorator that turns a generator function into a function that runs the
 335         generator as a Task and returns the Task's finished blocker.
 336         @param name: the name for the Task"""
 337         def deco(fn):
 338                 def run(*args, **kwargs):
 339                         return Task(fn(*args, **kwargs), name).finished
 340                 run.__name__ = fn.__name__
 341                 return run
 342         return deco
 343
 344 def async(fn):
 345         """Decorator that turns a generator function into a function that runs the
 346         generator as a Task and returns the Task's finished blocker."""
 347         def run(*args, **kwargs):
 348                 return Task(fn(*args, **kwargs), fn.__name__).finished
 349         run.__name__ = fn.__name__
 350         return run
 351
 352 def wait_for_blocker(blocker):
 353         """Run a recursive mainloop until blocker is triggered.
 354         @param blocker: event to wait on
 355         @type blocker: L{Blocker}
 356         @since: 0.53
 357         """
 358         assert wait_for_blocker.loop is None    # Avoid recursion
 359
 360         if not blocker.happened:
 361                 def quitter():
 362                         yield blocker
 363                         wait_for_blocker.loop.quit()
 364                 Task(quitter(), "quitter")
 365
 366                 wait_for_blocker.loop = gobject.MainLoop(gobject.main_context_default())
 367                 try:
 368                         debug(_("Entering mainloop, waiting for %s"), blocker)
 369                         wait_for_blocker.loop.run()
 370                 finally:
 371                         wait_for_blocker.loop = None
 372
 373                 assert blocker.happened, "Someone quit the main loop!"
 374
 375         check(blocker)
 376 wait_for_blocker.loop = None