From 84baa5e0656a66ea15fa9b48c421ce850639dbe9 Mon Sep 17 00:00:00 2001 From: Job Evers-Meltzer Date: Wed, 14 Dec 2016 13:57:19 -0600 Subject: [PATCH] dht: refactor _msgTimeout --- lbrynet/dht/protocol.py | 59 +++++++++++++++++++++++------------------ 1 file changed, 33 insertions(+), 26 deletions(-) diff --git a/lbrynet/dht/protocol.py b/lbrynet/dht/protocol.py index f99711f36..3530f78f6 100644 --- a/lbrynet/dht/protocol.py +++ b/lbrynet/dht/protocol.py @@ -282,34 +282,41 @@ class KademliaProtocol(protocol.DatagramProtocol): def _msgTimeout(self, messageID): """ Called when an RPC request message times out """ # Find the message that timed out - if self._sentMessages.has_key(messageID): - remoteContactID, df = self._sentMessages[messageID][0:2] - if self._partialMessages.has_key(messageID): - # We are still receiving this message - # See if any progress has been made; if not, kill the message - if self._partialMessagesProgress.has_key(messageID): - same_length = ( - len(self._partialMessagesProgress[messageID]) == - len(self._partialMessages[messageID])) - if same_length: - # No progress has been made - del self._partialMessagesProgress[messageID] - del self._partialMessages[messageID] - df.errback(failure.Failure(TimeoutError(remoteContactID))) - return - # Reset the RPC timeout timer - timeoutCall = reactor.callLater( - constants.rpcTimeout, self._msgTimeout, messageID) #IGNORE:E1101 - self._sentMessages[messageID] = (remoteContactID, df, timeoutCall) - return - del self._sentMessages[messageID] - # The message's destination node is now considered to be dead; - # raise an (asynchronous) TimeoutError exception and update the host node - self._node.removeContact(remoteContactID) - df.errback(failure.Failure(TimeoutError(remoteContactID))) - else: + if not self._sentMessages.has_key(messageID): # This should never be reached log.error("deferred timed out, but is not present in sent messages list!") + return + remoteContactID, df = self._sentMessages[messageID][0:2] + if self._partialMessages.has_key(messageID): + # We are still receiving this message + self._msgTimeoutInProgress(messageID, remoteContactID, df) + return + del self._sentMessages[messageID] + # The message's destination node is now considered to be dead; + # raise an (asynchronous) TimeoutError exception and update the host node + self._node.removeContact(remoteContactID) + df.errback(failure.Failure(TimeoutError(remoteContactID))) + + def _msgTimeoutInProgress(self, messageID, remoteContactID, df): + # See if any progress has been made; if not, kill the message + if self._hasProgressBeenMade(messageID): + # Reset the RPC timeout timer + timeoutCall = reactor.callLater(constants.rpcTimeout, self._msgTimeout, messageID) + self._sentMessages[messageID] = (remoteContactID, df, timeoutCall) + else: + # No progress has been made + del self._partialMessagesProgress[messageID] + del self._partialMessages[messageID] + df.errback(failure.Failure(TimeoutError(remoteContactID))) + + def _hasProgressBeenMade(self, messageID): + return ( + self._partialMessagesProgress.has_key(messageID) and + ( + len(self._partialMessagesProgress[messageID]) != + len(self._partialMessages[messageID]) + ) + ) def stopProtocol(self): """ Called when the transport is disconnected.