X-Git-Url: https://git.mxchange.org/?p=quix0rs-apt-p2p.git;a=blobdiff_plain;f=apt_p2p_Khashmir%2Fkrpc.py;h=b151a5527f15b6f210f14e83b565a790cfe750b7;hp=92d03a727a2211b586efc9e49009ef657806aefb;hb=50d3a09814b6539279022cdf4609645ecea3f99b;hpb=136c82d6c138402e93bdefb499aed967ba059385 diff --git a/apt_p2p_Khashmir/krpc.py b/apt_p2p_Khashmir/krpc.py index 92d03a7..b151a55 100644 --- a/apt_p2p_Khashmir/krpc.py +++ b/apt_p2p_Khashmir/krpc.py @@ -1,11 +1,6 @@ -## Copyright 2002-2003 Andrew Loewenstern, All Rights Reserved -# see LICENSE.txt for license information """The KRPC communication protocol implementation. -@var KRPC_INITIAL_DELAY: the number of seconds after which to try resending - the request, the resends will wait twice as long each time -@var KRPC_TIMEOUT: the number of seconds after which requests timeout @var UDP_PACKET_LIMIT: the maximum number of bytes that can be sent in a UDP packet without fragmentation @@ -50,8 +45,6 @@ from twisted.trial import unittest from khash import newID -KRPC_INITIAL_DELAY = 2 -KRPC_TIMEOUT = 14 UDP_PACKET_LIMIT = 1472 # Remote node errors @@ -185,7 +178,7 @@ class hostbroker(protocol.DatagramProtocol): # Create a new protocol object if necessary if not self.connections.has_key(addr): - conn = self.protocol(addr, self.server, self.stats, self.transport, self.config['SPEW']) + conn = self.protocol(addr, self.server, self.stats, self.transport, self.config) self.connections[addr] = conn else: conn = self.connections[addr] @@ -213,15 +206,19 @@ class KrpcRequest(defer.Deferred): @ivar method: the name of the method to call on the remote node @type data: C{string} @ivar data: the message to send to the remote node + @type config: C{dictionary} + @ivar config: the configuration parameters for the DHT @type delay: C{int} @ivar delay: the last timeout delay sent @type start: C{datetime} @ivar start: the time to request was started at - @type later: L{twisted.internet.interfaces.IDelayedCall} - @ivar later: the pending call to timeout the last sent request + @type laterNextTimeout: L{twisted.internet.interfaces.IDelayedCall} + @ivar laterNextTimeout: the pending call to timeout the last sent request + @type laterFinalTimeout: L{twisted.internet.interfaces.IDelayedCall} + @ivar laterFinalTimeout: the pending call to timeout the entire request """ - def __init__(self, protocol, newTID, method, data): + def __init__(self, protocol, newTID, method, data, config): """Initialize the request, and send it out. @type protocol: L{KRPC} @@ -231,37 +228,49 @@ class KrpcRequest(defer.Deferred): @param method: the name of the method to call on the remote node @type data: C{string} @param data: the message to send to the remote node + @type config: C{dictionary} + @param config: the configuration parameters for the DHT """ defer.Deferred.__init__(self) self.protocol = protocol self.tid = newTID self.method = method self.data = data - self.delay = KRPC_INITIAL_DELAY + self.config = config + self.delay = self.config.get('KRPC_INITIAL_DELAY', 2) self.start = datetime.now() - self.later = None - self.send() + self.laterNextTimeout = None + self.laterFinalTimeout = reactor.callLater(self.config.get('KRPC_TIMEOUT', 9), self.finalTimeout) + reactor.callLater(0, self.send) def send(self): """Send the request to the remote node.""" - assert not self.later, 'There is already a pending request' - self.later = reactor.callLater(self.delay, self.timeOut) - self.protocol.sendData(self.method, self.data) + assert not self.laterNextTimeout, 'There is already a pending request' + self.laterNextTimeout = reactor.callLater(self.delay, self.nextTimeout) + try: + self.protocol.sendData(self.method, self.data) + except: + log.err() - def timeOut(self): + def nextTimeout(self): """Check for a unrecoverable timeout, otherwise resend.""" - self.later = None - delay = datetime.now() - self.start - if delay > timedelta(seconds = KRPC_TIMEOUT): - log.msg('%r timed out after %0.2f sec' % - (self.tid, delay.seconds + delay.microseconds/1000000.0)) - self.protocol.timeOut(self.tid, self.method) + self.laterNextTimeout = None + if datetime.now() - self.start > timedelta(seconds = self.config.get('KRPC_TIMEOUT', 9)): + self.finalTimeout() elif self.protocol.stopped: log.msg('Timeout but can not resend %r, protocol has been stopped' % self.tid) else: self.delay *= 2 log.msg('Trying to resend %r now with delay %d sec' % (self.tid, self.delay)) - self.send() + reactor.callLater(0, self.send) + + def finalTimeout(self): + """Timeout the request after an unrecoverable timeout.""" + self.dropTimeOut() + delay = datetime.now() - self.start + log.msg('%r timed out after %0.2f sec' % + (self.tid, delay.seconds + delay.microseconds/1000000.0)) + self.protocol.timeOut(self.tid, self.method) def callback(self, resp): self.dropTimeOut() @@ -273,9 +282,12 @@ class KrpcRequest(defer.Deferred): def dropTimeOut(self): """Cancel the timeout call when a response is received.""" - if self.later and self.later.active(): - self.later.cancel() - self.later = None + if self.laterFinalTimeout and self.laterFinalTimeout.active(): + self.laterFinalTimeout.cancel() + self.laterFinalTimeout = None + if self.laterNextTimeout and self.laterNextTimeout.active(): + self.laterNextTimeout.cancel() + self.laterNextTimeout = None class KRPC: """The KRPC protocol implementation. @@ -287,8 +299,8 @@ class KRPC: @ivar stats: the statistics logger to save transport info @type addr: (C{string}, C{int}) @ivar addr: the IP address and port of the source node - @type noisy: C{boolean} - @ivar noisy: whether to log additional details of the protocol + @type config: C{dictionary} + @ivar config: the configuration parameters for the DHT @type tids: C{dictionary} @ivar tids: the transaction IDs outstanding for requests, keys are the transaction ID of the request, values are the deferreds to call with @@ -297,7 +309,7 @@ class KRPC: @ivar stopped: whether the protocol has been stopped """ - def __init__(self, addr, server, stats, transport, spew = False): + def __init__(self, addr, server, stats, transport, config = {}): """Initialize the protocol. @type addr: (C{string}, C{int}) @@ -307,15 +319,15 @@ class KRPC: @type stats: L{stats.StatsLogger} @param stats: the statistics logger to save transport info @param transport: the transport to use for the protocol - @type spew: C{boolean} - @param spew: whether to log additional details of the protocol - (optional, defaults to False) + @type config: C{dictionary} + @param config: the configuration parameters for the DHT + (optional, defaults to using defaults) """ self.transport = transport self.factory = server self.stats = stats self.addr = addr - self.noisy = spew + self.config = config self.tids = {} self.stopped = False @@ -329,14 +341,14 @@ class KRPC: """ self.stats.receivedBytes(len(data)) if self.stopped: - if self.noisy: + if self.config.get('SPEW', False): log.msg("stopped, dropping message from %r: %s" % (addr, data)) # Bdecode the message try: msg = bdecode(data) except Exception, e: - if self.noisy: + if self.config.get('SPEW', False): log.msg("krpc bdecode error: ") log.err(e) return @@ -349,7 +361,7 @@ class KRPC: log.err(e) return - if self.noisy: + if self.config.get('SPEW', False): log.msg("%d received from %r: %s" % (self.factory.port, addr, msg)) # Process it based on its type @@ -365,7 +377,8 @@ class KRPC: ret = f(*(), **msg[ARG]) except KrpcError, e: log.msg('Got a Krpc error while running: krpc_%s' % msg[REQ]) - log.err(e) + if e[0] != KRPC_ERROR_INVALID_TOKEN: + log.err(e) self.stats.errorAction(msg[REQ]) olen = self._sendResponse(msg[REQ], addr, msg[TID], ERR, [e[0], e[1]]) @@ -389,9 +402,8 @@ class KRPC: self.stats.receivedAction('unknown') olen = self._sendResponse(msg[REQ], addr, msg[TID], ERR, [KRPC_ERROR_METHOD_UNKNOWN, "unknown method "+str(msg[REQ])]) - if self.noisy: - log.msg("%s >>> %s - %s %s %s" % (addr, self.factory.node.port, - ilen, msg[REQ], olen)) + + log.msg("%s >>> %s %s %s" % (addr, ilen, msg[REQ], olen)) elif msg[TYP] == RSP: # Responses get processed by their TID's deferred if self.tids.has_key(msg[TID]): @@ -402,7 +414,7 @@ class KRPC: req.callback(msg[RSP]) else: # no tid, this transaction was finished already... - if self.noisy: + if self.config.get('SPEW', False): log.msg('received response from %r for completed request: %r' % (msg[RSP]['id'], msg[TID])) elif msg[TYP] == ERR: @@ -418,7 +430,7 @@ class KRPC: (msg[ERR], msg[RSP]['id'], msg[TID])) else: # Received an unknown message type - if self.noisy: + if self.config.get('SPEW', False): log.msg("unknown message type: %r" % msg) if msg[TID] in self.tids: req = self.tids[msg[TID]] @@ -444,7 +456,7 @@ class KRPC: # Create the response message msg = {TID : tid, TYP : msgType, msgType : response} - if self.noisy: + if self.config.get('SPEW', False): log.msg("%d responding to %r: %s" % (self.factory.port, addr, msg)) out = bencode(msg) @@ -513,17 +525,18 @@ class KRPC: # Create the request message newTID = newID() msg = {TID : newTID, TYP : REQ, REQ : method, ARG : args} - if self.noisy: + if self.config.get('SPEW', False): log.msg("%d sending to %r: %s" % (self.factory.port, self.addr, msg)) data = bencode(msg) # Create the request object and save it with the TID - req = KrpcRequest(self, newTID, method, data) + req = KrpcRequest(self, newTID, method, data, self.config) self.tids[newTID] = req # Save the conclusion of the action req.addCallbacks(self.stats.responseAction, self.stats.failedAction, - callbackArgs = (method, ), errbackArgs = (method, )) + callbackArgs = (method, datetime.now()), + errbackArgs = (method, datetime.now())) return req @@ -535,11 +548,10 @@ class KRPC: @type data: C{string} @param data: the message to send to the remote node """ + self.transport.write(data, self.addr) self.stats.sentAction(method) self.stats.sentBytes(len(data)) - self.transport.write(data, self.addr) - def timeOut(self, badTID, method): """Call the deferred's errback if a timeout occurs. @@ -582,7 +594,8 @@ class Receiver(protocol.Factory): def make(port): from stats import StatsLogger af = Receiver() - a = hostbroker(af, StatsLogger(None, None, {}), {'SPEW': False}) + a = hostbroker(af, StatsLogger(None, None), + {'KRPC_TIMEOUT': 9, 'KRPC_INITIAL_DELAY': 2, 'SPEW': False}) a.protocol = KRPC p = reactor.listenUDP(port, a) return af, a, p