Package Products :: Package ZenUtils :: Module NJobs |
[hide private]
[frames] | no frames]

# Source Code for Module Products.ZenUtils.NJobs

     1  #! /usr/bin/env python
     2  ##############################################################################
     3  #
     4  # Copyright (C) Zenoss, Inc. 2007, all rights reserved.
     5  #
     6  # This content is made available according to terms specified in
     7  # License.zenoss under the directory where your Zenoss product is installed.
     8  #
     9  ##############################################################################
    10
    11
    12  __doc__='''NJobs
    13
    14  Run a list of jobs in parallel, limited to N at a time.
    15
    16  $Id$
    17  '''
    18
    19  __version__ = "$Revision$"[11:-2]
    20
    21  from twisted.internet import reactor, defer
    22
    23
    24 -class NJobs:
    25      "Run a list of jobs in parallel, limited to N at a time."
    26
    27 -    def __init__(self, max, callable, data):
    28          self.defer = defer.Deferred()
    29          self.results = []
    30          self.max = max
    31          self.callable = callable
    32          self.workQueue = data
    33          self.running = 0
    34
    35 -    def start(self):
    36          self._runSome()
    37          return self.defer
    38
    39 -    def status(self):
    40          return self.running, len(self.workQueue), len(self.results)
    41
    42 -    def _runSome(self):
    43          while self.running < self.max and self.workQueue:
    44              self.running += 1
    45              try:
    46                  d = self.callable(self.workQueue.pop())
    47              except Exception, ex:
    48                  self._finished(ex)
    49              else:
    50                  d.addBoth(self._finished)
    51          if self.running == 0 and not self.workQueue and not self.defer.called:
    52              self.defer.callback(self.results)
    53
    54 -    def _finished(self, result):
    55          self.running -= 1
    56          self.results.append(result)
    57          reactor.callLater(0, self._runSome)
    58