X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=apt_dht_Khashmir%2FDHT.py;h=36d7231fe75e48024d5fa3208054f861f97cd79b;hb=08e747941eaa12ddf466a49e4dadf292758d6cd2;hp=1c4416b5bc86c785d0953866b28e5f552683985b;hpb=faf2a6ecf4e024cee49c4495ad916b3229ad4b1a;p=quix0rs-apt-p2p.git diff --git a/apt_dht_Khashmir/DHT.py b/apt_dht_Khashmir/DHT.py index 1c4416b..36d7231 100644 --- a/apt_dht_Khashmir/DHT.py +++ b/apt_dht_Khashmir/DHT.py @@ -1,7 +1,8 @@ -import os +import os, sha, random -from twisted.internet import defer +from twisted.internet import defer, reactor +from twisted.internet.abstract import isIPAddress from twisted.trial import unittest from zope.interface import implements @@ -22,13 +23,16 @@ class DHT: self.bootstrap_node = False self.joining = None self.joined = False + self.storing = {} + self.retrieving = {} + self.retrieved = {} def loadConfig(self, config, section): """See L{apt_dht.interfaces.IDHT}.""" self.config_parser = config self.section = section - self.config = [] - self.cache_dir = self.config_parser.get('DEFAULT', 'cache_dir') + self.config = {} + self.cache_dir = self.config_parser.get(section, 'cache_dir') self.bootstrap = self.config_parser.getstringlist(section, 'BOOTSTRAP') self.bootstrap_node = self.config_parser.getboolean(section, 'BOOTSTRAP_NODE') for k in self.config_parser.options(section): @@ -40,13 +44,13 @@ class DHT: self.config[k] = self.config_parser.gettime(section, k) else: self.config[k] = self.config_parser.get(section, k) - if 'PORT' not in self.config: - self.config['PORT'] = self.config_parser.getint('DEFAULT', 'PORT') def join(self): """See L{apt_dht.interfaces.IDHT}.""" if self.config is None: raise DHTError, "configuration not loaded" + if self.joining: + raise DHTError, "a join is already in progress" self.khashmir = Khashmir(self.config, self.cache_dir) @@ -54,9 +58,16 @@ class DHT: for node in self.bootstrap: host, port = node.rsplit(':', 1) port = int(port) - self.khashmir.addContact(host, port, self._join_single) + if isIPAddress(host): + self._join_gotIP(host, port) + else: + reactor.resolve(host).addCallback(self._join_gotIP, port) return self.joining + + def _join_gotIP(self, ip, port): + """Called after an IP address has been found for a single bootstrap node.""" + self.khashmir.addContact(ip, port, self._join_single) def _join_single(self): """Called when a single bootstrap node has been added.""" @@ -67,19 +78,35 @@ class DHT: if not self.joined: self.joined = True if len(result) > 0 or self.bootstrap_node: - self.joining.callback(result) + df = self.joining + self.joining = None + df.callback(result) else: - self.joining.errback(DHTError('could not find any nodes to bootstrap to')) + df = self.joining + self.joining = None + df.errback(DHTError('could not find any nodes to bootstrap to')) def leave(self): """See L{apt_dht.interfaces.IDHT}.""" if self.config is None: raise DHTError, "configuration not loaded" - if self.joined: + if self.joined or self.joining: + if self.joining: + self.joining.errback(DHTError('still joining when leave was called')) + self.joining = None self.joined = False self.khashmir.shutdown() + def normalizeKey(self, key): + """Normalize a key's length suitable for insertion in the DHT.""" + key_bytes = (self.config['HASH_LENGTH'] - 1) // 8 + 1 + if len(key) < key_bytes: + key = key + '\000'*(key_bytes - len(key)) + elif len(key) > key_bytes: + key = key[:key_bytes] + return key + def getValue(self, key): """See L{apt_dht.interfaces.IDHT}.""" if self.config is None: @@ -88,9 +115,25 @@ class DHT: raise DHTError, "have not joined a network yet" d = defer.Deferred() - self.khashmir.valueForKey(key, d.callback) + key = self.normalizeKey(key) + if key not in self.retrieving: + self.khashmir.valueForKey(key, self._getValue) + self.retrieving.setdefault(key, []).append(d) return d + def _getValue(self, key, result): + if result: + self.retrieved.setdefault(key, []).extend(result) + else: + final_result = [] + if key in self.retrieved: + final_result = self.retrieved[key] + del self.retrieved[key] + for i in range(len(self.retrieving[key])): + d = self.retrieving[key].pop(0) + d.callback(final_result) + del self.retrieving[key] + def storeValue(self, key, value): """See L{apt_dht.interfaces.IDHT}.""" if self.config is None: @@ -98,7 +141,24 @@ class DHT: if not self.joined: raise DHTError, "have not joined a network yet" - self.khashmir.storeValueForKey(key, value) + key = self.normalizeKey(key) + if key in self.storing and value in self.storing[key]: + raise DHTError, "already storing that key with the same value" + + d = defer.Deferred() + self.khashmir.storeValueForKey(key, value, self._storeValue) + self.storing.setdefault(key, {})[value] = d + return d + + def _storeValue(self, key, value, result): + if key in self.storing and value in self.storing[key]: + if len(result) > 0: + self.storing[key][value].callback(result) + else: + self.storing[key][value].errback(DHTError('could not store value %s in key %s' % (value, key))) + del self.storing[key][value] + if len(self.storing[key].keys()) == 0: + del self.storing[key] class TestSimpleDHT(unittest.TestCase): """Unit tests for the DHT.""" @@ -124,6 +184,13 @@ class TestSimpleDHT(unittest.TestCase): self.b.bootstrap = ["127.0.0.1:4044"] self.b.cache_dir = '/tmp' + def test_normalizeKey(self): + self.failUnless(self.a.normalizeKey('12345678901234567890') == '12345678901234567890') + self.failUnless(self.a.normalizeKey('12345678901234567') == '12345678901234567\000\000\000') + self.failUnless(self.a.normalizeKey('1234567890123456789012345') == '12345678901234567890') + self.failUnless(self.a.normalizeKey('1234567890123456789') == '1234567890123456789\000') + self.failUnless(self.a.normalizeKey('123456789012345678901') == '12345678901234567890') + def test_bootstrap_join(self): d = self.a.join() return d @@ -138,22 +205,64 @@ class TestSimpleDHT(unittest.TestCase): d.addCallback(self.node_join) d.addCallback(self.lastDefer.callback) return self.lastDefer + + def value_stored(self, result, value): + self.stored -= 1 + if self.stored == 0: + self.get_values() + def store_values(self, result): + self.stored = 3 + d = self.a.storeValue(sha.new('4045').digest(), str(4045*3)) + d.addCallback(self.value_stored, 4045) + d = self.a.storeValue(sha.new('4044').digest(), str(4044*2)) + d.addCallback(self.value_stored, 4044) + d = self.b.storeValue(sha.new('4045').digest(), str(4045*2)) + d.addCallback(self.value_stored, 4045) + + def check_values(self, result, values): + self.checked -= 1 + self.failUnless(len(result) == len(values)) + for v in result: + self.failUnless(v in values) + if self.checked == 0: + self.lastDefer.callback(1) + + def get_values(self): + self.checked = 4 + d = self.a.getValue(sha.new('4044').digest()) + d.addCallback(self.check_values, [str(4044*2)]) + d = self.b.getValue(sha.new('4044').digest()) + d.addCallback(self.check_values, [str(4044*2)]) + d = self.a.getValue(sha.new('4045').digest()) + d.addCallback(self.check_values, [str(4045*2), str(4045*3)]) + d = self.b.getValue(sha.new('4045').digest()) + d.addCallback(self.check_values, [str(4045*2), str(4045*3)]) + + def test_store(self): + from twisted.internet.base import DelayedCall + DelayedCall.debug = True + self.lastDefer = defer.Deferred() + d = self.a.join() + d.addCallback(self.node_join) + d.addCallback(self.store_values) + return self.lastDefer + def tearDown(self): self.a.leave() try: - os.unlink(self.a.khashmir.db) + os.unlink(self.a.khashmir.store.db) except: pass self.b.leave() try: - os.unlink(self.b.khashmir.db) + os.unlink(self.b.khashmir.store.db) except: pass class TestMultiDHT(unittest.TestCase): - timeout = 10 + timeout = 60 num = 20 DHT_DEFAULTS = {'PORT': 9977, 'K': 8, 'HASH_LENGTH': 160, 'CHECKPOINT_INTERVAL': 900, 'CONCURRENT_REQS': 4, @@ -164,7 +273,7 @@ class TestMultiDHT(unittest.TestCase): def setUp(self): self.l = [] - self.startport = 4088 + self.startport = 4081 for i in range(self.num): self.l.append(DHT()) self.l[i].config = self.DHT_DEFAULTS.copy() @@ -181,15 +290,58 @@ class TestMultiDHT(unittest.TestCase): d.addCallback(self.lastDefer.callback) def test_join(self): + self.timeout = 2 self.lastDefer = defer.Deferred() d = self.l[0].join() d.addCallback(self.node_join, 1) return self.lastDefer + def store_values(self, result, i = 0, j = 0): + if j > i: + j -= i+1 + i += 1 + if i == len(self.l): + self.get_values() + else: + d = self.l[j].storeValue(sha.new(str(self.startport+i)).digest(), str((self.startport+i)*(j+1))) + d.addCallback(self.store_values, i, j+1) + + def get_values(self, result = None, check = None, i = 0, j = 0): + if result is not None: + self.failUnless(len(result) == len(check)) + for v in result: + self.failUnless(v in check) + if j >= len(self.l): + j -= len(self.l) + i += 1 + if i == len(self.l): + self.lastDefer.callback(1) + else: + d = self.l[i].getValue(sha.new(str(self.startport+j)).digest()) + check = [] + for k in range(self.startport+j, (self.startport+j)*(j+1)+1, self.startport+j): + check.append(str(k)) + d.addCallback(self.get_values, check, i, j + random.randrange(1, min(len(self.l), 10))) + + def store_join(self, result, next_node): + d = self.l[next_node].join() + if next_node + 1 < len(self.l): + d.addCallback(self.store_join, next_node + 1) + else: + d.addCallback(self.store_values) + + def test_store(self): + from twisted.internet.base import DelayedCall + DelayedCall.debug = True + self.lastDefer = defer.Deferred() + d = self.l[0].join() + d.addCallback(self.store_join, 1) + return self.lastDefer + def tearDown(self): for i in self.l: try: i.leave() - os.unlink(i.khashmir.db) + os.unlink(i.khashmir.store.db) except: pass