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