khashmir.py

   1 ## Copyright 2002-2003 Andrew Loewenstern, All Rights Reserved
   2 # see LICENSE.txt for license information
   3
   4 from const import reactor
   5 import const
   6
   7 import time
   8
   9 from sha import sha
  10
  11 from ktable import KTable, K
  12 from knode import KNode as Node
  13
  14 from khash import newID, newIDInRange
  15
  16 from actions import FindNode, GetValue, KeyExpirer, StoreValue
  17 import krpc
  18
  19 from twisted.internet.defer import Deferred
  20 from twisted.internet import protocol
  21 from twisted.python import threadable
  22 from twisted.application import service, internet
  23 from twisted.web import server
  24 threadable.init()
  25 import sys
  26
  27 import sqlite  ## find this at http://pysqlite.sourceforge.net/
  28
  29 class KhashmirDBExcept(Exception):
  30     pass
  31
  32 # this is the main class!
  33 class Khashmir(protocol.Factory):
  34     __slots__ = ('listener', 'node', 'table', 'store', 'app', 'last', 'protocol')
  35     def __init__(self, host, port, db='khashmir.db'):
  36         self.setup(host, port, db)
  37
  38     def setup(self, host, port, db='khashmir.db'):
  39         self._findDB(db)
  40         self.port = port
  41         self.node = self._loadSelfNode(host, port)
  42         self.table = KTable(self.node)
  43         self.app = service.Application("krpc")
  44         self.udp = krpc.hostbroker(self)
  45         self.udp.protocol = krpc.KRPC
  46         self.listenport = reactor.listenUDP(port, self.udp)
  47         self.last = time.time()
  48         self._loadRoutingTable()
  49         KeyExpirer(store=self.store)
  50         #self.refreshTable(force=1)
  51         reactor.callLater(60, self.checkpoint, (1,))
  52
  53     def __del__(self):
  54         self.listenport.stopListening()
  55
  56     def _loadSelfNode(self, host, port):
  57         c = self.store.cursor()
  58         c.execute('select id from self where num = 0;')
  59         if c.rowcount > 0:
  60             id = c.fetchone()[0]
  61         else:
  62             id = newID()
  63         return Node().init(id, host, port)
  64
  65     def _saveSelfNode(self):
  66         self.store.autocommit = 0
  67         c = self.store.cursor()
  68         c.execute('delete from self where num = 0;')
  69         c.execute("insert into self values (0, %s);", sqlite.encode(self.node.id))
  70         self.store.commit()
  71         self.store.autocommit = 1
  72
  73     def checkpoint(self, auto=0):
  74         self._saveSelfNode()
  75         self._dumpRoutingTable()
  76         if auto:
  77             reactor.callLater(const.CHECKPOINT_INTERVAL, self.checkpoint)
  78
  79     def _findDB(self, db):
  80         import os
  81         try:
  82             os.stat(db)
  83         except OSError:
  84             self._createNewDB(db)
  85         else:
  86             self._loadDB(db)
  87
  88     def _loadDB(self, db):
  89         try:
  90             self.store = sqlite.connect(db=db)
  91             self.store.autocommit = 1
  92         except:
  93             import traceback
  94             raise KhashmirDBExcept, "Couldn't open DB", traceback.exc_traceback
  95
  96     def _createNewDB(self, db):
  97         self.store = sqlite.connect(db=db)
  98         self.store.autocommit = 1
  99         s = """
 100             create table kv (key binary, value binary, time timestamp, primary key (key, value));
 101             create index kv_key on kv(key);
 102             create index kv_timestamp on kv(time);
 103
 104             create table nodes (id binary primary key, host text, port number);
 105
 106             create table self (num number primary key, id binary);
 107             """
 108         c = self.store.cursor()
 109         c.execute(s)
 110
 111     def _dumpRoutingTable(self):
 112         """
 113             save routing table nodes to the database
 114         """
 115         self.store.autocommit = 0;
 116         c = self.store.cursor()
 117         c.execute("delete from nodes where id not NULL;")
 118         for bucket in self.table.buckets:
 119             for node in bucket.l:
 120                 d = node.senderDict()
 121                 c.execute("insert into nodes values (%s, %s, %s);", (sqlite.encode(d['id']), d['host'], d['port']))
 122         self.store.commit()
 123         self.store.autocommit = 1;
 124
 125     def _loadRoutingTable(self):
 126         """
 127             load routing table nodes from database
 128             it's usually a good idea to call refreshTable(force=1) after loading the table
 129         """
 130         c = self.store.cursor()
 131         c.execute("select * from nodes;")
 132         for rec in c.fetchall():
 133             n = Node().initWithDict({'id':rec[0], 'host':rec[1], 'port':int(rec[2])})
 134             n.conn = self.udp.connectionForAddr((n.host, n.port))
 135             self.table.insertNode(n, contacted=0)
 136
 137
 138     #######
 139     #######  LOCAL INTERFACE    - use these methods!
 140     def addContact(self, host, port, callback=None):
 141         """
 142             ping this node and add the contact info to the table on pong!
 143         """
 144         n =Node().init(const.NULL_ID, host, port)
 145         n.conn = self.udp.connectionForAddr((n.host, n.port))
 146         self.sendPing(n, callback=callback)
 147
 148     ## this call is async!
 149     def findNode(self, id, callback, errback=None):
 150         """ returns the contact info for node, or the k closest nodes, from the global table """
 151         # get K nodes out of local table/cache, or the node we want
 152         nodes = self.table.findNodes(id)
 153         d = Deferred()
 154         if errback:
 155             d.addCallbacks(callback, errback)
 156         else:
 157             d.addCallback(callback)
 158         if len(nodes) == 1 and nodes[0].id == id :
 159             d.callback(nodes)
 160         else:
 161             # create our search state
 162             state = FindNode(self, id, d.callback)
 163             reactor.callFromThread(state.goWithNodes, nodes)
 164
 165
 166     ## also async
 167     def valueForKey(self, key, callback, searchlocal = 1):
 168         """ returns the values found for key in global table
 169             callback will be called with a list of values for each peer that returns unique values
 170             final callback will be an empty list - probably should change to 'more coming' arg
 171         """
 172         nodes = self.table.findNodes(key)
 173
 174         # get locals
 175         if searchlocal:
 176             l = self.retrieveValues(key)
 177             if len(l) > 0:
 178                 reactor.callLater(0, callback, (l))
 179         else:
 180             l = []
 181
 182         # create our search state
 183         state = GetValue(self, key, callback)
 184         reactor.callFromThread(state.goWithNodes, nodes, l)
 185
 186     ## async, callback indicates nodes we got a response from (but no guarantee they didn't drop it on the floor)
 187     def storeValueForKey(self, key, value, callback=None):
 188         """ stores the value for key in the global table, returns immediately, no status
 189             in this implementation, peers respond but don't indicate status to storing values
 190             a key can have many values
 191         """
 192         def _storeValueForKey(nodes, key=key, value=value, response=callback , table=self.table):
 193             if not response:
 194                 # default callback
 195                 def _storedValueHandler(sender):
 196                     pass
 197                 response=_storedValueHandler
 198             action = StoreValue(self.table, key, value, response)
 199             reactor.callFromThread(action.goWithNodes, nodes)
 200
 201         # this call is asynch
 202         self.findNode(key, _storeValueForKey)
 203
 204
 205     def insertNode(self, n, contacted=1):
 206         """
 207         insert a node in our local table, pinging oldest contact in bucket, if necessary
 208
 209         If all you have is a host/port, then use addContact, which calls this method after
 210         receiving the PONG from the remote node.  The reason for the seperation is we can't insert
 211         a node into the table without it's peer-ID.  That means of course the node passed into this
 212         method needs to be a properly formed Node object with a valid ID.
 213         """
 214         old = self.table.insertNode(n, contacted=contacted)
 215         if old and (time.time() - old.lastSeen) > const.MIN_PING_INTERVAL and old.id != self.node.id:
 216             # the bucket is full, check to see if old node is still around and if so, replace it
 217
 218             ## these are the callbacks used when we ping the oldest node in a bucket
 219             def _staleNodeHandler(oldnode=old, newnode = n):
 220                 """ called if the pinged node never responds """
 221                 self.table.replaceStaleNode(old, newnode)
 222
 223             def _notStaleNodeHandler(dict, old=old):
 224                 """ called when we get a pong from the old node """
 225                 dict = dict['rsp']
 226                 sender = dict['sender']
 227                 if sender['id'] == old.id:
 228                     self.table.justSeenNode(old.id)
 229
 230             df = old.ping(self.node.senderDict())
 231             df.addCallbacks(_notStaleNodeHandler, _staleNodeHandler)
 232
 233     def sendPing(self, node, callback=None):
 234         """
 235             ping a node
 236         """
 237         df = node.ping(self.node.senderDict())
 238         ## these are the callbacks we use when we issue a PING
 239         def _pongHandler(dict, node=node, table=self.table, callback=callback):
 240             _krpc_sender = dict['_krpc_sender']
 241             dict = dict['rsp']
 242             sender = dict['sender']
 243             if node.id != const.NULL_ID and node.id != sender['id']:
 244                 # whoah, got response from different peer than we were expecting
 245                 self.table.invalidateNode(node)
 246             else:
 247                 sender['host'] = _krpc_sender[0]
 248                 sender['port'] = _krpc_sender[1]
 249                 n = Node().initWithDict(sender)
 250                 n.conn = self.udp.connectionForAddr((n.host, n.port))
 251                 table.insertNode(n)
 252                 if callback:
 253                     callback()
 254         def _defaultPong(err, node=node, table=self.table, callback=callback):
 255             table.nodeFailed(node)
 256             if callback:
 257                 callback()
 258
 259         df.addCallbacks(_pongHandler,_defaultPong)
 260
 261     def findCloseNodes(self, callback=lambda a: None):
 262         """
 263             This does a findNode on the ID one away from our own.
 264             This will allow us to populate our table with nodes on our network closest to our own.
 265             This is called as soon as we start up with an empty table
 266         """
 267         id = self.node.id[:-1] + chr((ord(self.node.id[-1]) + 1) % 256)
 268         self.findNode(id, callback)
 269
 270     def refreshTable(self, force=0):
 271         """
 272             force=1 will refresh table regardless of last bucket access time
 273         """
 274         def callback(nodes):
 275             pass
 276
 277         for bucket in self.table.buckets:
 278             if force or (time.time() - bucket.lastAccessed >= const.BUCKET_STALENESS):
 279                 id = newIDInRange(bucket.min, bucket.max)
 280                 self.findNode(id, callback)
 281
 282
 283     def retrieveValues(self, key):
 284         c = self.store.cursor()
 285         c.execute("select value from kv where key = %s;", sqlite.encode(key))
 286         t = c.fetchone()
 287         l = []
 288         while t:
 289             l.append(t['value'])
 290             t = c.fetchone()
 291         return l
 292
 293     #####
 294     ##### INCOMING MESSAGE HANDLERS
 295
 296     def krpc_ping(self, sender, _krpc_sender):
 297         """
 298             takes sender dict = {'id', <id>, 'port', port} optional keys = 'ip'
 299             returns sender dict
 300         """
 301         sender['host'] = _krpc_sender[0]
 302         sender['port'] = _krpc_sender[1]
 303         n = Node().initWithDict(sender)
 304         n.conn = self.udp.connectionForAddr((n.host, n.port))
 305         self.insertNode(n, contacted=0)
 306         return {"sender" : self.node.senderDict()}
 307
 308     def krpc_find_node(self, target, sender, _krpc_sender):
 309         nodes = self.table.findNodes(target)
 310         nodes = map(lambda node: node.senderDict(), nodes)
 311         sender['host'] = _krpc_sender[0]
 312         sender['port'] = _krpc_sender[1]
 313         n = Node().initWithDict(sender)
 314         n.conn = self.udp.connectionForAddr((n.host, n.port))
 315         self.insertNode(n, contacted=0)
 316         return {"nodes" : nodes, "sender" : self.node.senderDict()}
 317
 318     def krpc_store_value(self, key, value, sender, _krpc_sender):
 319         t = "%0.6f" % time.time()
 320         c = self.store.cursor()
 321         try:
 322             c.execute("insert into kv values (%s, %s, %s);", (sqlite.encode(key), sqlite.encode(value), t))
 323         except sqlite.IntegrityError, reason:
 324             # update last insert time
 325             c.execute("update kv set time = %s where key = %s and value = %s;", (t, sqlite.encode(key), sqlite.encode(value)))
 326         sender['host'] = _krpc_sender[0]
 327         sender['port'] = _krpc_sender[1]
 328         n = Node().initWithDict(sender)
 329         n.conn = self.udp.connectionForAddr((n.host, n.port))
 330         self.insertNode(n, contacted=0)
 331         return {"sender" : self.node.senderDict()}
 332
 333     def krpc_find_value(self, key, sender, _krpc_sender):
 334         sender['host'] = _krpc_sender[0]
 335         sender['port'] = _krpc_sender[1]
 336         n = Node().initWithDict(sender)
 337         n.conn = self.udp.connectionForAddr((n.host, n.port))
 338         self.insertNode(n, contacted=0)
 339
 340         l = self.retrieveValues(key)
 341         if len(l) > 0:
 342             return {'values' : l, "sender": self.node.senderDict()}
 343         else:
 344             nodes = self.table.findNodes(key)
 345             nodes = map(lambda node: node.senderDict(), nodes)
 346             return {'nodes' : nodes, "sender": self.node.senderDict()}
 347