p2pool/util/deferral.py

   1 from __future__ import division
   2
   3 import random
   4
   5 from twisted.internet import defer, reactor
   6 from twisted.python import failure, log
   7
   8 def sleep(t):
   9     d = defer.Deferred()
  10     reactor.callLater(t, d.callback, None)
  11     return d
  12
  13 def retry(message, delay):
  14     '''
  15     @retry('Error getting block:', 1)
  16     @defer.inlineCallbacks
  17     def get_block(hash):
  18         ...
  19     '''
  20
  21     def retry2(func):
  22         @defer.inlineCallbacks
  23         def f(*args, **kwargs):
  24             while True:
  25                 try:
  26                     result = yield func(*args, **kwargs)
  27                 except:
  28                     log.err(None, message)
  29                     yield sleep(delay)
  30                 else:
  31                     defer.returnValue(result)
  32         return f
  33     return retry2
  34
  35 class ReplyMatcher(object):
  36     '''
  37     Converts request/got response interface to deferred interface
  38     '''
  39
  40     def __init__(self, func, timeout=5):
  41         self.func = func
  42         self.timeout = timeout
  43         self.map = {}
  44
  45     def __call__(self, id):
  46         self.func(id)
  47         uniq = random.randrange(2**256)
  48         df = defer.Deferred()
  49         def timeout():
  50             df, timer = self.map[id].pop(uniq)
  51             df.errback(failure.Failure(defer.TimeoutError('in ReplyMatcher')))
  52             if not self.map[id]:
  53                 del self.map[id]
  54         self.map.setdefault(id, {})[uniq] = (df, reactor.callLater(self.timeout, timeout))
  55         return df
  56
  57     def got_response(self, id, resp):
  58         if id not in self.map:
  59             return
  60         for df, timer in self.map.pop(id).itervalues():
  61             timer.cancel()
  62             df.callback(resp)