--- /dev/null
+## Copyright 2002-2004 Andrew Loewenstern, All Rights Reserved
+# see LICENSE.txt for license information
+
+"""Details of how to perform actions on remote peers."""
+
+from twisted.internet import reactor
+from twisted.python import log
+
+from khash import intify
+from util import uncompact
+
+class ActionBase:
+ """Base class for some long running asynchronous proccesses like finding nodes or values.
+
+ @type caller: L{khashmir.Khashmir}
+ @ivar caller: the DHT instance that is performing the action
+ @type target: C{string}
+ @ivar target: the target of the action, usually a DHT key
+ @type config: C{dictionary}
+ @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 num: C{long}
+ @ivar num: the target key in integer form
+ @type queried: C{dictionary}
+ @ivar queried: the nodes that have been queried for this action,
+ keys are node IDs, values are the node itself
+ @type answered: C{dictionary}
+ @ivar answered: the nodes that have answered the queries
+ @type found: C{dictionary}
+ @ivar found: nodes that have been found so far by the action
+ @type sorted_nodes: C{list} of L{node.Node}
+ @ivar sorted_nodes: a sorted list of nodes by there proximity to the key
+ @type results: C{dictionary}
+ @ivar results: keys are the results found so far by the action
+ @type desired_results: C{int}
+ @ivar desired_results: the minimum number of results that are needed
+ before the action should stop
+ @type callback: C{method}
+ @ivar callback: the method to call with the results
+ @type outstanding: C{int}
+ @ivar outstanding: the number of requests currently outstanding
+ @type outstanding_results: C{int}
+ @ivar outstanding_results: the number of results that are expected from
+ the requests that are currently outstanding
+ @type finished: C{boolean}
+ @ivar finished: whether the action is done
+ @type sort: C{method}
+ @ivar sort: used to sort nodes by their proximity to the target
+ """
+
+ def __init__(self, caller, target, callback, config, action, num_results = None):
+ """Initialize the action.
+
+ @type caller: L{khashmir.Khashmir}
+ @param caller: the DHT instance that is performing the action
+ @type target: C{string}
+ @param target: the target of the action, usually a DHT key
+ @type callback: C{method}
+ @param callback: the method to call with the results
+ @type config: C{dictionary}
+ @param config: the configuration variables for the DHT
+ @type action: C{string}
+ @param action: the name of the action to call on remote nodes
+ @type num_results: C{int}
+ @param num_results: the minimum number of results that are needed before
+ the action should stop (optional, defaults to getting all the results)
+
+ """
+
+ self.caller = caller
+ self.target = target
+ self.config = config
+ self.action = action
+ self.num = intify(target)
+ self.queried = {}
+ self.answered = {}
+ self.found = {}
+ self.sorted_nodes = []
+ self.results = {}
+ self.desired_results = num_results
+ self.callback = callback
+ self.outstanding = 0
+ self.outstanding_results = 0
+ self.finished = False
+
+ def sort(a, b, num=self.num):
+ """Sort nodes relative to the ID we are looking for."""
+ x, y = num ^ a.num, num ^ b.num
+ if x > y:
+ return 1
+ elif x < y:
+ return -1
+ return 0
+ self.sort = sort
+
+ #{ Main operation
+ def goWithNodes(self, nodes):
+ """Start the action's process with a list of nodes to contact."""
+ for node in nodes:
+ if node.id == self.caller.node.id:
+ continue
+ else:
+ self.found[node.id] = node
+ self.sortNodes()
+ self.schedule()
+
+ def schedule(self):
+ """Schedule requests to be sent to remote nodes."""
+ # Check if we are already done
+ if self.desired_results and ((len(self.results) >= abs(self.desired_results)) or
+ (self.desired_results < 0 and
+ len(self.answered) >= self.config['STORE_REDUNDANCY'])):
+ self.finished = True
+ result = self.generateResult()
+ reactor.callLater(0, self.callback, *result)
+
+ if self.finished or (self.desired_results and
+ len(self.results) + self.outstanding_results >= abs(self.desired_results)):
+ return
+
+ # Loop for each node that should be processed
+ for node in self.getNodesToProcess():
+ # Don't send requests twice or to ourself
+ if node.id not in self.queried and node.id != self.caller.node.id:
+ self.queried[node.id] = 1
+
+ # Get the action to call on the node
+ try:
+ f = getattr(node, self.action)
+ except AttributeError:
+ log.msg("%s doesn't have a %s method!" % (node, self.action))
+ else:
+ # Get the arguments to the action's method
+ try:
+ args, expected_results = self.generateArgs(node)
+ except ValueError:
+ pass
+ else:
+ # Call the action on the remote node
+ self.outstanding += 1
+ self.outstanding_results += expected_results
+ df = f(self.caller.node.id, *args)
+ df.addCallbacks(self.gotResponse, self.actionFailed,
+ callbackArgs = (node, expected_results),
+ errbackArgs = (node, expected_results))
+
+ # We might have to stop for now
+ if (self.outstanding >= self.config['CONCURRENT_REQS'] or
+ (self.desired_results and
+ len(self.results) + self.outstanding_results >= abs(self.desired_results))):
+ break
+
+ assert self.outstanding >= 0
+ assert self.outstanding_results >= 0
+
+ # If no requests are outstanding, then we are done
+ if self.outstanding == 0:
+ self.finished = True
+ result = self.generateResult()
+ reactor.callLater(0, self.callback, *result)
+
+ def gotResponse(self, dict, node, expected_results):
+ """Receive a response from a remote node."""
+ self.caller.insertNode(node)
+ if self.finished or self.answered.has_key(node.id):
+ # a day late and a dollar short
+ return
+ self.outstanding -= 1
+ self.outstanding_results -= expected_results
+ self.answered[node.id] = 1
+ self.processResponse(dict['rsp'])
+ self.schedule()
+
+ def actionFailed(self, err, node, expected_results):
+ """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)
+ self.caller.table.nodeFailed(node)
+ self.outstanding -= 1
+ self.outstanding_results -= expected_results
+ self.schedule()
+
+ def handleGotNodes(self, nodes):
+ """Process any received node contact info in the response.
+
+ Not called by default, but suitable for being called by
+ L{processResponse} in a recursive node search.
+ """
+ for compact_node in nodes:
+ node_contact = uncompact(compact_node)
+ node = self.caller.Node(node_contact)
+ if not self.found.has_key(node.id):
+ self.found[node.id] = node
+
+ def sortNodes(self):
+ """Sort the nodes, if necessary.
+
+ Assumes nodes are never removed from the L{found} dictionary.
+ """
+ if len(self.sorted_nodes) != len(self.found):
+ self.sorted_nodes = self.found.values()
+ self.sorted_nodes.sort(self.sort)
+
+ #{ Subclass for specific actions
+ def getNodesToProcess(self):
+ """Generate a list of nodes to process next.
+
+ This implementation is suitable for a recurring search over all nodes.
+ """
+ self.sortNodes()
+ return self.sorted_nodes[:self.config['K']]
+
+ def generateArgs(self, node):
+ """Generate the arguments to the node's action.
+
+ These arguments will be appended to our node ID when calling the action.
+ Also return the number of results expected from this request.
+
+ @raise ValueError: if the node should not be queried
+ """
+ return (self.target, ), 0
+
+ def processResponse(self, dict):
+ """Process the response dictionary received from the remote node."""
+ self.handleGotNodes(dict['nodes'])
+
+ def generateResult(self, nodes):
+ """Create the final result to return to the L{callback} function."""
+ return []
+
+
+class FindNode(ActionBase):
+ """Find the closest nodes to the key."""
+
+ def __init__(self, caller, target, callback, config, action="findNode"):
+ ActionBase.__init__(self, caller, target, callback, config, action)
+
+ def processResponse(self, dict):
+ """Save the token received from each node."""
+ if dict["id"] in self.found:
+ self.found[dict["id"]].updateToken(dict.get('token', ''))
+ self.handleGotNodes(dict['nodes'])
+
+ def generateResult(self):
+ """Result is the K closest nodes to the target."""
+ self.sortNodes()
+ return (self.sorted_nodes[:self.config['K']], )
+
+
+class FindValue(ActionBase):
+ """Find the closest nodes to the key and check for values."""
+
+ def __init__(self, caller, target, callback, config, action="findValue"):
+ ActionBase.__init__(self, caller, target, callback, config, action)
+
+ def processResponse(self, dict):
+ """Save the number of values each node has."""
+ if dict["id"] in self.found:
+ self.found[dict["id"]].updateNumValues(dict.get('num', 0))
+ self.handleGotNodes(dict['nodes'])
+
+ def generateResult(self):
+ """Result is the nodes that have values, sorted by proximity to the key."""
+ self.sortNodes()
+ return ([node for node in self.sorted_nodes if node.num_values > 0], )
+
+
+class GetValue(ActionBase):
+ """Retrieve values from a list of nodes."""
+
+ def __init__(self, caller, target, local_results, num_results, callback, config, action="getValue"):
+ """Initialize the action with the locally available results.
+
+ @type local_results: C{list} of C{string}
+ @param local_results: the values that were available in this node
+ """
+ ActionBase.__init__(self, caller, target, callback, config, action, num_results)
+ if local_results:
+ for result in local_results:
+ self.results[result] = 1
+
+ def getNodesToProcess(self):
+ """Nodes are never added, always return the same sorted node list."""
+ return self.sorted_nodes
+
+ def generateArgs(self, node):
+ """Arguments include the number of values to request."""
+ if node.num_values > 0:
+ # Request all desired results from each node, just to be sure.
+ num_values = abs(self.desired_results) - len(self.results)
+ assert num_values > 0
+ if num_values > node.num_values:
+ num_values = 0
+ return (self.target, num_values), node.num_values
+ else:
+ raise ValueError, "Don't try and get values from this node because it doesn't have any"
+
+ def processResponse(self, dict):
+ """Save the returned values, calling the L{callback} each time there are new ones."""
+ if dict.has_key('values'):
+ def x(y, z=self.results):
+ if not z.has_key(y):
+ z[y] = 1
+ return y
+ else:
+ return None
+ z = len(dict['values'])
+ v = filter(None, map(x, dict['values']))
+ if len(v):
+ reactor.callLater(0, self.callback, self.target, v)
+
+ def generateResult(self):
+ """Results have all been returned, now send the empty list to end the action."""
+ return (self.target, [])
+
+
+class StoreValue(ActionBase):
+ """Store a value in a list of nodes."""
+
+ def __init__(self, caller, target, value, num_results, callback, config, action="storeValue"):
+ """Initialize the action with the value to store.
+
+ @type value: C{string}
+ @param value: the value to store in the nodes
+ """
+ ActionBase.__init__(self, caller, target, callback, config, action, num_results)
+ self.value = value
+
+ def getNodesToProcess(self):
+ """Nodes are never added, always return the same sorted list."""
+ return self.sorted_nodes
+
+ def generateArgs(self, node):
+ """Args include the value to store and the node's token."""
+ if node.token:
+ return (self.target, self.value, node.token), 1
+ else:
+ raise ValueError, "Don't store at this node since we don't know it's token"
+
+ def processResponse(self, dict):
+ """Save the response, though it should be nothin but the ID."""
+ self.results[dict["id"]] = dict
+
+ def generateResult(self):
+ """Return all the response IDs received."""
+ return (self.target, self.value, self.results.values())
projects / quix0rs-apt-p2p.git / blobdiff