p2pool/util/deferral.py

   1 from __future__ import division
   2
   3 import itertools
   4 import random
   5 import sys
   6
   7 from twisted.internet import defer, reactor
   8 from twisted.python import failure, log
   9
  10 def sleep(t):
  11     d = defer.Deferred()
  12     reactor.callLater(t, d.callback, None)
  13     return d
  14
  15 def run_repeatedly(f, *args, **kwargs):
  16     current_dc = [None]
  17     def step():
  18         delay = f(*args, **kwargs)
  19         current_dc[0] = reactor.callLater(delay, step)
  20     step()
  21     def stop():
  22         current_dc[0].cancel()
  23     return stop
  24
  25 class RetrySilentlyException(Exception):
  26     pass
  27
  28 def retry(message='Error:', delay=3, max_retries=None, traceback=True):
  29     '''
  30     @retry('Error getting block:', 1)
  31     @defer.inlineCallbacks
  32     def get_block(hash):
  33         ...
  34     '''
  35
  36     def retry2(func):
  37         @defer.inlineCallbacks
  38         def f(*args, **kwargs):
  39             for i in itertools.count():
  40                 try:
  41                     result = yield func(*args, **kwargs)
  42                 except Exception, e:
  43                     if i == max_retries:
  44                         raise
  45                     if not isinstance(e, RetrySilentlyException):
  46                         if traceback:
  47                             log.err(None, message)
  48                         else:
  49                             print >>sys.stderr, message, e
  50                     yield sleep(delay)
  51                 else:
  52                     defer.returnValue(result)
  53         return f
  54     return retry2
  55
  56 class ReplyMatcher(object):
  57     '''
  58     Converts request/got response interface to deferred interface
  59     '''
  60
  61     def __init__(self, func, timeout=5):
  62         self.func = func
  63         self.timeout = timeout
  64         self.map = {}
  65
  66     def __call__(self, id):
  67         if id not in self.map:
  68             self.func(id)
  69         df = defer.Deferred()
  70         def timeout():
  71             self.map[id].remove((df, timer))
  72             if not self.map[id]:
  73                 del self.map[id]
  74             df.errback(failure.Failure(defer.TimeoutError('in ReplyMatcher')))
  75         timer = reactor.callLater(self.timeout, timeout)
  76         self.map.setdefault(id, set()).add((df, timer))
  77         return df
  78
  79     def got_response(self, id, resp):
  80         if id not in self.map:
  81             return
  82         for df, timer in self.map.pop(id):
  83             df.callback(resp)
  84             timer.cancel()
  85
  86 class GenericDeferrer(object):
  87     '''
  88     Converts query with identifier/got response interface to deferred interface
  89     '''
  90
  91     def __init__(self, max_id, func, timeout=5):
  92         self.max_id = max_id
  93         self.func = func
  94         self.timeout = timeout
  95         self.map = {}
  96
  97     def __call__(self, *args, **kwargs):
  98         while True:
  99             id = random.randrange(self.max_id)
 100             if id not in self.map:
 101                 break
 102         df = defer.Deferred()
 103         def timeout():
 104             self.map.pop(id)
 105             df.errback(failure.Failure(defer.TimeoutError('in GenericDeferrer')))
 106         timer = reactor.callLater(self.timeout, timeout)
 107         self.func(id, *args, **kwargs)
 108         self.map[id] = df, timer
 109         return df
 110
 111     def got_response(self, id, resp):
 112         if id not in self.map:
 113             return
 114         df, timer = self.map.pop(id)
 115         timer.cancel()
 116         df.callback(resp)
 117
 118     def respond_all(self, resp):
 119         while self.map:
 120             df, timer = self.map.popitem()
 121             timer.cancel()
 122             df.errback(resp)
 123
 124 class NotNowError(Exception):
 125     pass
 126
 127 class DeferredCacher(object):
 128     '''
 129     like memoize, but for functions that return Deferreds
 130
 131     @DeferredCacher
 132     def f(x):
 133         ...
 134         return df
 135
 136     @DeferredCacher.with_backing(bsddb.hashopen(...))
 137     def f(x):
 138         ...
 139         return df
 140     '''
 141
 142     @classmethod
 143     def with_backing(cls, backing):
 144         return lambda func: cls(func, backing)
 145
 146     def __init__(self, func, backing=None):
 147         if backing is None:
 148             backing = {}
 149
 150         self.func = func
 151         self.backing = backing
 152         self.waiting = {}
 153
 154     @defer.inlineCallbacks
 155     def __call__(self, key):
 156         if key in self.waiting:
 157             yield self.waiting[key]
 158
 159         if key in self.backing:
 160             defer.returnValue(self.backing[key])
 161         else:
 162             self.waiting[key] = defer.Deferred()
 163             try:
 164                 value = yield self.func(key)
 165             finally:
 166                 self.waiting.pop(key).callback(None)
 167
 168         self.backing[key] = value
 169         defer.returnValue(value)
 170
 171     _nothing = object()
 172     def call_now(self, key, default=_nothing):
 173         if key in self.backing:
 174             return self.backing[key]
 175         if key not in self.waiting:
 176             self.waiting[key] = defer.Deferred()
 177             def cb(value):
 178                 self.backing[key] = value
 179                 self.waiting.pop(key).callback(None)
 180             def eb(fail):
 181                 self.waiting.pop(key).callback(None)
 182                 if fail.check(RetrySilentlyException):
 183                     return
 184                 print
 185                 print 'Error when requesting noncached value:'
 186                 fail.printTraceback()
 187                 print
 188             self.func(key).addCallback(cb).addErrback(eb)
 189         if default is not self._nothing:
 190             return default
 191         raise NotNowError(key)