ktable.py

   1 ## Copyright 2002 Andrew Loewenstern, All Rights Reserved
   2
   3 import time
   4 from bisect import *
   5 from types import *
   6
   7 import hash
   8 import const
   9 from const import K, HASH_LENGTH
  10 from node import Node
  11
  12 class KTable:
  13     """local routing table for a kademlia like distributed hash table"""
  14     def __init__(self, node):
  15         # this is the root node, a.k.a. US!
  16         self.node = node
  17         self.buckets = [KBucket([], 0L, 2L**HASH_LENGTH)]
  18         self.insertNode(node)
  19
  20     def _bucketIndexForInt(self, int):
  21         """the index of the bucket that should hold int"""
  22         return bisect_left(self.buckets, int)
  23
  24     def findNodes(self, id):
  25         """k nodes in our own local table closest to the ID.
  26
  27         NOTE: response may actually include ourself, it's your responsibilty
  28         to not send messages to yourself if it matters."""
  29
  30         if isinstance(id, str):
  31             int = hash.intify(id)
  32         elif isinstance(id, Node):
  33             int = id.int
  34         elif isinstance(id, int) or isinstance(id, long):
  35             int = id
  36         else:
  37             raise TypeError, "findNodes requires an int, string, or Node"
  38
  39         nodes = []
  40         i = self._bucketIndexForInt(int)
  41
  42         # if this node is already in our table then return it
  43         if int in self.buckets[i].l: return [int]
  44
  45         nodes = nodes + self.buckets[i].l
  46         if len(nodes) < K:
  47             # need more nodes
  48             min = i - 1
  49             max = i + 1
  50             while len(nodes) < K and (min >= 0 or max < len(self.buckets)):
  51                 #ASw: note that this requires K be even
  52                 if min >= 0:
  53                     nodes = nodes + self.buckets[min].l
  54                 if max < len(self.buckets):
  55                     nodes = nodes + self.buckets[max].l
  56                 min = min - 1
  57                 max = max + 1
  58
  59         nodes.sort(lambda a, b, int=int: cmp(int ^ a.int, int ^ b.int))
  60         return nodes[:K]
  61
  62     def _splitBucket(self, a):
  63         diff = (a.max - a.min) / 2
  64         b = KBucket([], a.max - diff, a.max)
  65         self.buckets.insert(self.buckets.index(a.min) + 1, b)
  66         a.max = a.max - diff
  67         # transfer nodes to new bucket
  68         for anode in a.l[:]:
  69             if anode.int >= a.max:
  70                 a.l.remove(anode)
  71                 b.l.append(anode)
  72
  73     def replaceStaleNode(self, stale, new):
  74         """this is used by clients to replace a node returned by insertNode after
  75         it fails to respond to a Pong message"""
  76         i = self._bucketIndexForInt(stale.int)
  77         try:
  78             it = self.buckets[i].l.index(stale.int)
  79         except ValueError:
  80             return
  81
  82         del(self.buckets[i].l[it])
  83         if new:
  84             self.buckets[i].l.append(new)
  85
  86     def insertNode(self, node, contacted=1):
  87         """
  88         this insert the node, returning None if successful, returns the oldest node in the bucket if it's full
  89         the caller responsible for pinging the returned node and calling replaceStaleNode if it is found to be stale!!
  90         contacted means that yes, we contacted THEM and we know the node is available
  91         """
  92         assert node.id != " "*20
  93         if node.id == self.node.id: return
  94         # get the bucket for this node
  95         i = self. _bucketIndexForInt(node.int)
  96         # check to see if node is in the bucket already
  97         try:
  98             it = self.buckets[i].l.index(node.int)
  99         except ValueError:
 100             # no
 101             pass
 102         else:
 103             if contacted:
 104                 node.updateLastSeen()
 105                 # move node to end of bucket
 106                 xnode = self.buckets[i].l[it]
 107                 del(self.buckets[i].l[it])
 108                 # note that we removed the original and replaced it with the new one
 109                 # utilizing this nodes new contact info
 110                 self.buckets[i].l.append(xnode)
 111                 self.buckets[i].touch()
 112             return
 113
 114         # we don't have this node, check to see if the bucket is full
 115         if len(self.buckets[i].l) < K:
 116             # no, append this node and return
 117             if contacted:
 118                 node.updateLastSeen()
 119             self.buckets[i].l.append(node)
 120             self.buckets[i].touch()
 121             return
 122
 123         # bucket is full, check to see if self.node is in the bucket
 124         if not (self.buckets[i].min <= self.node < self.buckets[i].max):
 125             return self.buckets[i].l[0]
 126
 127         # this bucket is full and contains our node, split the bucket
 128         if len(self.buckets) >= HASH_LENGTH:
 129             # our table is FULL, this is really unlikely
 130             print "Hash Table is FULL!  Increase K!"
 131             return
 132
 133         self._splitBucket(self.buckets[i])
 134
 135         # now that the bucket is split and balanced, try to insert the node again
 136         return self.insertNode(node)
 137
 138     def justSeenNode(self, node):
 139         """call this any time you get a message from a node
 140         it will update it in the table if it's there """
 141         try:
 142             n = self.findNodes(node.int)[0]
 143         except IndexError:
 144             return None
 145         else:
 146             tstamp = n.lastSeen
 147             n.updateLastSeen()
 148             return tstamp
 149
 150     def nodeFailed(self, node):
 151         """ call this when a node fails to respond to a message, to invalidate that node """
 152         try:
 153             n = self.findNodes(node.int)[0]
 154         except IndexError:
 155             return None
 156         else:
 157             if n.msgFailed() >= const.MAX_FAILURES:
 158                 self.replaceStaleNode(n, None)
 159
 160 class KBucket:
 161     __slots = ['min', 'max', 'lastAccessed']
 162     def __init__(self, contents, min, max):
 163         self.l = contents
 164         self.min = min
 165         self.max = max
 166         self.lastAccessed = time.time()
 167
 168     def touch(self):
 169         self.lastAccessed = time.time()
 170
 171     def getNodeWithInt(self, int):
 172         if int in self.l: return int
 173         else: raise ValueError
 174
 175     def __repr__(self):
 176         return "<KBucket %d items (%d to %d)>" % (len(self.l), self.min, self.max)
 177
 178     ## Comparators
 179     # necessary for bisecting list of buckets with a hash expressed as an integer or a distance
 180     # compares integer or node object with the bucket's range
 181     def __lt__(self, a):
 182         if isinstance(a, Node): a = a.int
 183         return self.max <= a
 184     def __le__(self, a):
 185         if isinstance(a, Node): a = a.int
 186         return self.min < a
 187     def __gt__(self, a):
 188         if isinstance(a, Node): a = a.int
 189         return self.min > a
 190     def __ge__(self, a):
 191         if isinstance(a, Node): a = a.int
 192         return self.max >= a
 193     def __eq__(self, a):
 194         if isinstance(a, Node): a = a.int
 195         return self.min <= a and self.max > a
 196     def __ne__(self, a):
 197         if isinstance(a, Node): a = a.int
 198         return self.min >= a or self.max < a
 199
 200
 201 ### UNIT TESTS ###
 202 import unittest
 203
 204 class TestKTable(unittest.TestCase):
 205     def setUp(self):
 206         self.a = Node().init(hash.newID(), 'localhost', 2002)
 207         self.t = KTable(self.a)
 208         print self.t.buckets[0].l
 209
 210     def test_replace_stale_node(self):
 211         self.b = Node().init(hash.newID(), 'localhost', 2003)
 212         self.t.replaceStaleNode(self.a, self.b)
 213         assert len(self.t.buckets[0].l) == 1
 214         assert self.t.buckets[0].l[0].id == self.b.id
 215
 216 if __name__ == "__main__":
 217     unittest.main()