Revert "remove deferral.sleep's canceller, as it wasn't used anywhere and caused...

[p2pool.git] / p2pool / util / deferral.py

diff --git a/p2pool/util/deferral.py b/p2pool/util/deferral.py
index 28589ca..aed3adc 100644 (file)
--- a/p2pool/util/deferral.py
+++ b/p2pool/util/deferral.py
@@ -1,16 +1,31 @@
 from __future__ import division
 
 import itertools
+import random
+import sys
 
 from twisted.internet import defer, reactor
 from twisted.python import failure, log
 
 def sleep(t):
-    d = defer.Deferred()
-    reactor.callLater(t, d.callback, None)
+    d = defer.Deferred(canceller=lambda d_: dc.cancel())
+    dc = reactor.callLater(t, d.callback, None)
     return d
 
-def retry(message, delay, max_retries=None):
+def run_repeatedly(f, *args, **kwargs):
+    current_dc = [None]
+    def step():
+        delay = f(*args, **kwargs)
+        current_dc[0] = reactor.callLater(delay, step)
+    step()
+    def stop():
+        current_dc[0].cancel()
+    return stop
+
+class RetrySilentlyException(Exception):
+    pass
+
+def retry(message='Error:', delay=3, max_retries=None, traceback=True):
     '''
     @retry('Error getting block:', 1)
     @defer.inlineCallbacks
@@ -24,10 +39,14 @@ def retry(message, delay, max_retries=None):
             for i in itertools.count():
                 try:
                     result = yield func(*args, **kwargs)
-                except:
+                except Exception, e:
                     if i == max_retries:
                         raise
-                    log.err(None, message)
+                    if not isinstance(e, RetrySilentlyException):
+                        if traceback:
+                            log.err(None, message)
+                        else:
+                            print >>sys.stderr, message, e
                     yield sleep(delay)
                 else:
                     defer.returnValue(result)
@@ -69,10 +88,11 @@ class GenericDeferrer(object):
     Converts query with identifier/got response interface to deferred interface
     '''
     
-    def __init__(self, max_id, func, timeout=5):
+    def __init__(self, max_id, func, timeout=5, on_timeout=lambda: None):
         self.max_id = max_id
         self.func = func
         self.timeout = timeout
+        self.on_timeout = on_timeout
         self.map = {}
     
     def __call__(self, *args, **kwargs):
@@ -80,13 +100,20 @@ class GenericDeferrer(object):
             id = random.randrange(self.max_id)
             if id not in self.map:
                 break
-        df = defer.Deferred()
+        def cancel(df):
+            df, timer = self.map.pop(id)
+            timer.cancel()
+        try:
+            df = defer.Deferred(cancel)
+        except TypeError:
+            df = defer.Deferred() # handle older versions of Twisted
         def timeout():
             self.map.pop(id)
             df.errback(failure.Failure(defer.TimeoutError('in GenericDeferrer')))
+            self.on_timeout()
         timer = reactor.callLater(self.timeout, timeout)
-        self.func(id, *args, **kwargs)
         self.map[id] = df, timer
+        self.func(id, *args, **kwargs)
         return df
     
     def got_response(self, id, resp):
@@ -95,6 +122,12 @@ class GenericDeferrer(object):
         df, timer = self.map.pop(id)
         timer.cancel()
         df.callback(resp)
+    
+    def respond_all(self, resp):
+        while self.map:
+            id, (df, timer) = self.map.popitem()
+            timer.cancel()
+            df.errback(resp)
 
 class NotNowError(Exception):
     pass
@@ -145,25 +178,22 @@ class DeferredCacher(object):
     
     _nothing = object()
     def call_now(self, key, default=_nothing):
-        if key in self.waiting:
-            if default is not self._nothing:
-                return default
-            raise NotNowError(key)
-        
         if key in self.backing:
             return self.backing[key]
-        else:
+        if key not in self.waiting:
             self.waiting[key] = defer.Deferred()
             def cb(value):
                 self.backing[key] = value
                 self.waiting.pop(key).callback(None)
             def eb(fail):
                 self.waiting.pop(key).callback(None)
+                if fail.check(RetrySilentlyException):
+                    return
                 print
                 print 'Error when requesting noncached value:'
                 fail.printTraceback()
                 print
             self.func(key).addCallback(cb).addErrback(eb)
-            if default is not self._nothing:
-                return default
-            raise NotNowError(key)
+        if default is not self._nothing:
+            return default
+        raise NotNowError(key)