initial kademlia files

1 files changed, 422 insertions, 0 deletions

diff --git a/kademlia/routingtable.py b/kademlia/routingtable.py
new file mode 100644
index 0000000..2f43a47
--- /dev/null
+++ b/kademlia/routingtable.py
@@ -0,0 +1,422 @@
+# This library is free software, distributed under the terms of
+# the GNU Lesser General Public License Version 3, or any later version.
+# See the COPYING file included in this archive
+#
+# The docstrings in this module contain epytext markup; API documentation
+# may be created by processing this file with epydoc: http://epydoc.sf.net
+
+import time, random
+
+import constants
+import kbucket
+from protocol import TimeoutError
+
+class RoutingTable(object):
+    """ Interface for RPC message translators/formatters
+    
+    Classes inheriting from this should provide a suitable routing table for
+    a parent Node object (i.e. the local entity in the Kademlia network)
+    """
+    def __init__(self, parentNodeID):
+        """
+        @param parentNodeID: The 160-bit node ID of the node to which this
+                             routing table belongs
+        @type parentNodeID: str
+        """
+    def addContact(self, contact):
+        """ Add the given contact to the correct k-bucket; if it already
+        exists, its status will be updated
+
+        @param contact: The contact to add to this node's k-buckets
+        @type contact: kademlia.contact.Contact
+        """
+    
+    def distance(self, keyOne, keyTwo):
+        """ Calculate the XOR result between two string variables
+        
+        @return: XOR result of two long variables
+        @rtype: long
+        """
+        valKeyOne = long(keyOne.encode('hex'), 16)
+        valKeyTwo = long(keyTwo.encode('hex'), 16)
+        return valKeyOne ^ valKeyTwo
+
+    def findCloseNodes(self, key, count, _rpcNodeID=None):
+        """ Finds a number of known nodes closest to the node/value with the
+        specified key.
+        
+        @param key: the 160-bit key (i.e. the node or value ID) to search for
+        @type key: str
+        @param count: the amount of contacts to return
+        @type count: int
+        @param _rpcNodeID: Used during RPC, this is be the sender's Node ID
+                           Whatever ID is passed in the paramater will get
+                           excluded from the list of returned contacts.
+        @type _rpcNodeID: str
+        
+        @return: A list of node contacts (C{kademlia.contact.Contact instances})
+                 closest to the specified key. 
+                 This method will return C{k} (or C{count}, if specified)
+                 contacts if at all possible; it will only return fewer if the
+                 node is returning all of the contacts that it knows of.
+        @rtype: list
+        """
+    def getContact(self, contactID):
+        """ Returns the (known) contact with the specified node ID
+        
+        @raise ValueError: No contact with the specified contact ID is known
+                           by this node
+        """
+    def getRefreshList(self, startIndex=0, force=False):
+        """ Finds all k-buckets that need refreshing, starting at the
+        k-bucket with the specified index, and returns IDs to be searched for
+        in order to refresh those k-buckets
+
+        @param startIndex: The index of the bucket to start refreshing at;
+                           this bucket and those further away from it will
+                           be refreshed. For example, when joining the
+                           network, this node will set this to the index of
+                           the bucket after the one containing it's closest
+                           neighbour.
+        @type startIndex: index
+        @param force: If this is C{True}, all buckets (in the specified range)
+                      will be refreshed, regardless of the time they were last
+                      accessed.
+        @type force: bool
+        
+        @return: A list of node ID's that the parent node should search for
+                 in order to refresh the routing Table
+        @rtype: list
+        """
+    def removeContact(self, contactID):
+        """ Remove the contact with the specified node ID from the routing
+        table
+        
+        @param contactID: The node ID of the contact to remove
+        @type contactID: str
+        """
+    def touchKBucket(self, key):
+        """ Update the "last accessed" timestamp of the k-bucket which covers
+        the range containing the specified key in the key/ID space
+        
+        @param key: A key in the range of the target k-bucket
+        @type key: str
+        """
+
+
+class TreeRoutingTable(RoutingTable):
+    """ This class implements a routing table used by a Node class.
+    
+    The Kademlia routing table is a binary tree whose leaves are k-buckets,
+    where each k-bucket contains nodes with some common prefix of their IDs.
+    This prefix is the k-bucket's position in the binary tree; it therefore
+    covers some range of ID values, and together all of the k-buckets cover
+    the entire 160-bit ID (or key) space (with no overlap).
+    
+    @note: In this implementation, nodes in the tree (the k-buckets) are
+    added dynamically, as needed; this technique is described in the 13-page
+    version of the Kademlia paper, in section 2.4. It does, however, use the
+    C{PING} RPC-based k-bucket eviction algorithm described in section 2.2 of
+    that paper.
+    """
+    def __init__(self, parentNodeID):
+        """
+        @param parentNodeID: The 160-bit node ID of the node to which this
+                             routing table belongs
+        @type parentNodeID: str
+        """
+        # Create the initial (single) k-bucket covering the range of the entire 160-bit ID space
+        self._buckets = [kbucket.KBucket(rangeMin=0, rangeMax=2**160)]
+        self._parentNodeID = parentNodeID
+
+    def addContact(self, contact):
+        """ Add the given contact to the correct k-bucket; if it already
+        exists, its status will be updated
+
+        @param contact: The contact to add to this node's k-buckets
+        @type contact: kademlia.contact.Contact
+        """
+        if contact.id == self._parentNodeID:
+            return
+
+        bucketIndex = self._kbucketIndex(contact.id)
+        try:
+            self._buckets[bucketIndex].addContact(contact)
+        except kbucket.BucketFull:
+            # The bucket is full; see if it can be split (by checking if its range includes the host node's id)
+            if self._buckets[bucketIndex].keyInRange(self._parentNodeID):
+                self._splitBucket(bucketIndex)
+                # Retry the insertion attempt
+                self.addContact(contact)
+            else:
+                # We can't split the k-bucket
+                # NOTE:
+                # In section 2.4 of the 13-page version of the Kademlia paper, it is specified that
+                # in this case, the new contact should simply be dropped. However, in section 2.2,
+                # it states that the head contact in the k-bucket (i.e. the least-recently seen node)
+                # should be pinged - if it does not reply, it should be dropped, and the new contact
+                # added to the tail of the k-bucket. This implementation follows section 2.2 regarding
+                # this point.            
+                headContact = self._buckets[bucketIndex]._contacts[0]
+    
+                def replaceContact(failure):
+                    """ Callback for the deferred PING RPC to see if the head
+                    node in the k-bucket is still responding
+                    
+                    @type failure: twisted.python.failure.Failure
+                    """
+                    failure.trap(TimeoutError)
+                    print '==replacing contact=='
+                    # Remove the old contact...
+                    deadContactID = failure.getErrorMessage()
+                    try:
+                        self._buckets[bucketIndex].removeContact(deadContactID)
+                    except ValueError:
+                        # The contact has already been removed (probably due to a timeout)
+                        pass
+                    # ...and add the new one at the tail of the bucket
+                    self.addContact(contact)
+                
+                # Ping the least-recently seen contact in this k-bucket
+                headContact = self._buckets[bucketIndex]._contacts[0]
+                df = headContact.ping()
+                # If there's an error (i.e. timeout), remove the head contact, and append the new one
+                df.addErrback(replaceContact)
+                
+    def findCloseNodes(self, key, count, _rpcNodeID=None):
+        """ Finds a number of known nodes closest to the node/value with the
+        specified key.
+        
+        @param key: the 160-bit key (i.e. the node or value ID) to search for
+        @type key: str
+        @param count: the amount of contacts to return
+        @type count: int
+        @param _rpcNodeID: Used during RPC, this is be the sender's Node ID
+                           Whatever ID is passed in the paramater will get
+                           excluded from the list of returned contacts.
+        @type _rpcNodeID: str
+        
+        @return: A list of node contacts (C{kademlia.contact.Contact instances})
+                 closest to the specified key. 
+                 This method will return C{k} (or C{count}, if specified)
+                 contacts if at all possible; it will only return fewer if the
+                 node is returning all of the contacts that it knows of.
+        @rtype: list
+        """
+        #if key == self.id:
+        #    bucketIndex = 0 #TODO: maybe not allow this to continue?
+        #else:
+        bucketIndex = self._kbucketIndex(key)
+        closestNodes = self._buckets[bucketIndex].getContacts(constants.k, _rpcNodeID)
+        # This method must return k contacts (even if we have the node with the specified key as node ID), 
+        # unless there is less than k remote nodes in the routing table
+        i = 1
+        canGoLower = bucketIndex-i >= 0
+        canGoHigher = bucketIndex+i < len(self._buckets)
+        # Fill up the node list to k nodes, starting with the closest neighbouring nodes known 
+        while len(closestNodes) < constants.k and (canGoLower or canGoHigher):
+            #TODO: this may need to be optimized
+            if canGoLower:
+                closestNodes.extend(self._buckets[bucketIndex-i].getContacts(constants.k - len(closestNodes), _rpcNodeID))
+                canGoLower = bucketIndex-(i+1) >= 0
+            if canGoHigher:
+                closestNodes.extend(self._buckets[bucketIndex+i].getContacts(constants.k - len(closestNodes), _rpcNodeID))
+                canGoHigher = bucketIndex+(i+1) < len(self._buckets)
+            i += 1
+        return closestNodes
+
+    def getContact(self, contactID):
+        """ Returns the (known) contact with the specified node ID
+        
+        @raise ValueError: No contact with the specified contact ID is known
+                           by this node
+        """
+        bucketIndex = self._kbucketIndex(contactID)
+        try:
+            contact = self._buckets[bucketIndex].getContact(contactID)
+        except ValueError:
+            raise
+        else:
+            return contact
+
+    def getRefreshList(self, startIndex=0, force=False):
+        """ Finds all k-buckets that need refreshing, starting at the
+        k-bucket with the specified index, and returns IDs to be searched for
+        in order to refresh those k-buckets
+
+        @param startIndex: The index of the bucket to start refreshing at;
+                           this bucket and those further away from it will
+                           be refreshed. For example, when joining the
+                           network, this node will set this to the index of
+                           the bucket after the one containing it's closest
+                           neighbour.
+        @type startIndex: index
+        @param force: If this is C{True}, all buckets (in the specified range)
+                      will be refreshed, regardless of the time they were last
+                      accessed.
+        @type force: bool
+        
+        @return: A list of node ID's that the parent node should search for
+                 in order to refresh the routing Table
+        @rtype: list
+        """
+        bucketIndex = startIndex
+        refreshIDs = []
+        for bucket in self._buckets[startIndex:]:
+            if force or (int(time.time()) - bucket.lastAccessed >= constants.refreshTimeout):
+                searchID = self._randomIDInBucketRange(bucketIndex)
+                refreshIDs.append(searchID)
+            bucketIndex += 1
+        return refreshIDs
+
+    def removeContact(self, contactID):
+        """ Remove the contact with the specified node ID from the routing
+        table
+        
+        @param contactID: The node ID of the contact to remove
+        @type contactID: str
+        """
+        bucketIndex = self._kbucketIndex(contactID)
+        try:
+            self._buckets[bucketIndex].removeContact(contactID)
+        except ValueError:
+            #print 'removeContact(): Contact not in routing table'
+            return
+
+    def touchKBucket(self, key):
+        """ Update the "last accessed" timestamp of the k-bucket which covers
+        the range containing the specified key in the key/ID space
+        
+        @param key: A key in the range of the target k-bucket
+        @type key: str
+        """
+        bucketIndex = self._kbucketIndex(key)
+        self._buckets[bucketIndex].lastAccessed = int(time.time())
+
+    def _kbucketIndex(self, key):
+        """ Calculate the index of the k-bucket which is responsible for the
+        specified key (or ID)
+        
+        @param key: The key for which to find the appropriate k-bucket index
+        @type key: str
+        
+        @return: The index of the k-bucket responsible for the specified key
+        @rtype: int
+        """
+        valKey = long(key.encode('hex'), 16)
+        i = 0
+        for bucket in self._buckets:
+            if bucket.keyInRange(valKey):
+                return i
+            else:
+                i += 1
+        return i
+
+    def _randomIDInBucketRange(self, bucketIndex):
+        """ Returns a random ID in the specified k-bucket's range
+        
+        @param bucketIndex: The index of the k-bucket to use
+        @type bucketIndex: int
+        """
+        idValue = random.randrange(self._buckets[bucketIndex].rangeMin, self._buckets[bucketIndex].rangeMax)
+        randomID = hex(idValue)[2:]
+        if randomID[-1] == 'L':
+            randomID = randomID[:-1]
+        if len(randomID) % 2 != 0:
+            randomID = '0' + randomID
+        randomID = randomID.decode('hex')
+        randomID = (20 - len(randomID))*'\x00' + randomID
+        return randomID
+
+    def _splitBucket(self, oldBucketIndex):
+        """ Splits the specified k-bucket into two new buckets which together
+        cover the same range in the key/ID space
+        
+        @param oldBucketIndex: The index of k-bucket to split (in this table's
+                               list of k-buckets)
+        @type oldBucketIndex: int
+        """
+        # Resize the range of the current (old) k-bucket
+        oldBucket = self._buckets[oldBucketIndex]
+        splitPoint = oldBucket.rangeMax - (oldBucket.rangeMax - oldBucket.rangeMin)/2
+        # Create a new k-bucket to cover the range split off from the old bucket
+        newBucket = kbucket.KBucket(splitPoint, oldBucket.rangeMax)
+        oldBucket.rangeMax = splitPoint
+        # Now, add the new bucket into the routing table tree
+        self._buckets.insert(oldBucketIndex + 1, newBucket)
+        # Finally, copy all nodes that belong to the new k-bucket into it...
+        for contact in oldBucket._contacts:
+            if newBucket.keyInRange(contact.id):
+                newBucket.addContact(contact)
+        # ...and remove them from the old bucket
+        for contact in newBucket._contacts:
+            oldBucket.removeContact(contact)
+
+class OptimizedTreeRoutingTable(TreeRoutingTable):
+    """ A version of the "tree"-type routing table specified by Kademlia,
+    along with contact accounting optimizations specified in section 4.1 of
+    of the 13-page version of the Kademlia paper.
+    """
+    def __init__(self, parentNodeID):
+        TreeRoutingTable.__init__(self, parentNodeID)
+        # Cache containing nodes eligible to replace stale k-bucket entries
+        self._replacementCache = {}
+        
+    def addContact(self, contact):
+        """ Add the given contact to the correct k-bucket; if it already
+        exists, its status will be updated
+
+        @param contact: The contact to add to this node's k-buckets
+        @type contact: kademlia.contact.Contact
+        """
+        if contact.id == self._parentNodeID:
+            return
+
+        # Initialize/reset the "successively failed RPC" counter
+        contact.failedRPCs = 0
+
+        bucketIndex = self._kbucketIndex(contact.id)
+        try:
+            self._buckets[bucketIndex].addContact(contact)
+        except kbucket.BucketFull:
+            # The bucket is full; see if it can be split (by checking if its range includes the host node's id)
+            if self._buckets[bucketIndex].keyInRange(self._parentNodeID):
+                self._splitBucket(bucketIndex)
+                # Retry the insertion attempt
+                self.addContact(contact)
+            else:
+                # We can't split the k-bucket
+                # NOTE: This implementation follows section 4.1 of the 13 page version
+                # of the Kademlia paper (optimized contact accounting without PINGs
+                #- results in much less network traffic, at the expense of some memory)
+
+                # Put the new contact in our replacement cache for the corresponding k-bucket (or update it's position if it exists already)
+                if not self._replacementCache.has_key(bucketIndex):
+                    self._replacementCache[bucketIndex] = []
+                if contact in self._replacementCache[bucketIndex]:
+                    self._replacementCache[bucketIndex].remove(contact)
+                #TODO: Using k to limit the size of the contact replacement cache - maybe define a seperate value for this in constants.py?
+                elif len(self._replacementCache) >= constants.k:
+                    self._replacementCache.pop(0)
+                self._replacementCache[bucketIndex].append(contact)
+    
+    def removeContact(self, contactID):
+        """ Remove the contact with the specified node ID from the routing
+        table
+        
+        @param contactID: The node ID of the contact to remove
+        @type contactID: str
+        """
+        bucketIndex = self._kbucketIndex(contactID)
+        try:
+            contact = self._buckets[bucketIndex].getContact(contactID)
+        except ValueError:
+            #print 'removeContact(): Contact not in routing table'
+            return
+        contact.failedRPCs += 1
+        if contact.failedRPCs >= 5:        
+            self._buckets[bucketIndex].removeContact(contactID)
+            # Replace this stale contact with one from our replacemnent cache, if we have any
+            if self._replacementCache.has_key(bucketIndex):
+                if len(self._replacementCache[bucketIndex]) > 0:
+                    self._buckets[bucketIndex].addContact( self._replacementCache[bucketIndex].pop() )