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