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