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