Refactor timeout handling

Author: cpacia

dht/tests/test_protocol.py

@@ -10,7 +10,7 @@
 import nacl.hash
 from txrudp import connection, rudp, packet, constants
 from twisted.trial import unittest
-from twisted.internet import task, reactor, address, udp, defer
+from twisted.internet import task, address, udp, defer
 from dht.protocol import KademliaProtocol
 from dht.utils import digest
 from dht.storage import ForgetfulStorage
@@ -451,13 +451,11 @@ def handle_response(resp):
             self.assertTrue(resp[0])
             self.assertEqual(resp[1][0], "test")
             self.assertTrue(message_id not in self.protocol._outstanding)
-            self.assertFalse(timeout.active())
 
         message_id = digest("msgid")
         n = Node(digest("S"), self.addr1[0], self.addr1[1])
         d = defer.Deferred()
-        timeout = reactor.callLater(5, self.protocol._timeout, message_id)
-        self.protocol._outstanding[message_id] = (d, timeout)
+        self.protocol._outstanding[message_id] = (d, self.addr1)
         self.protocol._acceptResponse(message_id, ["test"], n)
 
         return d.addCallback(handle_response)
@@ -466,22 +464,17 @@ def test_unknownRPC(self):
         self.assertFalse(self.handler.receive_message(str(random.getrandbits(1400))))
 
     def test_timeout(self):
-        self._connecting_to_connected()
-        self.wire_protocol[self.addr1] = self.con
 
-        def test_remove_outstanding():
-            self.assertTrue(len(self.protocol._outstanding) == 0)
-
-        def test_deffered(d):
-            self.assertFalse(d[0])
-            test_remove_outstanding()
+        def handle_response(resp, n):
+            self.assertFalse(resp[0])
+            self.assertIsNone(resp[1])
+            self.assertTrue(self.protocol.router.isNewNode(n))
 
         n = Node(digest("S"), self.addr1[0], self.addr1[1])
-        d = self.protocol.callPing(n)
-        self.clock.advance(6)
-        connection.REACTOR.runUntilCurrent()
-        self.clock.advance(6)
-        return d.addCallback(test_deffered)
+        d = defer.Deferred().addCallback(handle_response, n)
+        self.protocol._outstanding["msgID"] = [d, self.addr1]
+        self.protocol.router.addContact(n)
+        self.protocol.timeout(self.addr1, n)
 
     def test_transferKeyValues(self):
         self._connecting_to_connected()

rpcudp.py

@@ -11,13 +11,12 @@
 from binascii import hexlify
 from hashlib import sha1
 from base64 import b64encode
-from twisted.internet import reactor
 from twisted.internet import defer
 from log import Logger
 from protos.message import Message, Command
 from dht import node
 from constants import PROTOCOL_VERSION
-from protos.message import NOT_FOUND, GET_IMAGE, GET_CONTRACT
+from protos.message import NOT_FOUND
 
 
 class RPCProtocol:
@@ -103,8 +102,7 @@ def _acceptResponse(self, msgID, data, sender):
             self.log.debug("received response for message id %s from %s" % msgargs)
         else:
             self.log.warning("received 404 error response from %s" % sender)
-        d, timeout = self._outstanding[msgID]
-        timeout.cancel()
+        d = self._outstanding[msgID][0]
         d.callback((True, data))
         del self._outstanding[msgID]
 
@@ -137,31 +135,17 @@ def _sendResponse(self, response, funcname, msgID, sender, connection):
         data = m.SerializeToString()
         connection.send_message(data)
 
-    def _timeout(self, msgID, address):
+    def timeout(self, address, node_to_remove):
         """
-        If a message times out we are first going to try hole punching because
-        the node may be behind a restricted NAT. If it is successful, the original
-        should get through. This timeout will only fire if the hole punching
-        fails.
+        This timeout is called by the txrudp connection handler. We will run through the
+        outstanding messages and callback false on any waiting on this IP address.
         """
-        # pylint: disable=pointless-string-statement
-        """
-        Hole punching disabled for now
-
-        seed = SEED_NODE_TESTNET if self.multiplexer.testnet else SEED_NODE
-        if not hp and self.multiplexer.ip_address[0] != seed[0]:
-            args = (address[0], address[1], b64encode(msgID))
-            self.log.debug("did not receive reply from %s:%s for msgID %s, trying hole punching..." % args)
-            self.hole_punch(seed, address[0], address[1], "True")
-            timeout = reactor.callLater(self._waitTimeout, self._timeout, msgID, address, True)
-            self._outstanding[msgID][1] = timeout
-        else:
-        """
-        args = (b64encode(msgID), self._waitTimeout)
-        self.log.warning("did not receive reply for msg id %s within %i seconds" % args)
-        self._outstanding[msgID][0].callback((False, None))
-        del self._outstanding[msgID]
-        self.multiplexer[address].shutdown()
+        if node_to_remove is not None:
+            self.router.removeContact(node_to_remove)
+        for msgID, val in self._outstanding.items():
+            if address == val[1]:
+                val[0].callback((False, None))
+                del self._outstanding[msgID]
 
     def rpc_hole_punch(self, sender, ip, port, relay="False"):
         """
@@ -175,12 +159,6 @@ def rpc_hole_punch(self, sender, ip, port, relay="False"):
             self.log.debug("punching through NAT for %s:%s" % (ip, port))
             self.multiplexer.send_datagram(" ", (ip, int(port)))
 
-    def _get_waitTimeout(self, command):
-        if command == GET_IMAGE or command == GET_CONTRACT:
-            return 100
-        else:
-            return self._waitTimeout
-
     def __getattr__(self, name):
         if name.startswith("_") or name.startswith("rpc_"):
             return object.__getattr__(self, name)
@@ -202,8 +180,7 @@ def func(address, *args):
             m.testnet = self.multiplexer.testnet
             data = m.SerializeToString()
             d = defer.Deferred()
-            timeout = reactor.callLater(self._get_waitTimeout(m.command), self._timeout, msgID, address)
-            self._outstanding[msgID] = [d, timeout]
+            self._outstanding[msgID] = [d, address]
             self.multiplexer.send_message(data, address)
             self.log.debug("calling remote function %s on %s (msgid %s)" % (name, address, b64encode(msgID)))
             return d

wireprotocol.py

@@ -6,7 +6,7 @@
 from twisted.internet.task import LoopingCall
 from twisted.internet import task, reactor
 from interfaces import MessageProcessor
-from protos.message import Message, FIND_VALUE
+from protos.message import Message
 from log import Logger
 from dht.node import Node
 from protos.message import PING, NOT_FOUND
@@ -74,15 +74,15 @@ def receive_message(self, datagram):
                 return False
 
         def handle_shutdown(self):
-            if self.connection is not None:
-                for processor in self.processors:
-                    if FIND_VALUE in processor and self.node is not None:
-                        processor.router.removeContact(self.node)
-                reactor.callLater(90, self.connection.unregister)
-                if self.addr:
-                    self.log.info("connection with %s terminated" % self.addr)
+            for processor in self.processors:
+                processor.timeout((self.connection.dest_addr[0], self.connection.dest_addr[1]), self.node)
+            reactor.callLater(90, self.connection.unregister)
+            if self.addr:
+                self.log.info("connection with %s terminated" % self.addr)
+            try:
                 self.keep_alive_loop.stop()
-                self.connection = None
+            except Exception:
+                pass
 
         def keep_alive(self):
             for processor in self.processors: