X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=khashmir.py;h=ca6c3d0d1d5cd7c15e0b56f84b127615503d308b;hb=2f1a76beca50cc9600df404c927753b11f94c87e;hp=7eec1c80fe1bc0b843a29a6fbaca792176f31a7c;hpb=9c09fb6ef6a0668bcc5d532e07641ed27fd2ab84;p=quix0rs-apt-p2p.git diff --git a/khashmir.py b/khashmir.py index 7eec1c8..ca6c3d0 100644 --- a/khashmir.py +++ b/khashmir.py @@ -1,6 +1,8 @@ ## Copyright 2002 Andrew Loewenstern, All Rights Reserved from const import reactor +import const + import time from pickle import loads, dumps from sha import sha @@ -23,9 +25,6 @@ from bsddb3._db import DBNotFoundError from xmlrpclib import Binary -# don't ping unless it's been at least this many seconds since we've heard from a peer -MAX_PING_INTERVAL = 60 * 15 # fifteen minutes - # this is the main class! @@ -92,11 +91,16 @@ class Khashmir(xmlrpc.XMLRPC): def valueForKey(self, key, callback): """ returns the values found for key in global table """ nodes = self.table.findNodes(key) - # decode values, they will be base64 encoded + + # get locals + l = self.retrieveValues(key) + if len(l) > 0: + reactor.callFromThread(callback, l) + # create our search state state = GetValue(self, key, callback) - reactor.callFromThread(state.goWithNodes, nodes) - + reactor.callFromThread(state.goWithNodes, nodes, l) + ## async, but in the current implementation there is no guarantee a store does anything so there is no callback right now @@ -106,21 +110,27 @@ class Khashmir(xmlrpc.XMLRPC): values are stored in peers on a first-come first-served basis this will probably change so more than one value can be stored under a key """ - def _storeValueForKey(nodes, key=key, value=value, response=callback , default= lambda t: "didn't respond"): + def _storeValueForKey(nodes, key=key, value=value, response=callback , table=self.table): if not callback: # default callback - this will get called for each successful store value def _storedValueHandler(sender): pass response=_storedValueHandler + for node in nodes: + def cb(t, table = table, node=node, resp=response): + self.table.insertNode(node) + response(t) if node.id != self.node.id: + def default(err, node=node, table=table): + table.nodeFailed(node) df = node.storeValue(key, value, self.node.senderDict()) - df.addCallbacks(response, default) + df.addCallback(cb) # this call is asynch self.findNode(key, _storeValueForKey) - def insertNode(self, n): + def insertNode(self, n, contacted=1): """ insert a node in our local table, pinging oldest contact in bucket, if necessary @@ -129,8 +139,8 @@ class Khashmir(xmlrpc.XMLRPC): a node into the table without it's peer-ID. That means of course the node passed into this method needs to be a properly formed Node object with a valid ID. """ - old = self.table.insertNode(n) - if old and (time.time() - old.lastSeen) > MAX_PING_INTERVAL and old.id != self.node.id: + old = self.table.insertNode(n, contacted=contacted) + if old and (time.time() - old.lastSeen) > const.MIN_PING_INTERVAL and old.id != self.node.id: # the bucket is full, check to see if old node is still around and if so, replace it ## these are the callbacks used when we ping the oldest node in a bucket @@ -139,10 +149,12 @@ class Khashmir(xmlrpc.XMLRPC): self.table.replaceStaleNode(old, newnode) def _notStaleNodeHandler(sender, old=old): - """ called when we get a ping from the remote node """ + """ called when we get a pong from the old node """ + sender, conn = sender + sender['host'] = conn['host'] sender = Node().initWithDict(sender) if sender.id == old.id: - self.table.insertNode(old) + self.table.justSeenNode(old) df = old.ping(self.node.senderDict()) df.addCallbacks(_notStaleNodeHandler, _staleNodeHandler) @@ -155,19 +167,18 @@ class Khashmir(xmlrpc.XMLRPC): df = node.ping(self.node.senderDict()) ## these are the callbacks we use when we issue a PING def _pongHandler(sender, id=node.id, host=node.host, port=node.port, table=self.table): + sender = sender[0] if id != 20 * ' ' and id != sender['id'].data: # whoah, got response from different peer than we were expecting pass else: - #print "Got PONG from %s at %s:%s" % (`msg['id']`, t.target.host, t.target.port) sender['host'] = host sender['port'] = port n = Node().initWithDict(sender) table.insertNode(n) return - def _defaultPong(err): - # this should probably increment a failed message counter and dump the node if it gets over a threshold - return + def _defaultPong(err, node=node, table=self.table): + table.nodeFailed(node) df.addCallbacks(_pongHandler,_defaultPong) @@ -196,6 +207,19 @@ class Khashmir(xmlrpc.XMLRPC): self.findNode(id, callback) + def retrieveValues(self, key): + if self.kw.has_key(key): + c = self.kw.cursor() + tup = c.set(key) + l = [] + while(tup and tup[0] == key): + h1 = tup[1] + v = loads(self.store[h1])[1] + l.append(v) + tup = c.next() + return l + return [] + ##### ##### INCOMING MESSAGE HANDLERS @@ -207,7 +231,7 @@ class Khashmir(xmlrpc.XMLRPC): ip = self.crequest.getClientIP() sender['host'] = ip n = Node().initWithDict(sender) - self.insertNode(n) + self.insertNode(n, contacted=0) return self.node.senderDict() def xmlrpc_find_node(self, target, sender): @@ -216,7 +240,7 @@ class Khashmir(xmlrpc.XMLRPC): ip = self.crequest.getClientIP() sender['host'] = ip n = Node().initWithDict(sender) - self.insertNode(n) + self.insertNode(n, contacted=0) return nodes, self.node.senderDict() def xmlrpc_store_value(self, key, value, sender): @@ -237,7 +261,7 @@ class Khashmir(xmlrpc.XMLRPC): ip = self.crequest.getClientIP() sender['host'] = ip n = Node().initWithDict(sender) - self.insertNode(n) + self.insertNode(n, contacted=0) return self.node.senderDict() def xmlrpc_find_value(self, key, sender): @@ -245,17 +269,10 @@ class Khashmir(xmlrpc.XMLRPC): key = key.data sender['host'] = ip n = Node().initWithDict(sender) - self.insertNode(n) + self.insertNode(n, contacted=0) - if self.kw.has_key(key): - c = self.kw.cursor() - tup = c.set(key) - l = [] - while(tup): - h1 = tup[1] - v = loads(self.store[h1])[1] - l.append(v) - tup = c.next() + l = self.retrieveValues(key) + if len(l) > 0: l = map(lambda v: Binary(v), l) return {'values' : l}, self.node.senderDict() else: @@ -368,7 +385,7 @@ def test_find_value(l, quiet=0): try: if(len(values) == 0): if not self.found: - print "find FAILED" + print "find NOT FOUND" else: print "find FOUND" sys.stdout.flush() @@ -386,14 +403,18 @@ def test_find_value(l, quiet=0): d.valueForKey(key, cb(fc).callback) fc.wait() -def test_one(port): +def test_one(host, port): import thread - k = Khashmir('localhost', port) + k = Khashmir(host, port) thread.start_new_thread(k.app.run, ()) return k if __name__ == "__main__": - l = test_build_net() + import sys + n = 8 + if len(sys.argv) > 1: + n = int(sys.argv[1]) + l = test_build_net(peers=n) time.sleep(3) print "finding nodes..." for i in range(10):