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         # get the bucket for this node
 117         i = self. _bucketIndexForInt(node.int)
 118         ## check to see if node is in the bucket already
 119         try:
 120             it = self.buckets[i].l.index(node.int)
 121         except ValueError:
 122             ## no
 123             pass
 124         else:
 125             if contacted:
 126                 node.updateLastSeen()
 127                 # move node to end of bucket
 128                 del(self.buckets[i].l[it])
 129                 # note that we removed the original and replaced it with the new one
 130                 # utilizing this nodes new contact info
 131                 self.buckets[i].l.append(node)
 132                 self.buckets[i].touch()
 133             return
 134
 135         # we don't have this node, check to see if the bucket is full
 136         if len(self.buckets[i].l) < K:
 137             # no, append this node and return
 138             if contacted:
 139                 node.updateLastSeen()
 140             self.buckets[i].l.append(node)
 141             self.buckets[i].touch()
 142             return
 143
 144         # bucket is full, check to see if self.node is in the bucket
 145         try:
 146             me = self.buckets[i].l.index(self.node)
 147         except ValueError:
 148             return self.buckets[i].l[0]
 149
 150         ## this bucket is full and contains our node, split the bucket
 151         if len(self.buckets) >= HASH_LENGTH:
 152             # our table is FULL
 153             print "Hash Table is FULL!  Increase K!"
 154             return
 155
 156         self._splitBucket(self.buckets[i])
 157
 158         ## now that the bucket is split and balanced, try to insert the node again
 159         return self.insertNode(node)
 160
 161     def justSeenNode(self, node):
 162         """ call this any time you get a message from a node, to update it in the table if it's there """
 163         try:
 164             n = self.findNodes(node.int)[0]
 165         except IndexError:
 166             return None
 167         else:
 168             tstamp = n.lastSeen
 169             n.updateLastSeen()
 170             return tstamp
 171
 172     def nodeFailed(self, node):
 173         """ call this when a node fails to respond to a message, to invalidate that node """
 174         try:
 175             n = self.findNodes(node.int)[0]
 176         except IndexError:
 177             return None
 178         else:
 179             if(n.msgFailed() >= const.MAX_FAILURES):
 180                 self.replaceStaleNode(n, None)
 181
 182 class KBucket:
 183     __slots = ['min', 'max', 'lastAccessed']
 184     def __init__(self, contents, min, max):
 185         self.l = contents
 186         self.min = min
 187         self.max = max
 188         self.lastAccessed = time.time()
 189
 190     def touch(self):
 191         self.lastAccessed = time.time()
 192
 193     def getNodeWithInt(self, int):
 194         try:
 195             return self.l[self.l.index(int)]
 196         except IndexError:
 197             raise ValueError
 198
 199     def __repr__(self):
 200         return "<KBucket items: %d     min: %d\tmax: %d>" % (len(self.l), self.min, self.max)
 201
 202     ## comparators, necessary for bisecting list of buckets with a hash expressed as an integer or a distance
 203     ## compares integer or node object with the bucket's range
 204     def __lt__(self, a):
 205         if type(a) == InstanceType:
 206             a = a.int
 207         return self.max <= a
 208     def __le__(self, a):
 209         if type(a) == InstanceType:
 210             a = a.int
 211         return self.min < a
 212     def __gt__(self, a):
 213         if type(a) == InstanceType:
 214             a = a.int
 215         return self.min > a
 216     def __ge__(self, a):
 217         if type(a) == InstanceType:
 218             a = a.int
 219         return self.max >= a
 220     def __eq__(self, a):
 221         if type(a) == InstanceType:
 222             a = a.int
 223         return self.min <= a and self.max > a
 224     def __ne__(self, a):
 225         if type(a) == InstanceType:
 226             a = a.int
 227         return self.min >= a or self.max < a
 228
 229
 230
 231 ##############
 232 import unittest
 233
 234 class TestKTable(unittest.TestCase):
 235     def setUp(self):
 236         self.a = Node().init(hash.newID(), 'localhost', 2002)
 237         self.t = KTable(self.a)
 238
 239     def test_replace_stale_node(self):
 240         self.b = Node().init(hash.newID(), 'localhost', 2003)
 241         self.t.replaceStaleNode(self.a, self.b)
 242         assert(len(self.t.buckets[0].l) == 1)
 243         assert(self.t.buckets[0].l[0].id == self.b.id)
 244
 245 if __name__ == "__main__":
 246     unittest.main()