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         try:
 149             me = self.buckets[i].min <= self.node < self.buckets[i].max
 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         except IndexError:
 200             raise ValueError
 201
 202     def __repr__(self):
 203         return "<KBucket items: %d     min: %d\tmax: %d>" % (len(self.l), self.min, self.max)
 204
 205     ## comparators, necessary for bisecting list of buckets with a hash expressed as an integer or a distance
 206     ## compares integer or node object with the bucket's range
 207     def __lt__(self, a):
 208         if type(a) == InstanceType:
 209             a = a.int
 210         return self.max <= a
 211     def __le__(self, a):
 212         if type(a) == InstanceType:
 213             a = a.int
 214         return self.min < a
 215     def __gt__(self, a):
 216         if type(a) == InstanceType:
 217             a = a.int
 218         return self.min > a
 219     def __ge__(self, a):
 220         if type(a) == InstanceType:
 221             a = a.int
 222         return self.max >= a
 223     def __eq__(self, a):
 224         if type(a) == InstanceType:
 225             a = a.int
 226         return self.min <= a and self.max > a
 227     def __ne__(self, a):
 228         if type(a) == InstanceType:
 229             a = a.int
 230         return self.min >= a or self.max < a
 231
 232
 233
 234 ##############
 235 import unittest
 236
 237 class TestKTable(unittest.TestCase):
 238     def setUp(self):
 239         self.a = Node().init(hash.newID(), 'localhost', 2002)
 240         self.t = KTable(self.a)
 241
 242     def test_replace_stale_node(self):
 243         self.b = Node().init(hash.newID(), 'localhost', 2003)
 244         self.t.replaceStaleNode(self.a, self.b)
 245         assert(len(self.t.buckets[0].l) == 1)
 246         assert(self.t.buckets[0].l[0].id == self.b.id)
 247
 248 if __name__ == "__main__":
 249     unittest.main()