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

# Source Code for Module Products.ZenUtils.NJobs

     1  ###########################################################################
     2  #
     3  # This program is part of Zenoss Core, an open source monitoring platform.
     4  # Copyright (C) 2007, Zenoss Inc.
     5  #
     6  # This program is free software; you can redistribute it and/or modify it
     7  # under the terms of the GNU General Public License version 2 or (at your
     8  # option) any later version as published by the Free Software Foundation.
     9  #
    10  # For complete information please visit: http://www.zenoss.com/oss/
    11  #
    12  ###########################################################################
    13  #! /usr/bin/env python
    14
    15  __doc__='''NJobs
    16
    17  Run a list of jobs in parallel, limited to N at a time.
    18
    19  $Id$
    20  '''
    21
    22  __version__ = "$Revision$"[11:-2]
    23
    24  from twisted.internet import reactor, defer
    25
    26
    27 -class NJobs:
    28      "Run a list of jobs in parallel, limited to N at a time."
    29
    30 -    def __init__(self, max, callable, data):
    31          self.defer = defer.Deferred()
    32          self.results = []
    33          self.max = max
    34          self.callable = callable
    35          self.workQueue = data
    36          self.running = 0
    37
    38 -    def start(self):
    39          self._runSome()
    40          return self.defer
    41
    42 -    def status(self):
    43          return self.running, len(self.workQueue), len(self.results)
    44
    45 -    def _runSome(self):
    46          while self.running < self.max and self.workQueue:
    47              self.running += 1
    48              try:
    49                  d = self.callable(self.workQueue.pop())
    50              except Exception, ex:
    51                  self._finished(ex)
    52              else:
    53                  d.addBoth(self._finished)
    54          if self.running == 0 and not self.workQueue and not self.defer.called:
    55              self.defer.callback(self.results)
    56
    57 -    def _finished(self, result):
    58          self.running -= 1
    59          self.results.append(result)
    60          reactor.callLater(0, self._runSome)
    61