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             return [self.buckets[i].l[index]]
  65
  66         nodes = nodes + self.buckets[i].l
  67         if len(nodes) >= K:
  68             nodes.sort(sort)
  69             return nodes[:K]
  70         else:
  71             # need more nodes
  72             min = i - 1
  73             max = i + 1
  74             while (len(nodes) < K and (min >= 0 or max < len(self.buckets))):
  75                 if min >= 0:
  76                     nodes = nodes + self.buckets[min].l
  77                 if max < len(self.buckets):
  78                     nodes = nodes + self.buckets[max].l
  79                 min = min - 1
  80                 max = max + 1
  81             nodes.sort(sort)
  82             return nodes[:K]
  83
  84     def _splitBucket(self, a):
  85         diff = (a.max - a.min) / 2
  86         b = KBucket([], a.max - diff, a.max)
  87         self.buckets.insert(self.buckets.index(a.min) + 1, b)
  88         a.max = a.max - diff
  89         # transfer nodes to new bucket
  90         for anode in a.l[:]:
  91             if anode.int >= a.max:
  92                 a.l.remove(anode)
  93                 b.l.append(anode)
  94
  95     def replaceStaleNode(self, stale, new):
  96         """ this is used by clients to replace a node returned by insertNode after
  97             it fails to respond to a Pong message
  98         """
  99         i = self._bucketIndexForInt(stale.int)
 100         try:
 101             it = self.buckets[i].l.index(stale.int)
 102         except ValueError:
 103             return
 104
 105         del(self.buckets[i].l[it])
 106         if new:
 107             self.buckets[i].l.append(new)
 108
 109     def insertNode(self, node, contacted=1):
 110         """
 111         this insert the node, returning None if successful, returns the oldest node in the bucket if it's full
 112         the caller responsible for pinging the returned node and calling replaceStaleNode if it is found to be stale!!
 113         contacted means that yes, we contacted THEM and we know the node is available
 114         """
 115         assert(node.id != " "*20)
 116         if node.id == self.node.id:
 117             return
 118         # get the bucket for this node
 119         i = self. _bucketIndexForInt(node.int)
 120         ## check to see if node is in the bucket already
 121         try:
 122             it = self.buckets[i].l.index(node.int)
 123         except ValueError:
 124             ## no
 125             pass
 126         else:
 127             if contacted:
 128                 node.updateLastSeen()
 129                 # move node to end of bucket
 130                 xnode = self.buckets[i].l[it]
 131                 del(self.buckets[i].l[it])
 132                 # note that we removed the original and replaced it with the new one
 133                 # utilizing this nodes new contact info
 134                 self.buckets[i].l.append(xnode)
 135                 self.buckets[i].touch()
 136             return
 137
 138         # we don't have this node, check to see if the bucket is full
 139         if len(self.buckets[i].l) < K:
 140             # no, append this node and return
 141             if contacted:
 142                 node.updateLastSeen()
 143             self.buckets[i].l.append(node)
 144             self.buckets[i].touch()
 145             return
 146
 147         # bucket is full, check to see if self.node is in the bucket
 148         if not (self.buckets[i].min <= self.node < self.buckets[i].max):
 149             return self.buckets[i].l[0]
 150
 151         ## this bucket is full and contains our node, split the bucket
 152         if len(self.buckets) >= HASH_LENGTH:
 153             # our table is FULL
 154             print "Hash Table is FULL!  Increase K!"
 155             return
 156
 157         self._splitBucket(self.buckets[i])
 158
 159         ## now that the bucket is split and balanced, try to insert the node again
 160         return self.insertNode(node)
 161
 162     def justSeenNode(self, node):
 163         """ call this any time you get a message from a node, to update it in the table if it's there """
 164         try:
 165             n = self.findNodes(node.int)[0]
 166         except IndexError:
 167             return None
 168         else:
 169             tstamp = n.lastSeen
 170             n.updateLastSeen()
 171             return tstamp
 172
 173     def nodeFailed(self, node):
 174         """ call this when a node fails to respond to a message, to invalidate that node """
 175         try:
 176             n = self.findNodes(node.int)[0]
 177         except IndexError:
 178             return None
 179         else:
 180             if(n.msgFailed() >= const.MAX_FAILURES):
 181                 self.replaceStaleNode(n, None)
 182
 183 class KBucket:
 184     __slots = ['min', 'max', 'lastAccessed']
 185     def __init__(self, contents, min, max):
 186         self.l = contents
 187         self.min = min
 188         self.max = max
 189         self.lastAccessed = time.time()
 190
 191     def touch(self):
 192         self.lastAccessed = time.time()
 193
 194     def getNodeWithInt(self, int):
 195         try:
 196             return self.l[self.l.index(int)]
 197         except IndexError:
 198             raise ValueError
 199
 200     def __repr__(self):
 201         return "<KBucket items: %d     min: %d\tmax: %d>" % (len(self.l), self.min, self.max)
 202
 203     ## comparators, necessary for bisecting list of buckets with a hash expressed as an integer or a distance
 204     ## compares integer or node object with the bucket's range
 205     def __lt__(self, a):
 206         if type(a) == InstanceType:
 207             a = a.int
 208         return self.max <= a
 209     def __le__(self, a):
 210         if type(a) == InstanceType:
 211             a = a.int
 212         return self.min < a
 213     def __gt__(self, a):
 214         if type(a) == InstanceType:
 215             a = a.int
 216         return self.min > a
 217     def __ge__(self, a):
 218         if type(a) == InstanceType:
 219             a = a.int
 220         return self.max >= a
 221     def __eq__(self, a):
 222         if type(a) == InstanceType:
 223             a = a.int
 224         return self.min <= a and self.max > a
 225     def __ne__(self, a):
 226         if type(a) == InstanceType:
 227             a = a.int
 228         return self.min >= a or self.max < a
 229
 230
 231
 232 ##############
 233 import unittest
 234
 235 class TestKTable(unittest.TestCase):
 236     def setUp(self):
 237         self.a = Node().init(hash.newID(), 'localhost', 2002)
 238         self.t = KTable(self.a)
 239
 240     def test_replace_stale_node(self):
 241         self.b = Node().init(hash.newID(), 'localhost', 2003)
 242         self.t.replaceStaleNode(self.a, self.b)
 243         assert(len(self.t.buckets[0].l) == 1)
 244         assert(self.t.buckets[0].l[0].id == self.b.id)
 245
 246 if __name__ == "__main__":
 247     unittest.main()