ktable.py

   1 ## Copyright 2002 Andrew Loewenstern, All Rights Reserved
   2
   3 import hash
   4 from bisect import *
   5 import time
   6 from types import *
   7
   8 import const
   9 from node import Node
  10
  11 # The all-powerful, magical Kademlia "k" constant, bucket depth
  12 K = 8
  13
  14 # how many bits wide is our hash?
  15 HASH_LENGTH = 160
  16
  17
  18 # the local routing table for a kademlia like distributed hash table
  19 class KTable:
  20     def __init__(self, node):
  21         # this is the root node, a.k.a. US!
  22         self.node = node
  23         self.buckets = [KBucket([], 0L, 2L**HASH_LENGTH)]
  24         self.insertNode(node)
  25
  26     def _bucketIndexForInt(self, int):
  27         """returns the index of the bucket that should hold int"""
  28         return bisect_left(self.buckets, int)
  29
  30     def findNodes(self, id):
  31         """ return k nodes in our own local table closest to the ID
  32             ignoreSelf means we will return K closest nodes to ourself if we search for our own ID
  33             note, K closest nodes may actually include ourself, it's the callers responsibilty to
  34             not send messages to itself if it matters
  35         """
  36         if type(id) == StringType:
  37             int = hash.intify(id)
  38         elif type(id) == InstanceType:
  39             int = id.int
  40         elif type(id) == IntType or type(id) == LongType:
  41             int = id
  42         else:
  43             raise TypeError, "findLocalNodes requires an int, string, or Node instance type"
  44
  45         nodes = []
  46
  47         def sort(a, b, int=int):
  48             """ this function is for sorting nodes relative to the ID we are looking for """
  49             x, y = int ^ a.int, int ^ b.int
  50             if x > y:
  51                 return 1
  52             elif x < y:
  53                 return -1
  54             return 0
  55
  56         i = self._bucketIndexForInt(int)
  57
  58         ## see if this node is already in our table and return it
  59         try:
  60             index = self.buckets[i].l.index(int)
  61         except ValueError:
  62             pass
  63         else:
  64             self.buckets[i].touch()
  65             return [self.buckets[i].l[index]]
  66
  67         nodes = nodes + self.buckets[i].l
  68         if len(nodes) == K:
  69             nodes.sort(sort)
  70             return nodes
  71         else:
  72             # need more nodes
  73             min = i - 1
  74             max = i + 1
  75             while (len(nodes) < K and (min >= 0 and max < len(self.buckets))):
  76                 if min >= 0:
  77                     nodes = nodes + self.buckets[min].l
  78                     self.buckets[min].touch()
  79                 if max < len(self.buckets):
  80                     nodes = nodes + self.buckets[max].l
  81                     self.buckets[max].touch()
  82
  83             nodes.sort(sort)
  84             return nodes[:K-1]
  85
  86     def _splitBucket(self, a):
  87         diff = (a.max - a.min) / 2
  88         b = KBucket([], a.max - diff, a.max)
  89         self.buckets.insert(self.buckets.index(a.min) + 1, b)
  90         a.max = a.max - diff
  91         # transfer nodes to new bucket
  92         for anode in a.l[:]:
  93             if anode.int >= a.max:
  94                 a.l.remove(anode)
  95                 b.l.append(anode)
  96
  97     def replaceStaleNode(self, stale, new):
  98         """ this is used by clients to replace a node returned by insertNode after
  99             it fails to respond to a Pong message
 100         """
 101         i = self._bucketIndexForInt(stale.int)
 102         try:
 103             it = self.buckets[i].l.index(stale.int)
 104         except ValueError:
 105             return
 106
 107         del(self.buckets[i].l[it])
 108         if new:
 109             self.buckets[i].l.append(new)
 110
 111     def insertNode(self, node):
 112         """
 113         this insert the node, returning None if successful, returns the oldest node in the bucket if it's full
 114         the caller responsible for pinging the returned node and calling replaceStaleNode if it is found to be stale!!
 115         """
 116         assert(node.id != " "*20)
 117         # get the bucket for this node
 118         i = self. _bucketIndexForInt(node.int)
 119         ## check to see if node is in the bucket already
 120         try:
 121             it = self.buckets[i].l.index(node.int)
 122         except ValueError:
 123             ## no
 124             pass
 125         else:
 126             node.updateLastSeen()
 127             # move node to end of bucket
 128             del(self.buckets[i].l[it])
 129             self.buckets[i].l.append(node)
 130             self.buckets[i].touch()
 131             return
 132
 133         # we don't have this node, check to see if the bucket is full
 134         if len(self.buckets[i].l) < K:
 135             # no, append this node and return
 136             self.buckets[i].l.append(node)
 137             self.buckets[i].touch()
 138             return
 139
 140         # bucket is full, check to see if self.node is in the bucket
 141         try:
 142             me = self.buckets[i].l.index(self.node)
 143         except ValueError:
 144             return self.buckets[i].l[0]
 145
 146         ## this bucket is full and contains our node, split the bucket
 147         if len(self.buckets) >= HASH_LENGTH:
 148             # our table is FULL
 149             print "Hash Table is FULL!  Increase K!"
 150             return
 151
 152         self._splitBucket(self.buckets[i])
 153
 154         ## now that the bucket is split and balanced, try to insert the node again
 155         return self.insertNode(node)
 156
 157     def justSeenNode(self, node):
 158         """ call this any time you get a message from a node, to update it in the table if it's there """
 159         try:
 160             n = self.findNodes(node.int)[0]
 161         except IndexError:
 162             return None
 163         else:
 164             tstamp = n.lastSeen
 165             n.updateLastSeen()
 166             return tstamp
 167
 168     def nodeFailed(self, node):
 169         """ call this when a node fails to respond to a message, to invalidate that node """
 170         try:
 171             n = self.findNodes(node.int)[0]
 172         except IndexError:
 173             return None
 174         else:
 175             if(n.msgFailed() >= const.MAX_FAILURES):
 176                 self.replaceStaleNode(n, None)
 177
 178 class KBucket:
 179     __slots = ['min', 'max', 'lastAccessed']
 180     def __init__(self, contents, min, max):
 181         self.l = contents
 182         self.min = min
 183         self.max = max
 184         self.lastAccessed = time.time()
 185
 186     def touch(self):
 187         self.lastAccessed = time.time()
 188
 189     def getNodeWithInt(self, int):
 190         try:
 191             return self.l[self.l.index(int)]
 192             self.touch()
 193         except IndexError:
 194             raise ValueError
 195
 196     def __repr__(self):
 197         return "<KBucket items: %d     min: %d\tmax: %d>" % (len(self.l), self.min, self.max)
 198
 199     ## comparators, necessary for bisecting list of buckets with a hash expressed as an integer or a distance
 200     ## compares integer or node object with the bucket's range
 201     def __lt__(self, a):
 202         if type(a) == InstanceType:
 203             a = a.int
 204         return self.max <= a
 205     def __le__(self, a):
 206         if type(a) == InstanceType:
 207             a = a.int
 208         return self.min < a
 209     def __gt__(self, a):
 210         if type(a) == InstanceType:
 211             a = a.int
 212         return self.min > a
 213     def __ge__(self, a):
 214         if type(a) == InstanceType:
 215             a = a.int
 216         return self.max >= a
 217     def __eq__(self, a):
 218         if type(a) == InstanceType:
 219             a = a.int
 220         return self.min <= a and self.max > a
 221     def __ne__(self, a):
 222         if type(a) == InstanceType:
 223             a = a.int
 224         return self.min >= a or self.max < a
 225
 226
 227
 228 ##############
 229 import unittest
 230
 231 class TestKTable(unittest.TestCase):
 232     def setUp(self):
 233         self.a = Node().init(hash.newID(), 'localhost', 2002)
 234         self.t = KTable(self.a)
 235
 236     def test_replace_stale_node(self):
 237         self.b = Node().init(hash.newID(), 'localhost', 2003)
 238         self.t.replaceStaleNode(self.a, self.b)
 239         assert(len(self.t.buckets[0].l) == 1)
 240         assert(self.t.buckets[0].l[0].id == self.b.id)
 241
 242 if __name__ == "__main__":
 243     unittest.main()