diff options
| -rw-r--r-- | kademlia/__init__.py | 30 | ||||
| -rw-r--r-- | kademlia/constants.py | 46 | ||||
| -rw-r--r-- | kademlia/contact.py | 58 | ||||
| -rw-r--r-- | kademlia/datastore.py | 183 | ||||
| -rw-r--r-- | kademlia/encoding.py | 143 | ||||
| -rw-r--r-- | kademlia/kbucket.py | 134 | ||||
| -rw-r--r-- | kademlia/msgformat.py | 72 | ||||
| -rw-r--r-- | kademlia/msgtypes.py | 46 | ||||
| -rw-r--r-- | kademlia/node.py | 786 | ||||
| -rw-r--r-- | kademlia/protocol.py | 305 | ||||
| -rw-r--r-- | kademlia/routingtable.py | 422 |
11 files changed, 2225 insertions, 0 deletions
diff --git a/kademlia/__init__.py b/kademlia/__init__.py new file mode 100644 index 0000000..4ed7d81 --- /dev/null +++ b/kademlia/__init__.py @@ -0,0 +1,30 @@ +# 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 + +""" Kademlia DHT implementation + +This package contains Entangled's implementation of the Kademlia +distributed hash table (DHT). + +The main modules in this package are "C{node}" (which contains the Kademlia +implementation's main interface, namely the C{Node} class), "C{datastore}" +(physical data storage mechanisms), "C{constants}" (several constant values +defining the Kademlia network), "C{routingtable}" (different Kademlia routing +table implementations) and "C{protocol}" (actual network communications). + +The Node class is directly exposed in the main Entangled package +("C{entangled}") as KademliaNode, and as Node in this package +("C{entangled.kademlia}"). It is designed to be customizable; the data storage +mechansims may (and should) be directly specified by client applications via +the node's contructor. The same holds true for the node's routing table and +network protocol used. This potentially allows the Kademlia node to be used +with a TCP-based protocol, instead of the provided UDP-based one. + +Client applications should also modify the values found in +C{entangled.kademlia.constants} to suit their needs. Refer to the C{constants} +module for documentation on what these values control. +""" + +from node import Node +from datastore import DictDataStore, SQLiteDataStore diff --git a/kademlia/constants.py b/kademlia/constants.py new file mode 100644 index 0000000..ab1d3f5 --- /dev/null +++ b/kademlia/constants.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python +# +# 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 + +""" This module defines the charaterizing constants of the Kademlia network + +C{checkRefreshInterval} and C{udpDatagramMaxSize} are implementation-specific +constants, and do not affect general Kademlia operation. +""" + +######### KADEMLIA CONSTANTS ########### + +#: Small number Representing the degree of parallelism in network calls +alpha = 3 + +#: Maximum number of contacts stored in a bucket; this should be an even number +k = 8 + +#: Timeout for network operations (in seconds) +rpcTimeout = 5 + +# Delay between iterations of iterative node lookups (for loose parallelism) (in seconds) +iterativeLookupDelay = rpcTimeout / 2 + +#: If a k-bucket has not been used for this amount of time, refresh it (in seconds) +refreshTimeout = 3600 # 1 hour +#: The interval at which nodes replicate (republish/refresh) data they are holding +replicateInterval = refreshTimeout +# The time it takes for data to expire in the network; the original publisher of the data +# will also republish the data at this time if it is still valid +dataExpireTimeout = 86400 # 24 hours + +######## IMPLEMENTATION-SPECIFIC CONSTANTS ########### + +#: The interval in which the node should check its whether any buckets need refreshing, +#: or whether any data needs to be republished (in seconds) +checkRefreshInterval = refreshTimeout/5 + +#: Max size of a single UDP datagram, in bytes. If a message is larger than this, it will +#: be spread accross several UDP packets. +udpDatagramMaxSize = 8192 # 8 KB diff --git a/kademlia/contact.py b/kademlia/contact.py new file mode 100644 index 0000000..9fc4ed4 --- /dev/null +++ b/kademlia/contact.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python +# +# 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 + +class Contact(object): + """ Encapsulation for remote contact + + This class contains information on a single remote contact, and also + provides a direct RPC API to the remote node which it represents + """ + def __init__(self, id, ipAddress, udpPort, networkProtocol, firstComm=0): + self.id = id + self.address = ipAddress + self.port = udpPort + self._networkProtocol = networkProtocol + self.commTime = firstComm + + def __eq__(self, other): + if isinstance(other, Contact): + return self.id == other.id + elif isinstance(other, str): + return self.id == other + else: + return False + + def __ne__(self, other): + if isinstance(other, Contact): + return self.id != other.id + elif isinstance(other, str): + return self.id != other + else: + return True + + def __str__(self): + return '<%s.%s object; IP address: %s, UDP port: %d>' % (self.__module__, self.__class__.__name__, self.address, self.port) + + def __getattr__(self, name): + """ This override allows the host node to call a method of the remote + node (i.e. this contact) as if it was a local function. + + For instance, if C{remoteNode} is a instance of C{Contact}, the + following will result in C{remoteNode}'s C{test()} method to be + called with argument C{123}:: + remoteNode.test(123) + + Such a RPC method call will return a Deferred, which will callback + when the contact responds with the result (or an error occurs). + This happens via this contact's C{_networkProtocol} object (i.e. the + host Node's C{_protocol} object). + """ + def _sendRPC(*args, **kwargs): + return self._networkProtocol.sendRPC(self, name, args, **kwargs) + return _sendRPC diff --git a/kademlia/datastore.py b/kademlia/datastore.py new file mode 100644 index 0000000..45633e1 --- /dev/null +++ b/kademlia/datastore.py @@ -0,0 +1,183 @@ +#!/usr/bin/env python +# +# 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 UserDict +import sqlite3 +import cPickle as pickle +import time +import os + + +class DataStore(UserDict.DictMixin): + """ Interface for classes implementing physical storage (for data + published via the "STORE" RPC) for the Kademlia DHT + + @note: This provides an interface for a dict-like object + """ + def keys(self): + """ Return a list of the keys in this data store """ + + def lastPublished(self, key): + """ Get the time the C{(key, value)} pair identified by C{key} + was last published """ + + def originalPublisherID(self, key): + """ Get the original publisher of the data's node ID + + @param key: The key that identifies the stored data + @type key: str + + @return: Return the node ID of the original publisher of the + C{(key, value)} pair identified by C{key}. + """ + + def originalPublishTime(self, key): + """ Get the time the C{(key, value)} pair identified by C{key} + was originally published """ + + def setItem(self, key, value, lastPublished, originallyPublished, originalPublisherID): + """ Set the value of the (key, value) pair identified by C{key}; + this should set the "last published" value for the (key, value) + pair to the current time + """ + + def __getitem__(self, key): + """ Get the value identified by C{key} """ + + def __setitem__(self, key, value): + """ Convenience wrapper to C{setItem}; this accepts a tuple in the + format: (value, lastPublished, originallyPublished, originalPublisherID) """ + self.setItem(key, *value) + + def __delitem__(self, key): + """ Delete the specified key (and its value) """ + +class DictDataStore(DataStore): + """ A datastore using an in-memory Python dictionary """ + def __init__(self): + # Dictionary format: + # { <key>: (<value>, <lastPublished>, <originallyPublished> <originalPublisherID>) } + self._dict = {} + + def keys(self): + """ Return a list of the keys in this data store """ + return self._dict.keys() + + def lastPublished(self, key): + """ Get the time the C{(key, value)} pair identified by C{key} + was last published """ + return self._dict[key][1] + + def originalPublisherID(self, key): + """ Get the original publisher of the data's node ID + + @param key: The key that identifies the stored data + @type key: str + + @return: Return the node ID of the original publisher of the + C{(key, value)} pair identified by C{key}. + """ + return self._dict[key][3] + + def originalPublishTime(self, key): + """ Get the time the C{(key, value)} pair identified by C{key} + was originally published """ + return self._dict[key][2] + + def setItem(self, key, value, lastPublished, originallyPublished, originalPublisherID): + """ Set the value of the (key, value) pair identified by C{key}; + this should set the "last published" value for the (key, value) + pair to the current time + """ + self._dict[key] = (value, lastPublished, originallyPublished, originalPublisherID) + + def __getitem__(self, key): + """ Get the value identified by C{key} """ + return self._dict[key][0] + + def __delitem__(self, key): + """ Delete the specified key (and its value) """ + del self._dict[key] + + +class SQLiteDataStore(DataStore): + """ Example of a SQLite database-based datastore + """ + def __init__(self, dbFile=':memory:'): + """ + @param dbFile: The name of the file containing the SQLite database; if + unspecified, an in-memory database is used. + @type dbFile: str + """ + createDB = not os.path.exists(dbFile) + self._db = sqlite3.connect(dbFile) + self._db.isolation_level = None + self._db.text_factory = str + if createDB: + self._db.execute('CREATE TABLE data(key, value, lastPublished, originallyPublished, originalPublisherID)') + self._cursor = self._db.cursor() + + def keys(self): + """ Return a list of the keys in this data store """ + keys = [] + try: + self._cursor.execute("SELECT key FROM data") + for row in self._cursor: + keys.append(row[0].decode('hex')) + finally: + return keys + + def lastPublished(self, key): + """ Get the time the C{(key, value)} pair identified by C{key} + was last published """ + return int(self._dbQuery(key, 'lastPublished')) + + def originalPublisherID(self, key): + """ Get the original publisher of the data's node ID + + @param key: The key that identifies the stored data + @type key: str + + @return: Return the node ID of the original publisher of the + C{(key, value)} pair identified by C{key}. + """ + return self._dbQuery(key, 'originalPublisherID') + + def originalPublishTime(self, key): + """ Get the time the C{(key, value)} pair identified by C{key} + was originally published """ + return int(self._dbQuery(key, 'originallyPublished')) + + def setItem(self, key, value, lastPublished, originallyPublished, originalPublisherID): + # Encode the key so that it doesn't corrupt the database + encodedKey = key.encode('hex') + self._cursor.execute("select key from data where key=:reqKey", {'reqKey': encodedKey}) + if self._cursor.fetchone() == None: + self._cursor.execute('INSERT INTO data(key, value, lastPublished, originallyPublished, originalPublisherID) VALUES (?, ?, ?, ?, ?)', (encodedKey, buffer(pickle.dumps(value, pickle.HIGHEST_PROTOCOL)), lastPublished, originallyPublished, originalPublisherID)) + else: + self._cursor.execute('UPDATE data SET value=?, lastPublished=?, originallyPublished=?, originalPublisherID=? WHERE key=?', (buffer(pickle.dumps(value, pickle.HIGHEST_PROTOCOL)), lastPublished, originallyPublished, originalPublisherID, encodedKey)) + + def _dbQuery(self, key, columnName, unpickle=False): + try: + self._cursor.execute("SELECT %s FROM data WHERE key=:reqKey" % columnName, {'reqKey': key.encode('hex')}) + row = self._cursor.fetchone() + value = str(row[0]) + except TypeError: + raise KeyError, key + else: + if unpickle: + return pickle.loads(value) + else: + return value + + def __getitem__(self, key): + return self._dbQuery(key, 'value', unpickle=True) + + def __delitem__(self, key): + self._cursor.execute("DELETE FROM data WHERE key=:reqKey", {'reqKey': key.encode('hex')}) diff --git a/kademlia/encoding.py b/kademlia/encoding.py new file mode 100644 index 0000000..b0eca32 --- /dev/null +++ b/kademlia/encoding.py @@ -0,0 +1,143 @@ +#!/usr/bin/env python +# +# 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 + +class DecodeError(Exception): + """ Should be raised by an C{Encoding} implementation if decode operation + fails + """ + +class Encoding(object): + """ Interface for RPC message encoders/decoders + + All encoding implementations used with this library should inherit and + implement this. + """ + def encode(self, data): + """ Encode the specified data + + @param data: The data to encode + This method has to support encoding of the following + types: C{str}, C{int} and C{long} + Any additional data types may be supported as long as the + implementing class's C{decode()} method can successfully + decode them. + + @return: The encoded data + @rtype: str + """ + def decode(self, data): + """ Decode the specified data string + + @param data: The data (byte string) to decode. + @type data: str + + @return: The decoded data (in its correct type) + """ + +class Bencode(Encoding): + """ Implementation of a Bencode-based algorithm (Bencode is the encoding + algorithm used by Bittorrent). + + @note: This algorithm differs from the "official" Bencode algorithm in + that it can encode/decode floating point values in addition to + integers. + """ + + def encode(self, data): + """ Encoder implementation of the Bencode algorithm + + @param data: The data to encode + @type data: int, long, tuple, list, dict or str + + @return: The encoded data + @rtype: str + """ + if type(data) in (int, long): + return 'i%de' % data + elif type(data) == str: + return '%d:%s' % (len(data), data) + elif type(data) in (list, tuple): + encodedListItems = '' + for item in data: + encodedListItems += self.encode(item) + return 'l%se' % encodedListItems + elif type(data) == dict: + encodedDictItems = '' + keys = data.keys() + keys.sort() + for key in keys: + encodedDictItems += self.encode(key) + encodedDictItems += self.encode(data[key]) + return 'd%se' % encodedDictItems + elif type(data) == float: + # This (float data type) is a non-standard extension to the original Bencode algorithm + return 'f%fe' % data + elif data == None: + # This (None/NULL data type) is a non-standard extension to the original Bencode algorithm + return 'n' + else: + raise TypeError, "Cannot bencode '%s' object" % type(data) + + def decode(self, data): + """ Decoder implementation of the Bencode algorithm + + @param data: The encoded data + @type data: str + + @note: This is a convenience wrapper for the recursive decoding + algorithm, C{_decodeRecursive} + + @return: The decoded data, as a native Python type + @rtype: int, list, dict or str + """ + if len(data) == 0: + raise DecodeError, 'Cannot decode empty string' + return self._decodeRecursive(data)[0] + + @staticmethod + def _decodeRecursive(data, startIndex=0): + """ Actual implementation of the recursive Bencode algorithm + + Do not call this; use C{decode()} instead + """ + if data[startIndex] == 'i': + endPos = data[startIndex:].find('e')+startIndex + return (int(data[startIndex+1:endPos]), endPos+1) + elif data[startIndex] == 'l': + startIndex += 1 + decodedList = [] + while data[startIndex] != 'e': + listData, startIndex = Bencode._decodeRecursive(data, startIndex) + decodedList.append(listData) + return (decodedList, startIndex+1) + elif data[startIndex] == 'd': + startIndex += 1 + decodedDict = {} + while data[startIndex] != 'e': + key, startIndex = Bencode._decodeRecursive(data, startIndex) + value, startIndex = Bencode._decodeRecursive(data, startIndex) + decodedDict[key] = value + return (decodedDict, startIndex) + elif data[startIndex] == 'f': + # This (float data type) is a non-standard extension to the original Bencode algorithm + endPos = data[startIndex:].find('e')+startIndex + return (float(data[startIndex+1:endPos]), endPos+1) + elif data[startIndex] == 'n': + # This (None/NULL data type) is a non-standard extension to the original Bencode algorithm + return (None, startIndex+1) + else: + splitPos = data[startIndex:].find(':')+startIndex + try: + length = int(data[startIndex:splitPos]) + except ValueError, e: + raise DecodeError, e + startIndex = splitPos+1 + endPos = startIndex+length + bytes = data[startIndex:endPos] + return (bytes, endPos) diff --git a/kademlia/kbucket.py b/kademlia/kbucket.py new file mode 100644 index 0000000..b6eaedc --- /dev/null +++ b/kademlia/kbucket.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python +# +# 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 constants + +class BucketFull(Exception): + """ Raised when the bucket is full """ + + +class KBucket(object): + """ Description - later + """ + def __init__(self, rangeMin, rangeMax): + """ + @param rangeMin: The lower boundary for the range in the 160-bit ID + space covered by this k-bucket + @param rangeMax: The upper boundary for the range in the ID space + covered by this k-bucket + """ + self.lastAccessed = 0 + self.rangeMin = rangeMin + self.rangeMax = rangeMax + self._contacts = list() + + def addContact(self, contact): + """ Add contact to _contact list in the right order. This will move the + contact to the end of the k-bucket if it is already present. + + @raise kademlia.kbucket.BucketFull: Raised when the bucket is full and + the contact isn't in the bucket + already + + @param contact: The contact to add + @type contact: kademlia.contact.Contact + """ + if contact in self._contacts: + # Move the existing contact to the end of the list + # - using the new contact to allow add-on data (e.g. optimization-specific stuff) to pe updated as well + self._contacts.remove(contact) + self._contacts.append(contact) + elif len(self._contacts) < constants.k: + self._contacts.append(contact) + else: + raise BucketFull("No space in bucket to insert contact") + + def getContact(self, contactID): + """ Get the contact specified node ID""" + index = self._contacts.index(contactID) + return self._contacts[index] + + def getContacts(self, count=-1, excludeContact=None): + """ Returns a list containing up to the first count number of contacts + + @param count: The amount of contacts to return (if 0 or less, return + all contacts) + @type count: int + @param excludeContact: A contact to exclude; if this contact is in + the list of returned values, it will be + discarded before returning. If a C{str} is + passed as this argument, it must be the + contact's ID. + @type excludeContact: kademlia.contact.Contact or str + + + @raise IndexError: If the number of requested contacts is too large + + @return: Return up to the first count number of contacts in a list + If no contacts are present an empty is returned + @rtype: list + """ + # Return all contacts in bucket + if count <= 0: + count = len(self._contacts) + + # Get current contact number + currentLen = len(self._contacts) + + # If count greater than k - return only k contacts + if count > constants.k: + count = constants.k + + # Check if count value in range and, + # if count number of contacts are available + if not currentLen: + contactList = list() + + # length of list less than requested amount + elif currentLen < count: + contactList = self._contacts[0:currentLen] + # enough contacts in list + else: + contactList = self._contacts[0:count] + + if excludeContact in contactList: + contactList.remove(excludeContact) + + return contactList + + def removeContact(self, contact): + """ Remove given contact from list + + @param contact: The contact to remove, or a string containing the + contact's node ID + @type contact: kademlia.contact.Contact or str + + @raise ValueError: The specified contact is not in this bucket + """ + self._contacts.remove(contact) + + def keyInRange(self, key): + """ Tests whether the specified key (i.e. node ID) is in the range + of the 160-bit ID space covered by this k-bucket (in otherwords, it + returns whether or not the specified key should be placed in this + k-bucket) + + @param key: The key to test + @type key: str or int + + @return: C{True} if the key is in this k-bucket's range, or C{False} + if not. + @rtype: bool + """ + if isinstance(key, str): + key = long(key.encode('hex'), 16) + return self.rangeMin <= key < self.rangeMax + + def __len__(self): + return len(self._contacts) diff --git a/kademlia/msgformat.py b/kademlia/msgformat.py new file mode 100644 index 0000000..c0c5ce8 --- /dev/null +++ b/kademlia/msgformat.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python +# +# 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 msgtypes + +class MessageTranslator(object): + """ Interface for RPC message translators/formatters + + Classes inheriting from this should provide a translation services between + the classes used internally by this Kademlia implementation and the actual + data that is transmitted between nodes. + """ + def fromPrimitive(self, msgPrimitive): + """ Create an RPC Message from a message's string representation + + @param msgPrimitive: The unencoded primitive representation of a message + @type msgPrimitive: str, int, list or dict + + @return: The translated message object + @rtype: entangled.kademlia.msgtypes.Message + """ + + def toPrimitive(self, message): + """ Create a string representation of a message + + @param message: The message object + @type message: msgtypes.Message + + @return: The message's primitive representation in a particular + messaging format + @rtype: str, int, list or dict + """ + +class DefaultFormat(MessageTranslator): + """ The default on-the-wire message format for this library """ + typeRequest, typeResponse, typeError = range(3) + headerType, headerMsgID, headerNodeID, headerPayload, headerArgs = range(5) + + def fromPrimitive(self, msgPrimitive): + msgType = msgPrimitive[self.headerType] + if msgType == self.typeRequest: + msg = msgtypes.RequestMessage(msgPrimitive[self.headerNodeID], msgPrimitive[self.headerPayload], msgPrimitive[self.headerArgs], msgPrimitive[self.headerMsgID]) + elif msgType == self.typeResponse: + msg = msgtypes.ResponseMessage(msgPrimitive[self.headerMsgID], msgPrimitive[self.headerNodeID], msgPrimitive[self.headerPayload]) + elif msgType == self.typeError: + msg = msgtypes.ErrorMessage(msgPrimitive[self.headerMsgID], msgPrimitive[self.headerNodeID], msgPrimitive[self.headerPayload], msgPrimitive[self.headerArgs]) + else: + # Unknown message, no payload + msg = msgtypes.Message(msgPrimitive[self.headerMsgID], msgPrimitive[self.headerNodeID]) + return msg + + def toPrimitive(self, message): + msg = {self.headerMsgID: message.id, + self.headerNodeID: message.nodeID} + if isinstance(message, msgtypes.RequestMessage): + msg[self.headerType] = self.typeRequest + msg[self.headerPayload] = message.request + msg[self.headerArgs] = message.args + elif isinstance(message, msgtypes.ErrorMessage): + msg[self.headerType] = self.typeError + msg[self.headerPayload] = message.exceptionType + msg[self.headerArgs] = message.response + elif isinstance(message, msgtypes.ResponseMessage): + msg[self.headerType] = self.typeResponse + msg[self.headerPayload] = message.response + return msg diff --git a/kademlia/msgtypes.py b/kademlia/msgtypes.py new file mode 100644 index 0000000..4e6eb18 --- /dev/null +++ b/kademlia/msgtypes.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python +# +# 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 hashlib +import random + +class Message(object): + """ Base class for messages - all "unknown" messages use this class """ + def __init__(self, rpcID, nodeID): + self.id = rpcID + self.nodeID = nodeID + + +class RequestMessage(Message): + """ Message containing an RPC request """ + def __init__(self, nodeID, method, methodArgs, rpcID=None): + if rpcID == None: + hash = hashlib.sha1() + hash.update(str(random.getrandbits(255))) + rpcID = hash.digest() + Message.__init__(self, rpcID, nodeID) + self.request = method + self.args = methodArgs + + +class ResponseMessage(Message): + """ Message containing the result from a successful RPC request """ + def __init__(self, rpcID, nodeID, response): + Message.__init__(self, rpcID, nodeID) + self.response = response + + +class ErrorMessage(ResponseMessage): + """ Message containing the error from an unsuccessful RPC request """ + def __init__(self, rpcID, nodeID, exceptionType, errorMessage): + ResponseMessage.__init__(self, rpcID, nodeID, errorMessage) + if isinstance(exceptionType, type): + self.exceptionType = '%s.%s' % (exceptionType.__module__, exceptionType.__name__) + else: + self.exceptionType = exceptionType diff --git a/kademlia/node.py b/kademlia/node.py new file mode 100644 index 0000000..19f577d --- /dev/null +++ b/kademlia/node.py @@ -0,0 +1,786 @@ +#!/usr/bin/env python +# +# 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 hashlib, random, time + +from twisted.internet import defer + +import constants +import routingtable +import datastore +import protocol +import twisted.internet.reactor +import twisted.internet.threads +from contact import Contact + +def rpcmethod(func): + """ Decorator to expose Node methods as remote procedure calls + + Apply this decorator to methods in the Node class (or a subclass) in order + to make them remotely callable via the DHT's RPC mechanism. + """ + func.rpcmethod = True + return func + +class Node(object): + """ Local node in the Kademlia network + + This class represents a single local node in a Kademlia network; in other + words, this class encapsulates an Entangled-using application's "presence" + in a Kademlia network. + + In Entangled, all interactions with the Kademlia network by a client + application is performed via this class (or a subclass). + """ + def __init__(self, id=None, udpPort=4000, dataStore=None, routingTableClass=None, networkProtocol=None): + """ + @param dataStore: The data store to use. This must be class inheriting + from the C{DataStore} interface (or providing the + same API). How the data store manages its data + internally is up to the implementation of that data + store. + @type dataStore: entangled.kademlia.datastore.DataStore + @param routingTable: The routing table class to use. Since there exists + some ambiguity as to how the routing table should be + implemented in Kademlia, a different routing table + may be used, as long as the appropriate API is + exposed. This should be a class, not an object, + in order to allow the Node to pass an + auto-generated node ID to the routingtable object + upon instantiation (if necessary). + @type routingTable: entangled.kademlia.routingtable.RoutingTable + @param networkProtocol: The network protocol to use. This can be + overridden from the default to (for example) + change the format of the physical RPC messages + being transmitted. + @type networkProtocol: entangled.kademlia.protocol.KademliaProtocol + """ + if id != None: + self.id = id + else: + self.id = self._generateID() + self.port = udpPort + self._listeningPort = None # object implementing Twisted IListeningPort + # This will contain a deferred created when joining the network, to enable publishing/retrieving information from + # the DHT as soon as the node is part of the network (add callbacks to this deferred if scheduling such operations + # before the node has finished joining the network) + self._joinDeferred = None + # Create k-buckets (for storing contacts) + #self._buckets = [] + #for i in range(160): + # self._buckets.append(kbucket.KBucket()) + if routingTableClass == None: + self._routingTable = routingtable.OptimizedTreeRoutingTable(self.id) + else: + self._routingTable = routingTableClass(self.id) + + # Initialize this node's network access mechanisms + if networkProtocol == None: + self._protocol = protocol.KademliaProtocol(self) + else: + self._protocol = networkProtocol + # Initialize the data storage mechanism used by this node + if dataStore == None: + self._dataStore = datastore.DictDataStore() + else: + self._dataStore = dataStore + # Try to restore the node's state... + if 'nodeState' in self._dataStore: + state = self._dataStore['nodeState'] + self.id = state['id'] + for contactTriple in state['closestNodes']: + contact = Contact(contactTriple[0], contactTriple[1], contactTriple[2], self._protocol) + self._routingTable.addContact(contact) + + def __del__(self): + self._persistState() + self._listeningPort.stopListening() + + def joinNetwork(self, knownNodeAddresses=None): + """ Causes the Node to join the Kademlia network; normally, this + should be called before any other DHT operations. + + @param knownNodeAddresses: A sequence of tuples containing IP address + information for existing nodes on the + Kademlia network, in the format: + C{(<ip address>, (udp port>)} + @type knownNodeAddresses: tuple + """ + # Prepare the underlying Kademlia protocol + self._listeningPort = twisted.internet.reactor.listenUDP(self.port, self._protocol) #IGNORE:E1101 + # Create temporary contact information for the list of addresses of known nodes + if knownNodeAddresses != None: + bootstrapContacts = [] + for address, port in knownNodeAddresses: + contact = Contact(self._generateID(), address, port, self._protocol) + bootstrapContacts.append(contact) + else: + bootstrapContacts = None + # Initiate the Kademlia joining sequence - perform a search for this node's own ID + self._joinDeferred = self._iterativeFind(self.id, bootstrapContacts) +# #TODO: Refresh all k-buckets further away than this node's closest neighbour +# def getBucketAfterNeighbour(*args): +# for i in range(160): +# if len(self._buckets[i]) > 0: +# return i+1 +# return 160 +# df.addCallback(getBucketAfterNeighbour) +# df.addCallback(self._refreshKBuckets) + #protocol.reactor.callLater(10, self.printContacts) + self._joinDeferred.addCallback(self._persistState) + # Start refreshing k-buckets periodically, if necessary + twisted.internet.reactor.callLater(constants.checkRefreshInterval, self._refreshNode) #IGNORE:E1101 + + def printContacts(self): + print '\n\nNODE CONTACTS\n===============' + for i in range(len(self._routingTable._buckets)): + for contact in self._routingTable._buckets[i]._contacts: + print contact + print '==================================' + #twisted.internet.reactor.callLater(10, self.printContacts) + + def iterativeStore(self, key, value, originalPublisherID=None, age=0): + """ The Kademlia store operation + + Call this to store/republish data in the DHT. + + @param key: The hashtable key of the data + @type key: str + @param value: The actual data (the value associated with C{key}) + @type value: str + @param originalPublisherID: The node ID of the node that is the + B{original} publisher of the data + @type originalPublisherID: str + @param age: The relative age of the data (time in seconds since it was + originally published). Note that the original publish time + isn't actually given, to compensate for clock skew between + different nodes. + @type age: int + """ + #print ' iterativeStore called' + if originalPublisherID == None: + originalPublisherID = self.id + # Prepare a callback for doing "STORE" RPC calls + def executeStoreRPCs(nodes): + #print ' .....execStoreRPCs called' + if len(nodes) >= constants.k: + # If this node itself is closer to the key than the last (furthest) node in the list, + # we should store the value at ourselves as well + if self._routingTable.distance(key, self.id) < self._routingTable.distance(key, nodes[-1].id): + nodes.pop() + self.store(key, value, originalPublisherID=originalPublisherID, age=age) + else: + self.store(key, value, originalPublisherID=originalPublisherID, age=age) + for contact in nodes: + contact.store(key, value, originalPublisherID, age) + return nodes + # Find k nodes closest to the key... + df = self.iterativeFindNode(key) + # ...and send them STORE RPCs as soon as they've been found + df.addCallback(executeStoreRPCs) + return df + + def iterativeFindNode(self, key): + """ The basic Kademlia node lookup operation + + Call this to find a remote node in the P2P overlay network. + + @param key: the 160-bit key (i.e. the node or value ID) to search for + @type key: str + + @return: This immediately returns a deferred object, which will return + a list of k "closest" contacts (C{kademlia.contact.Contact} + objects) to the specified key as soon as the operation is + finished. + @rtype: twisted.internet.defer.Deferred + """ + return self._iterativeFind(key) + + def iterativeFindValue(self, key): + """ The Kademlia search operation (deterministic) + + Call this to retrieve data from the DHT. + + @param key: the 160-bit key (i.e. the value ID) to search for + @type key: str + + @return: This immediately returns a deferred object, which will return + either one of two things: + - If the value was found, it will return a Python + dictionary containing the searched-for key (the C{key} + parameter passed to this method), and its associated + value, in the format: + C{<str>key: <str>data_value} + - If the value was not found, it will return a list of k + "closest" contacts (C{kademlia.contact.Contact} objects) + to the specified key + @rtype: twisted.internet.defer.Deferred + """ + # Prepare a callback for this operation + outerDf = defer.Deferred() + def checkResult(result): + if type(result) == dict: + # We have found the value; now see who was the closest contact without it... + if 'closestNodeNoValue' in result: + # ...and store the key/value pair + contact = result['closestNodeNoValue'] + contact.store(key, result[key]) + outerDf.callback(result) + else: + # The value wasn't found, but a list of contacts was returned + # Now, see if we have the value (it might seem wasteful to search on the network + # first, but it ensures that all values are properly propagated through the + # network + if key in self._dataStore: + # Ok, we have the value locally, so use that + value = self._dataStore[key] + # Send this value to the closest node without it + if len(result) > 0: + contact = result[0] + contact.store(key, value) + outerDf.callback({key: value}) + else: + # Ok, value does not exist in DHT at all + outerDf.callback(result) + + # Execute the search + df = self._iterativeFind(key, rpc='findValue') + df.addCallback(checkResult) + return outerDf + + def addContact(self, contact): + """ Add/update the given contact; simple wrapper for the same method + in this object's RoutingTable object + + @param contact: The contact to add to this node's k-buckets + @type contact: kademlia.contact.Contact + """ + self._routingTable.addContact(contact) + + def removeContact(self, contactID): + """ Remove the contact with the specified node ID from this node's + table of known nodes. This is a simple wrapper for the same method + in this object's RoutingTable object + + @param contactID: The node ID of the contact to remove + @type contactID: str + """ + self._routingTable.removeContact(contactID) + + def findContact(self, contactID): + """ Find a entangled.kademlia.contact.Contact object for the specified + cotact ID + + @param contactID: The contact ID of the required Contact object + @type contactID: str + + @return: Contact object of remote node with the specified node ID, + or None if the contact was not found + @rtype: twisted.internet.defer.Deferred + """ + try: + contact = self._routingTable.getContact(contactID) + df = defer.Deferred() + df.callback(contact) + except ValueError: + def parseResults(nodes): + if contactID in nodes: + contact = nodes[nodes.index(contactID)] + return contact + else: + return None + df = self.iterativeFindNode(contactID) + df.addCallback(parseResults) + return df + + @rpcmethod + def ping(self): + """ Used to verify contact between two Kademlia nodes + + @rtype: str + """ + return 'pong' + + @rpcmethod + def store(self, key, value, originalPublisherID=None, age=0, **kwargs): + """ Store the received data in this node's local hash table + + @param key: The hashtable key of the data + @type key: str + @param value: The actual data (the value associated with C{key}) + @type value: str + @param originalPublisherID: The node ID of the node that is the + B{original} publisher of the data + @type originalPublisherID: str + @param age: The relative age of the data (time in seconds since it was + originally published). Note that the original publish time + isn't actually given, to compensate for clock skew between + different nodes. + @type age: int + + @rtype: str + + @todo: Since the data (value) may be large, passing it around as a buffer + (which is the case currently) might not be a good idea... will have + to fix this (perhaps use a stream from the Protocol class?) + """ + # Get the sender's ID (if any) + if '_rpcNodeID' in kwargs: + rpcSenderID = kwargs['_rpcNodeID'] + else: + rpcSenderID = None + + if originalPublisherID == None: + if rpcSenderID != None: + originalPublisherID = rpcSenderID + else: + raise TypeError, 'No publisher specifed, and RPC caller ID not available. Data requires an original publisher.' + + now = int(time.time()) + originallyPublished = now - age + self._dataStore.setItem(key, value, now, originallyPublished, originalPublisherID) + return 'OK' + + @rpcmethod + def findNode(self, key, **kwargs): + """ 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 + + @return: A list of contact triples 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 + """ + # Get the sender's ID (if any) + if '_rpcNodeID' in kwargs: + rpcSenderID = kwargs['_rpcNodeID'] + else: + rpcSenderID = None + contacts = self._routingTable.findCloseNodes(key, constants.k, rpcSenderID) + contactTriples = [] + for contact in contacts: + contactTriples.append( (contact.id, contact.address, contact.port) ) + return contactTriples + + @rpcmethod + def findValue(self, key, **kwargs): + """ Return the value associated with the specified key if present in + this node's data, otherwise execute FIND_NODE for the key + + @param key: The hashtable key of the data to return + @type key: str + + @return: A dictionary containing the requested key/value pair, + or a list of contact triples closest to the requested key. + @rtype: dict or list + """ + if key in self._dataStore: + return {key: self._dataStore[key]} + else: + return self.findNode(key, **kwargs) + +# 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 _generateID(self): + """ Generates a 160-bit pseudo-random identifier + + @return: A globally unique 160-bit pseudo-random identifier + @rtype: str + """ + hash = hashlib.sha1() + hash.update(str(random.getrandbits(255))) + return hash.digest() + + def _iterativeFind(self, key, startupShortlist=None, rpc='findNode'): + """ The basic Kademlia iterative lookup operation (for nodes/values) + + This builds a list of k "closest" contacts through iterative use of + the "FIND_NODE" RPC, or if C{findValue} is set to C{True}, using the + "FIND_VALUE" RPC, in which case the value (if found) may be returned + instead of a list of contacts + + @param key: the 160-bit key (i.e. the node or value ID) to search for + @type key: str + @param startupShortlist: A list of contacts to use as the starting + shortlist for this search; this is normally + only used when the node joins the network + @type startupShortlist: list + @param rpc: The name of the RPC to issue to remote nodes during the + Kademlia lookup operation (e.g. this sets whether this + algorithm should search for a data value (if + rpc='findValue') or not. It can thus be used to perform + other operations that piggy-back on the basic Kademlia + lookup operation (Entangled's "delete" RPC, for instance). + @type rpc: str + + @return: If C{findValue} is C{True}, the algorithm will stop as soon + as a data value for C{key} is found, and return a dictionary + containing the key and the found value. Otherwise, it will + return a list of the k closest nodes to the specified key + @rtype: twisted.internet.defer.Deferred + """ + if rpc != 'findNode': + findValue = True + else: + findValue = False + shortlist = [] + if startupShortlist == None: + shortlist = self._routingTable.findCloseNodes(key, constants.alpha) + if key != self.id: + # Update the "last accessed" timestamp for the appropriate k-bucket + self._routingTable.touchKBucket(key) + if len(shortlist) == 0: + # This node doesn't know of any other nodes + fakeDf = defer.Deferred() + fakeDf.callback([]) + return fakeDf + else: + # This is used during the bootstrap process; node ID's are most probably fake + shortlist = startupShortlist + + # List of active queries; len() indicates number of active probes + # - using lists for these variables, because Python doesn't allow binding a new value to a name in an enclosing (non-global) scope + activeProbes = [] + # List of contact IDs that have already been queried + alreadyContacted = [] + # Probes that were active during the previous iteration + # A list of found and known-to-be-active remote nodes + activeContacts = [] + # This should only contain one entry; the next scheduled iteration call + pendingIterationCalls = [] + prevClosestNode = [None] + findValueResult = {} + slowNodeCount = [0] + + def extendShortlist(responseTuple): + """ @type responseMsg: kademlia.msgtypes.ResponseMessage """ + # The "raw response" tuple contains the response message, and the originating address info + responseMsg = responseTuple[0] + originAddress = responseTuple[1] # tuple: (ip adress, udp port) + # Make sure the responding node is valid, and abort the operation if it isn't + if responseMsg.nodeID in activeContacts or responseMsg.nodeID == self.id: + return responseMsg.nodeID + + # Mark this node as active + if responseMsg.nodeID in shortlist: + # Get the contact information from the shortlist... + aContact = shortlist[shortlist.index(responseMsg.nodeID)] + else: + # If it's not in the shortlist; we probably used a fake ID to reach it + # - reconstruct the contact, using the real node ID this time + aContact = Contact(responseMsg.nodeID, originAddress[0], originAddress[1], self._protocol) + activeContacts.append(aContact) + # This makes sure "bootstrap"-nodes with "fake" IDs don't get queried twice + if responseMsg.nodeID not in alreadyContacted: + alreadyContacted.append(responseMsg.nodeID) + # Now grow extend the (unverified) shortlist with the returned contacts + result = responseMsg.response + #TODO: some validation on the result (for guarding against attacks) + # If we are looking for a value, first see if this result is the value + # we are looking for before treating it as a list of contact triples + if findValue == True and type(result) == dict: + # We have found the value + findValueResult[key] = result[key] + else: + if findValue == True: + # We are looking for a value, and the remote node didn't have it + # - mark it as the closest "empty" node, if it is + if 'closestNodeNoValue' in findValueResult: + if self._routingTable.distance(key, responseMsg.nodeID) < self._routingTable.distance(key, activeContacts[0].id): + findValueResult['closestNodeNoValue'] = aContact + else: + findValueResult['closestNodeNoValue'] = aContact + for contactTriple in result: + if isinstance(contactTriple, (list, tuple)) and len(contactTriple) == 3: + testContact = Contact(contactTriple[0], contactTriple[1], contactTriple[2], self._protocol) + if testContact not in shortlist: + shortlist.append(testContact) + return responseMsg.nodeID + + def removeFromShortlist(failure): + """ @type failure: twisted.python.failure.Failure """ + failure.trap(protocol.TimeoutError) + deadContactID = failure.getErrorMessage() + if deadContactID in shortlist: + shortlist.remove(deadContactID) + return deadContactID + + def cancelActiveProbe(contactID): + activeProbes.pop() + if len(activeProbes) <= constants.alpha/2 and len(pendingIterationCalls): + # Force the iteration + pendingIterationCalls[0].cancel() + del pendingIterationCalls[0] + #print 'forcing iteration =================' + searchIteration() + + # Send parallel, asynchronous FIND_NODE RPCs to the shortlist of contacts + def searchIteration(): + #print '==> searchiteration' + slowNodeCount[0] = len(activeProbes) + # Sort the discovered active nodes from closest to furthest + activeContacts.sort(lambda firstContact, secondContact, targetKey=key: cmp(self._routingTable.distance(firstContact.id, targetKey), self._routingTable.distance(secondContact.id, targetKey))) + # This makes sure a returning probe doesn't force calling this function by mistake + while len(pendingIterationCalls): + del pendingIterationCalls[0] + # See if should continue the search + if key in findValueResult: + #print '++++++++++++++ DONE (findValue found) +++++++++++++++\n\n' + outerDf.callback(findValueResult) + return + elif len(activeContacts) and findValue == False: + if (len(activeContacts) >= constants.k) or (activeContacts[0] == prevClosestNode[0] and len(activeProbes) == slowNodeCount[0]): + # TODO: Re-send the FIND_NODEs to all of the k closest nodes not already queried + # Ok, we're done; either we have accumulated k active contacts or no improvement in closestNode has been noted + #if len(activeContacts) >= constants.k: + # print '++++++++++++++ DONE (test for k active contacts) +++++++++++++++\n\n' + #else: + # print '++++++++++++++ DONE (test for closest node) +++++++++++++++\n\n' + outerDf.callback(activeContacts) + return + # The search continues... + if len(activeContacts): + prevClosestNode[0] = activeContacts[0] + contactedNow = 0 + shortlist.sort(lambda firstContact, secondContact, targetKey=key: cmp(self._routingTable.distance(firstContact.id, targetKey), self._routingTable.distance(secondContact.id, targetKey))) + # Store the current shortList length before contacting other nodes + prevShortlistLength = len(shortlist) + for contact in shortlist: + if contact.id not in alreadyContacted: + activeProbes.append(contact.id) + rpcMethod = getattr(contact, rpc) + df = rpcMethod(key, rawResponse=True) + df.addCallback(extendShortlist) + df.addErrback(removeFromShortlist) + df.addCallback(cancelActiveProbe) + alreadyContacted.append(contact.id) + contactedNow += 1 + if contactedNow == constants.alpha: + break + if len(activeProbes) > slowNodeCount[0] \ + or (len(shortlist) < constants.k and len(activeContacts) < len(shortlist) and len(activeProbes) > 0): + #print '----------- scheduling next call -------------' + # Schedule the next iteration if there are any active calls (Kademlia uses loose parallelism) + call = twisted.internet.reactor.callLater(constants.iterativeLookupDelay, searchIteration) #IGNORE:E1101 + pendingIterationCalls.append(call) + # Check for a quick contact response that made an update to the shortList + elif prevShortlistLength < len(shortlist): + # Ensure that the closest contacts are taken from the updated shortList + searchIteration() + else: + #print '++++++++++++++ DONE (logically) +++++++++++++\n\n' + # If no probes were sent, there will not be any improvement, so we're done + outerDf.callback(activeContacts) + + outerDf = defer.Deferred() + # Start the iterations + searchIteration() + return outerDf + +# def _kbucketIndex(self, key): +# """ Calculate the index of the k-bucket which is responsible for the +# specified key +# +# @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 +# """ +# distance = self._distance(self.id, key) +# bucketIndex = int(math.log(distance, 2)) +# return bucketIndex + +# 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 +# """ +# def makeIDString(distance): +# id = hex(distance)[2:] +# if id[-1] == 'L': +# id = id[:-1] +# if len(id) % 2 != 0: +# id = '0' + id +# id = id.decode('hex') +# id = (20 - len(id))*'\x00' + id +# return id +# min = math.pow(2, bucketIndex) +# max = math.pow(2, bucketIndex+1) +# distance = random.randrange(min, max) +# distanceStr = makeIDString(distance) +# randomID = makeIDString(self._distance(distanceStr, self.id)) +# return randomID + +# def _refreshKBuckets(self, startIndex=0, force=False): +# """ Refreshes all k-buckets that need refreshing, starting at the +# k-bucket with the specified index +# +# @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 +# """ +# #print '_refreshKbuckets called with index:',startIndex +# bucketIndex = [] +# bucketIndex.append(startIndex + 1) +# outerDf = defer.Deferred() +# def refreshNextKBucket(dfResult=None): +# #print ' refreshNexKbucket called; bucketindex is', bucketIndex[0] +# bucketIndex[0] += 1 +# while bucketIndex[0] < 160: +# if force or (int(time.time()) - self._buckets[bucketIndex[0]].lastAccessed >= constants.refreshTimeout): +# searchID = self._randomIDInBucketRange(bucketIndex[0]) +# self._buckets[bucketIndex[0]].lastAccessed = int(time.time()) +# #print ' refreshing bucket',bucketIndex[0] +# df = self.iterativeFindNode(searchID) +# df.addCallback(refreshNextKBucket) +# return +# else: +# bucketIndex[0] += 1 +# # If this is reached, we have refreshed all the buckets +# #print ' all buckets refreshed; initiating outer deferred callback' +# outerDf.callback(None) +# #print '_refreshKbuckets starting cycle' +# refreshNextKBucket() +# #print '_refreshKbuckets returning' +# return outerDf + + def _persistState(self, *args): + state = {'id': self.id, + 'closestNodes': self.findNode(self.id)} + now = int(time.time()) + self._dataStore.setItem('nodeState', state, now, now, self.id) + + def _refreshNode(self): + """ Periodically called to perform k-bucket refreshes and data + replication/republishing as necessary """ + #print 'refreshNode called' + df = self._refreshRoutingTable() + df.addCallback(self._republishData) + df.addCallback(self._scheduleNextNodeRefresh) + + def _refreshRoutingTable(self): + nodeIDs = self._routingTable.getRefreshList(0, False) + outerDf = defer.Deferred() + def searchForNextNodeID(dfResult=None): + if len(nodeIDs) > 0: + searchID = nodeIDs.pop() + df = self.iterativeFindNode(searchID) + df.addCallback(searchForNextNodeID) + else: + # If this is reached, we have finished refreshing the routing table + outerDf.callback(None) + # Start the refreshing cycle + searchForNextNodeID() + return outerDf + + def _republishData(self, *args): + #print '---republishData() called' + df = twisted.internet.threads.deferToThread(self._threadedRepublishData) + return df + + def _scheduleNextNodeRefresh(self, *args): + #print '==== sheduling next refresh' + twisted.internet.reactor.callLater(constants.checkRefreshInterval, self._refreshNode) + + def _threadedRepublishData(self, *args): + """ Republishes and expires any stored data (i.e. stored + C{(key, value pairs)} that need to be republished/expired + + This method should run in a deferred thread + """ + #print '== republishData called, node:',ord(self.id[0]) + expiredKeys = [] + for key in self._dataStore: + # Filter internal variables stored in the datastore + if key == 'nodeState': + continue + now = int(time.time()) + originalPublisherID = self._dataStore.originalPublisherID(key) + age = now - self._dataStore.originalPublishTime(key) + #print ' node:',ord(self.id[0]),'key:',ord(key[0]),'orig publishing time:',self._dataStore.originalPublishTime(key),'now:',now,'age:',age,'lastPublished age:',now - self._dataStore.lastPublished(key),'original pubID:', ord(originalPublisherID[0]) + if originalPublisherID == self.id: + # This node is the original publisher; it has to republish + # the data before it expires (24 hours in basic Kademlia) + if age >= constants.dataExpireTimeout: + #print ' REPUBLISHING key:', key + #self.iterativeStore(key, self._dataStore[key]) + twisted.internet.reactor.callFromThread(self.iterativeStore, key, self._dataStore[key]) + else: + # This node needs to replicate the data at set intervals, + # until it expires, without changing the metadata associated with it + # First, check if the data has expired + if age >= constants.dataExpireTimeout: + # This key/value pair has expired (and it has not been republished by the original publishing node + # - remove it + expiredKeys.append(key) + elif now - self._dataStore.lastPublished(key) >= constants.replicateInterval: + # ...data has not yet expired, and we need to replicate it + #print ' replicating key:', key,'age:',age + #self.iterativeStore(key=key, value=self._dataStore[key], originalPublisherID=originalPublisherID, age=age) + twisted.internet.reactor.callFromThread(self.iterativeStore, key=key, value=self._dataStore[key], originalPublisherID=originalPublisherID, age=age) + for key in expiredKeys: + #print ' expiring key:', key + del self._dataStore[key] + #print 'done with threadedDataRefresh()' + + +if __name__ == '__main__': + import sys + if len(sys.argv) < 2: + print 'Usage:\n%s UDP_PORT [KNOWN_NODE_IP KNOWN_NODE_PORT]' % sys.argv[0] + print 'or:\n%s UDP_PORT [FILE_WITH_KNOWN_NODES]' % sys.argv[0] + print '\nIf a file is specified, it should containg one IP address and UDP port\nper line, seperated by a space.' + sys.exit(1) + try: + usePort = int(sys.argv[1]) + except ValueError: + print '\nUDP_PORT must be an integer value.\n' + print 'Usage:\n%s UDP_PORT [KNOWN_NODE_IP KNOWN_NODE_PORT]' % sys.argv[0] + print 'or:\n%s UDP_PORT [FILE_WITH_KNOWN_NODES]' % sys.argv[0] + print '\nIf a file is specified, it should contain one IP address and UDP port\nper line, seperated by a space.' + sys.exit(1) + + if len(sys.argv) == 4: + knownNodes = [(sys.argv[2], int(sys.argv[3]))] + elif len(sys.argv) == 3: + knownNodes = [] + f = open(sys.argv[2], 'r') + lines = f.readlines() + f.close() + for line in lines: + ipAddress, udpPort = line.split() + knownNodes.append((ipAddress, int(udpPort))) + else: + knownNodes = None + + node = Node( udpPort=usePort ) + node.joinNetwork(knownNodes) + twisted.internet.reactor.run() diff --git a/kademlia/protocol.py b/kademlia/protocol.py new file mode 100644 index 0000000..54f440c --- /dev/null +++ b/kademlia/protocol.py @@ -0,0 +1,305 @@ +#!/usr/bin/env python +# +# 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 + +from twisted.internet import protocol, defer +from twisted.python import failure +import twisted.internet.reactor + +import constants +import encoding +import msgtypes +import msgformat +from contact import Contact + +reactor = twisted.internet.reactor + +class TimeoutError(Exception): + """ Raised when a RPC times out """ + +class KademliaProtocol(protocol.DatagramProtocol): + """ Implements all low-level network-related functions of a Kademlia node """ + msgSizeLimit = constants.udpDatagramMaxSize-26 + maxToSendDelay = 10**-3#0.05 + minToSendDelay = 10**-5#0.01 + + def __init__(self, node, msgEncoder=encoding.Bencode(), msgTranslator=msgformat.DefaultFormat()): + self._node = node + self._encoder = msgEncoder + self._translator = msgTranslator + self._sentMessages = {} + self._partialMessages = {} + self._partialMessagesProgress = {} + self._next = 0 + self._callLaterList = {} + + def sendRPC(self, contact, method, args, rawResponse=False): + """ Sends an RPC to the specified contact + + @param contact: The contact (remote node) to send the RPC to + @type contact: kademlia.contacts.Contact + @param method: The name of remote method to invoke + @type method: str + @param args: A list of (non-keyword) arguments to pass to the remote + method, in the correct order + @type args: tuple + @param rawResponse: If this is set to C{True}, the caller of this RPC + will receive a tuple containing the actual response + message object and the originating address tuple as + a result; in other words, it will not be + interpreted by this class. Unless something special + needs to be done with the metadata associated with + the message, this should remain C{False}. + @type rawResponse: bool + + @return: This immediately returns a deferred object, which will return + the result of the RPC call, or raise the relevant exception + if the remote node raised one. If C{rawResponse} is set to + C{True}, however, it will always return the actual response + message (which may be a C{ResponseMessage} or an + C{ErrorMessage}). + @rtype: twisted.internet.defer.Deferred + """ + msg = msgtypes.RequestMessage(self._node.id, method, args) + msgPrimitive = self._translator.toPrimitive(msg) + encodedMsg = self._encoder.encode(msgPrimitive) + + df = defer.Deferred() + if rawResponse: + df._rpcRawResponse = True + + # Set the RPC timeout timer + timeoutCall = reactor.callLater(constants.rpcTimeout, self._msgTimeout, msg.id) #IGNORE:E1101 + # Transmit the data + self._send(encodedMsg, msg.id, (contact.address, contact.port)) + self._sentMessages[msg.id] = (contact.id, df, timeoutCall) + return df + + def datagramReceived(self, datagram, address): + """ Handles and parses incoming RPC messages (and responses) + + @note: This is automatically called by Twisted when the protocol + receives a UDP datagram + """ + if datagram[0] == '\x00' and datagram[25] == '\x00': + totalPackets = (ord(datagram[1]) << 8) | ord(datagram[2]) + msgID = datagram[5:25] + seqNumber = (ord(datagram[3]) << 8) | ord(datagram[4]) + if msgID not in self._partialMessages: + self._partialMessages[msgID] = {} + self._partialMessages[msgID][seqNumber] = datagram[26:] + if len(self._partialMessages[msgID]) == totalPackets: + keys = self._partialMessages[msgID].keys() + keys.sort() + data = '' + for key in keys: + data += self._partialMessages[msgID][key] + datagram = data + del self._partialMessages[msgID] + else: + return + try: + msgPrimitive = self._encoder.decode(datagram) + except encoding.DecodeError: + # We received some rubbish here + return + + message = self._translator.fromPrimitive(msgPrimitive) + remoteContact = Contact(message.nodeID, address[0], address[1], self) + + # Refresh the remote node's details in the local node's k-buckets + self._node.addContact(remoteContact) + + if isinstance(message, msgtypes.RequestMessage): + # This is an RPC method request + self._handleRPC(remoteContact, message.id, message.request, message.args) + elif isinstance(message, msgtypes.ResponseMessage): + # Find the message that triggered this response + if self._sentMessages.has_key(message.id): + # Cancel timeout timer for this RPC + df, timeoutCall = self._sentMessages[message.id][1:3] + timeoutCall.cancel() + del self._sentMessages[message.id] + + if hasattr(df, '_rpcRawResponse'): + # The RPC requested that the raw response message and originating address be returned; do not interpret it + df.callback((message, address)) + elif isinstance(message, msgtypes.ErrorMessage): + # The RPC request raised a remote exception; raise it locally + if message.exceptionType.startswith('exceptions.'): + exceptionClassName = message.exceptionType[11:] + else: + localModuleHierarchy = self.__module__.split('.') + remoteHierarchy = message.exceptionType.split('.') + #strip the remote hierarchy + while remoteHierarchy[0] == localModuleHierarchy[0]: + remoteHierarchy.pop(0) + localModuleHierarchy.pop(0) + exceptionClassName = '.'.join(remoteHierarchy) + remoteException = None + try: + exec 'remoteException = %s("%s")' % (exceptionClassName, message.response) + except Exception: + # We could not recreate the exception; create a generic one + remoteException = Exception(message.response) + df.errback(remoteException) + else: + # We got a result from the RPC + df.callback(message.response) + else: + # If the original message isn't found, it must have timed out + #TODO: we should probably do something with this... + pass + + def _send(self, data, rpcID, address): + """ Transmit the specified data over UDP, breaking it up into several + packets if necessary + + If the data is spread over multiple UDP datagrams, the packets have the + following structure:: + | | | | | |||||||||||| 0x00 | + |Transmision|Total number|Sequence number| RPC ID |Header end| + | type ID | of packets |of this packet | | indicator| + | (1 byte) | (2 bytes) | (2 bytes) |(20 bytes)| (1 byte) | + | | | | | |||||||||||| | + + @note: The header used for breaking up large data segments will + possibly be moved out of the KademliaProtocol class in the + future, into something similar to a message translator/encoder + class (see C{kademlia.msgformat} and C{kademlia.encoding}). + """ + if len(data) > self.msgSizeLimit: + # We have to spread the data over multiple UDP datagrams, and provide sequencing information + # 1st byte is transmission type id, bytes 2 & 3 are the total number of packets in this transmission, bytes 4 & 5 are the sequence number for this specific packet + totalPackets = len(data) / self.msgSizeLimit + if len(data) % self.msgSizeLimit > 0: + totalPackets += 1 + encTotalPackets = chr(totalPackets >> 8) + chr(totalPackets & 0xff) + seqNumber = 0 + startPos = 0 + while seqNumber < totalPackets: + #reactor.iterate() #IGNORE:E1101 + packetData = data[startPos:startPos+self.msgSizeLimit] + encSeqNumber = chr(seqNumber >> 8) + chr(seqNumber & 0xff) + txData = '\x00%s%s%s\x00%s' % (encTotalPackets, encSeqNumber, rpcID, packetData) + self._sendNext(txData, address) + + startPos += self.msgSizeLimit + seqNumber += 1 + else: + self._sendNext(data, address) + + def _sendNext(self, txData, address): + """ Send the next UDP packet """ + ts = time.time() + delay = 0 + if ts >= self._next: + delay = self.minToSendDelay + self._next = ts + self.minToSendDelay + else: + delay = (self._next-ts) + self.maxToSendDelay + self._next += self.maxToSendDelay + if self.transport: + laterCall = reactor.callLater(delay, self.transport.write, txData, address) + for key in self._callLaterList.keys(): + if key <= ts: + del self._callLaterList[key] + self._callLaterList[self._next] = laterCall + + def _sendResponse(self, contact, rpcID, response): + """ Send a RPC response to the specified contact + """ + msg = msgtypes.ResponseMessage(rpcID, self._node.id, response) + msgPrimitive = self._translator.toPrimitive(msg) + encodedMsg = self._encoder.encode(msgPrimitive) + self._send(encodedMsg, rpcID, (contact.address, contact.port)) + + def _sendError(self, contact, rpcID, exceptionType, exceptionMessage): + """ Send an RPC error message to the specified contact + """ + msg = msgtypes.ErrorMessage(rpcID, self._node.id, exceptionType, exceptionMessage) + msgPrimitive = self._translator.toPrimitive(msg) + encodedMsg = self._encoder.encode(msgPrimitive) + self._send(encodedMsg, rpcID, (contact.address, contact.port)) + + def _handleRPC(self, senderContact, rpcID, method, args): + """ Executes a local function in response to an RPC request """ + # Set up the deferred callchain + def handleError(f): + self._sendError(senderContact, rpcID, f.type, f.getErrorMessage()) + + def handleResult(result): + self._sendResponse(senderContact, rpcID, result) + + df = defer.Deferred() + df.addCallback(handleResult) + df.addErrback(handleError) + + # Execute the RPC + func = getattr(self._node, method, None) + if callable(func) and hasattr(func, 'rpcmethod'): + # Call the exposed Node method and return the result to the deferred callback chain + try: + try: + # Try to pass the sender's node id to the function... + result = func(*args, **{'_rpcNodeID': senderContact.id, '_rpcNodeContact': senderContact}) + except TypeError: + # ...or simply call it if that fails + result = func(*args) + except Exception, e: + df.errback(failure.Failure(e)) + else: + df.callback(result) + else: + # No such exposed method + df.errback( failure.Failure( AttributeError('Invalid method: %s' % method) ) ) + + def _msgTimeout(self, messageID): + """ Called when an RPC request message times out """ + # Find the message that timed out + if self._sentMessages.has_key(messageID): + remoteContactID, df = self._sentMessages[messageID][0:2] + if self._partialMessages.has_key(messageID): + # We are still receiving this message + # See if any progress has been made; if not, kill the message + if self._partialMessagesProgress.has_key(messageID): + if len(self._partialMessagesProgress[messageID]) == len(self._partialMessages[messageID]): + # No progress has been made + del self._partialMessagesProgress[messageID] + del self._partialMessages[messageID] + df.errback(failure.Failure(TimeoutError(remoteContactID))) + return + # Reset the RPC timeout timer + timeoutCall = reactor.callLater(constants.rpcTimeout, self._msgTimeout, messageID) #IGNORE:E1101 + self._sentMessages[messageID] = (remoteContactID, df, timeoutCall) + return + del self._sentMessages[messageID] + # The message's destination node is now considered to be dead; + # raise an (asynchronous) TimeoutError exception and update the host node + self._node.removeContact(remoteContactID) + df.errback(failure.Failure(TimeoutError(remoteContactID))) + else: + # This should never be reached + print "ERROR: deferred timed out, but is not present in sent messages list!" + + def stopProtocol(self): + """ Called when the transport is disconnected. + + Will only be called once, after all ports are disconnected. + """ + for key in self._callLaterList.keys(): + try: + if key > time.time(): + self._callLaterList[key].cancel() + except Exception, e: + print e + del self._callLaterList[key] + #TODO: test: do we really need the reactor.iterate() call? + reactor.iterate() 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() ) |