src/allmydata/util/limiter.py

   1
   2 from twisted.internet import defer
   3 from foolscap.api import eventually
   4
   5 class ConcurrencyLimiter:
   6     """I implement a basic concurrency limiter. Add work to it in the form of
   7     (callable, args, kwargs) tuples. No more than LIMIT callables will be
   8     outstanding at any one time.
   9     """
  10
  11     def __init__(self, limit=10):
  12         self.limit = limit
  13         self.pending = []
  14         self.active = 0
  15
  16     def __repr__(self):
  17         return "<Limiter with %d/%d/%d>" % (self.active, len(self.pending),
  18                                             self.limit)
  19
  20     def add(self, cb, *args, **kwargs):
  21         d = defer.Deferred()
  22         task = (cb, args, kwargs, d)
  23         self.pending.append(task)
  24         self.maybe_start_task()
  25         return d
  26
  27     def maybe_start_task(self):
  28         if self.active >= self.limit:
  29             return
  30         if not self.pending:
  31             return
  32         (cb, args, kwargs, done_d) = self.pending.pop(0)
  33         self.active += 1
  34         d = defer.maybeDeferred(cb, *args, **kwargs)
  35         d.addBoth(self._done, done_d)
  36
  37     def _done(self, res, done_d):
  38         self.active -= 1
  39         eventually(done_d.callback, res)
  40         eventually(self.maybe_start_task)