Disconnect the insertNode calls from the callers so errors don't affect them.

[quix0rs-apt-p2p.git] / apt_p2p_Khashmir / actions.py

diff --git a/apt_p2p_Khashmir/actions.py b/apt_p2p_Khashmir/actions.py
index b54632065002dbd2a0dc6c4439451ffa3c2d5ab0..bd1cd2f6f1905672717b1b90e0358040afbdf8e9 100644 (file)
--- a/apt_p2p_Khashmir/actions.py
+++ b/apt_p2p_Khashmir/actions.py
@@ -3,10 +3,13 @@
 
 """Details of how to perform actions on remote peers."""
 
+from datetime import datetime
+
 from twisted.internet import reactor, defer
 from twisted.python import log
 
 from khash import intify
+from ktable import K
 from util import uncompact
 
 class ActionBase:
@@ -20,6 +23,8 @@ class ActionBase:
     @ivar config: the configuration variables for the DHT
     @type action: C{string}
     @ivar action: the name of the action to call on remote nodes
+    @type stats: L{stats.StatsLogger}
+    @ivar stats: the statistics modules to report to
     @type num: C{long}
     @ivar num: the target key in integer form
     @type queried: C{dictionary}
@@ -45,6 +50,8 @@ class ActionBase:
         the requests that are currently outstanding
     @type finished: C{boolean}
     @ivar finished: whether the action is done
+    @type started: C{datetime.datetime}
+    @ivar started: the time the action was started at
     @type sort: C{method}
     @ivar sort: used to sort nodes by their proximity to the target
     """
@@ -74,7 +81,8 @@ class ActionBase:
         self.target = target
         self.config = config
         self.action = action
-        stats.startedAction(action)
+        self.stats = stats
+        self.stats.startedAction(action)
         self.num = intify(target)
         self.queried = {}
         self.answered = {}
@@ -86,6 +94,7 @@ class ActionBase:
         self.outstanding = 0
         self.outstanding_results = 0
         self.finished = False
+        self.started = datetime.now()
     
         def sort(a, b, num=self.num):
             """Sort nodes relative to the ID we are looking for."""
@@ -100,6 +109,7 @@ class ActionBase:
     #{ Main operation
     def goWithNodes(self, nodes):
         """Start the action's process with a list of nodes to contact."""
+        self.started = datetime.now()
         for node in nodes:
             self.found[node.id] = node
         self.sortNodes()
@@ -107,6 +117,9 @@ class ActionBase:
     
     def schedule(self):
         """Schedule requests to be sent to remote nodes."""
+        if self.finished:
+            return
+        
         # Check if we are already done
         if self.desired_results and ((len(self.results) >= abs(self.desired_results)) or
                                      (self.desired_results < 0 and
@@ -114,9 +127,11 @@ class ActionBase:
             self.finished = True
             result = self.generateResult()
             reactor.callLater(0, self.callback, *result)
+            return
 
-        if self.finished or (self.desired_results and 
-                             len(self.results) + self.outstanding_results >= abs(self.desired_results)):
+        # Check if we have enough outstanding results coming
+        if (self.desired_results and 
+            len(self.results) + self.outstanding_results >= abs(self.desired_results)):
             return
         
         # Loop for each node that should be processed
@@ -171,7 +186,7 @@ class ActionBase:
 
     def gotResponse(self, dict, node, expected_results, df):
         """Receive a response from a remote node."""
-        self.caller.insertNode(node)
+        reactor.callLater(0, self.caller.insertNode, node)
         if self.finished or self.answered.has_key(node.id):
             # a day late and a dollar short
             return
@@ -183,8 +198,7 @@ class ActionBase:
 
     def actionFailed(self, err, node, expected_results, df):
         """Receive an error from a remote node."""
-        log.msg("action %s failed (%s) %s/%s" % (self.action, self.config['PORT'], node.host, node.port))
-        log.err(err)
+        log.msg("action %s failed on %s/%s: %s" % (self.action, node.host, node.port, err.getErrorMessage()))
         self.caller.table.nodeFailed(node)
         self.outstanding -= 1
         self.outstanding_results -= expected_results
@@ -218,7 +232,7 @@ class ActionBase:
         This implementation is suitable for a recurring search over all nodes.
         """
         self.sortNodes()
-        return self.sorted_nodes[:self.config['K']]
+        return self.sorted_nodes[:K]
     
     def generateArgs(self, node):
         """Generate the arguments to the node's action.
@@ -235,6 +249,7 @@ class ActionBase:
 
     def generateResult(self, nodes):
         """Create the final result to return to the L{callback} function."""
+        self.stats.completedAction(self.action, self.started)
         return []
         
 
@@ -253,7 +268,8 @@ class FindNode(ActionBase):
     def generateResult(self):
         """Result is the K closest nodes to the target."""
         self.sortNodes()
-        return (self.sorted_nodes[:self.config['K']], )
+        self.stats.completedAction(self.action, self.started)
+        return (self.sorted_nodes[:K], )
     
 
 class FindValue(ActionBase):
@@ -271,6 +287,7 @@ class FindValue(ActionBase):
     def generateResult(self):
         """Result is the nodes that have values, sorted by proximity to the key."""
         self.sortNodes()
+        self.stats.completedAction(self.action, self.started)
         return ([node for node in self.sorted_nodes if node.num_values > 0], )
     
 
@@ -312,6 +329,7 @@ class GetValue(ActionBase):
 
     def generateResult(self):
         """Results have all been returned, now send the empty list to end the action."""
+        self.stats.completedAction(self.action, self.started)
         return (self.target, [])
         
 
@@ -344,4 +362,5 @@ class StoreValue(ActionBase):
     
     def generateResult(self):
         """Return all the response IDs received."""
+        self.stats.completedAction(self.action, self.started)
         return (self.target, self.value, self.results.values())