r46786 - Merge hostname-endpoint-8014: HostnameEndpoint fix

Author: glyph
Date: Sun Feb 14 02:01:05 2016
New Revision: 46786

Added:
   trunk/twisted/topfiles/8014.bugfix
Modified:
   trunk/twisted/internet/endpoints.py
   trunk/twisted/internet/test/test_endpoints.py

Log:
Merge hostname-endpoint-8014: HostnameEndpoint fix

Author: glyph

Reviewer: herrwolfe

Fixes: #8014

twisted.internet.endpoints.HostnameEndpoint.connect no longer fails with an AlreadyCalledError when the Deferred it returns is cancelled after all outgoing connection attempts have been made but none have yet succeeded or failed.

Modified: trunk/twisted/internet/endpoints.py
==============================================================================
--- trunk/twisted/internet/endpoints.py	(original)
+++ trunk/twisted/internet/endpoints.py	Sun Feb 14 02:01:05 2016
@@ -635,25 +635,47 @@
 @implementer(interfaces.IStreamClientEndpoint)
 class HostnameEndpoint(object):
     """
-    A name-based endpoint that connects to the fastest amongst the
-    resolved host addresses.
+    A name-based endpoint that connects to the fastest amongst the resolved
+    host addresses.
 
     @ivar _getaddrinfo: A hook used for testing name resolution.
 
     @ivar _deferToThread: A hook used for testing deferToThread.
+
+    @cvar _DEFAULT_ATTEMPT_DELAY: The default time to use between attempts, in
+        seconds, when no C{attemptDelay} is given to
+        L{HostnameEndpoint.__init__}.
     """
     _getaddrinfo = staticmethod(socket.getaddrinfo)
     _deferToThread = staticmethod(threads.deferToThread)
+    _DEFAULT_ATTEMPT_DELAY = 0.3
 
-    def __init__(self, reactor, host, port, timeout=30, bindAddress=None):
+    def __init__(self, reactor, host, port, timeout=30, bindAddress=None,
+                 attemptDelay=None):
         """
+        Create a L{HostnameEndpoint}.
+
+        @param reactor: The reactor to use for connections and delayed calls.
+        @type reactor: provider of L{IReactorTCP} and L{IReactorTime}
+
         @param host: A hostname to connect to.
         @type host: L{bytes}
 
+        @param port: The port number to connect to.
+        @type port: L{int}
+
         @param timeout: For each individual connection attempt, the number of
             seconds to wait before assuming the connection has failed.
         @type timeout: L{int}
 
+        @param bindAddress: the local address of the network interface to make
+            the connections from.
+        @type bindAddress: L{bytes}
+
+        @param attemptDelay: The number of seconds to delay between connection
+            attempts.
+        @type attemptDelay: L{float}
+
         @see: L{twisted.internet.interfaces.IReactorTCP.connectTCP}
         """
         self._reactor = reactor
@@ -661,6 +683,9 @@
         self._port = port
         self._timeout = timeout
         self._bindAddress = bindAddress
+        if attemptDelay is None:
+            attemptDelay = self._DEFAULT_ATTEMPT_DELAY
+        self._attemptDelay = attemptDelay
 
 
     def connect(self, protocolFactory):
@@ -669,35 +694,14 @@
         connection which is established first.
         """
         wf = protocolFactory
-        pending = []
-
-        def _canceller(d):
-            """
-            The outgoing connection attempt was cancelled.  Fail that L{Deferred}
-            with an L{error.ConnectingCancelledError}.
-
-            @param d: The L{Deferred <defer.Deferred>} that was cancelled
-            @type d: L{Deferred <defer.Deferred>}
-
-            @return: C{None}
-            """
-            d.errback(error.ConnectingCancelledError(
-                HostnameAddress(self._host, self._port)))
-            for p in pending[:]:
-                p.cancel()
-
-        def errbackForGai(failure):
-            """
-            Errback for when L{_nameResolution} returns a Deferred that fires
-            with failure.
-            """
-            return defer.fail(error.DNSLookupError(
-                "Couldn't find the hostname '%s'" % (self._host,)))
-
-        def _endpoints(gaiResult):
+        d = self._nameResolution(self._host, self._port)
+        d.addErrback(lambda ignored: defer.fail(error.DNSLookupError(
+            "Couldn't find the hostname '%s'" % (self._host,))))
+        @d.addCallback
+        def gaiResultToEndpoints(gaiResult):
             """
             This method matches the host address family with an endpoint for
-            every address returned by GAI.
+            every address returned by C{getaddrinfo}.
 
             @param gaiResult: A list of 5-tuples as returned by GAI.
             @type gaiResult: list
@@ -709,72 +713,87 @@
                 elif family in [AF_INET]:
                     yield TCP4ClientEndpoint(self._reactor, sockaddr[0],
                             sockaddr[1], self._timeout, self._bindAddress)
-                        # Yields an endpoint for every address returned by GAI
+                    # Yields an endpoint for every address returned by GAI
+
+        def _canceller(d):
+            # This canceller must remain defined outside of
+            # `startConnectionAttempts`, because Deferred should not
+            # participate in cycles with their cancellers; that would create a
+            # potentially problematic circular reference and possibly
+            # gc.garbage.
+            d.errback(error.ConnectingCancelledError(
+                HostnameAddress(self._host, self._port)))
 
-        def attemptConnection(endpoints):
+        @d.addCallback
+        def startConnectionAttempts(endpoints):
             """
-            When L{endpoints} yields an endpoint, this method attempts to connect it.
+            Given a sequence of endpoints obtained via name resolution, start
+            connecting to a new one every C{self._attemptDelay} seconds until
+            one of the connections succeeds, all of them fail, or the attempt
+            is cancelled.
+
+            @param endpoints: an iterable of all the endpoints we might try to
+                connect to, as determined by name resolution.
+            @type endpoints: iterable of L{IStreamServerEndpoint}
+
+            @return: a Deferred that fires with the result of the
+                C{endpoint.connect} method that completes the fastest, or fails
+                with the first connection error it encountered if none of them
+                succeed.
+            @rtype: L{Deferred} failing with L{error.ConnectingCancelledError}
+                or firing with L{IProtocol}
             """
-            # The trial attempts for each endpoints, the recording of
-            # successful and failed attempts, and the algorithm to pick the
-            # winner endpoint goes here.
-            # Return a Deferred that fires with the endpoint that wins,
-            # or `failures` if none succeed.
-
-            endpointsListExhausted = []
-            successful = []
+            pending = []
             failures = []
             winner = defer.Deferred(canceller=_canceller)
 
-            def usedEndpointRemoval(connResult, connAttempt):
-                pending.remove(connAttempt)
-                return connResult
-
-            def afterConnectionAttempt(connResult):
-                if lc.running:
-                    lc.stop()
-
-                successful.append(True)
-                for p in pending[:]:
-                    p.cancel()
-                winner.callback(connResult)
-                return None
-
             def checkDone():
-                if endpointsListExhausted and not pending and not successful:
-                    winner.errback(failures.pop())
-
-            def connectFailed(reason):
-                failures.append(reason)
-                checkDone()
-                return None
+                if pending or checkDone.completed or checkDone.endpointsLeft:
+                    return
+                winner.errback(failures.pop())
+            checkDone.completed = False
+            checkDone.endpointsLeft = True
 
+            @LoopingCall
             def iterateEndpoint():
-                try:
-                    endpoint = next(endpoints)
-                except StopIteration:
+                endpoint = next(endpoints, None)
+                if endpoint is None:
                     # The list of endpoints ends.
-                    endpointsListExhausted.append(True)
-                    lc.stop()
+                    checkDone.endpointsLeft = False
+                    iterateEndpoint.stop()
+                    checkDone()
+                    return
+
+                eachAttempt = endpoint.connect(wf)
+                pending.append(eachAttempt)
+                @eachAttempt.addBoth
+                def noLongerPending(result):
+                    pending.remove(eachAttempt)
+                    return result
+                @eachAttempt.addCallback
+                def succeeded(result):
+                    if iterateEndpoint.running:
+                        iterateEndpoint.stop()
+                    winner.callback(result)
+                @eachAttempt.addErrback
+                def failed(reason):
+                    failures.append(reason)
                     checkDone()
-                else:
-                    dconn = endpoint.connect(wf)
-                    pending.append(dconn)
-                    dconn.addBoth(usedEndpointRemoval, dconn)
-                    dconn.addCallback(afterConnectionAttempt)
-                    dconn.addErrback(connectFailed)
-
-            lc = LoopingCall(iterateEndpoint)
-            lc.clock = self._reactor
-            lc.start(0.3)
+
+            iterateEndpoint.clock = self._reactor
+            iterateEndpoint.start(self._attemptDelay)
+
+            @winner.addBoth
+            def cancelRemainingPending(result):
+                checkDone.completed = True
+                for remaining in pending[:]:
+                    remaining.cancel()
+                return result
             return winner
 
-        d = self._nameResolution(self._host, self._port)
-        d.addErrback(errbackForGai)
-        d.addCallback(_endpoints)
-        d.addCallback(attemptConnection)
         return d
 
+
     def _nameResolution(self, host, port):
         """
         Resolve the hostname string into a tuple containig the host

Modified: trunk/twisted/internet/test/test_endpoints.py
==============================================================================
--- trunk/twisted/internet/test/test_endpoints.py	(original)
+++ trunk/twisted/internet/test/test_endpoints.py	Sun Feb 14 02:01:05 2016
@@ -1660,10 +1660,10 @@
         self.assertEqual(ep._getaddrinfo, socket.getaddrinfo)
 
 
-    def test_endpointConnectingCancelled(self):
+    def test_endpointConnectingCancelled(self, advance=None):
         """
         Calling L{Deferred.cancel} on the L{Deferred} returned from
-        L{IStreamClientEndpoint.connect} is errbacked with an expected
+        L{IStreamClientEndpoint.connect} will cause it to be errbacked with a
         L{ConnectingCancelledError} exception.
         """
         mreactor = MemoryReactor()
@@ -1675,6 +1675,8 @@
             mreactor, clientFactory)
 
         d = ep.connect(clientFactory)
+        if advance is not None:
+            mreactor.advance(advance)
         d.cancel()
         # When canceled, the connector will immediately notify its factory that
         # the connection attempt has failed due to a UserError.
@@ -1689,6 +1691,18 @@
         self.assertTrue(mreactor.tcpClients[0][2]._connector.stoppedConnecting)
 
 
+    def test_endpointConnectingCancelledAfterAllAttemptsStarted(self):
+        """
+        Calling L{Deferred.cancel} on the L{Deferred} returned from
+        L{IStreamClientEndpoint.connect} after enough time has passed that all
+        connection attempts have been initiated will cause it to be errbacked
+        with a L{ConnectingCancelledError} exception.
+        """
+        oneBetween = endpoints.HostnameEndpoint._DEFAULT_ATTEMPT_DELAY
+        advance = oneBetween + (oneBetween / 2.0)
+        self.test_endpointConnectingCancelled(advance=advance)
+
+
     def test_endpointConnectFailure(self):
         """
         If L{HostnameEndpoint.connect} is invoked and there is no server
@@ -1706,7 +1720,7 @@
             mreactor, clientFactory)
 
         d = ep.connect(clientFactory)
-        mreactor.advance(0.3)
+        mreactor.advance(endpoints.HostnameEndpoint._DEFAULT_ATTEMPT_DELAY)
         self.assertEqual(self.failureResultOf(d).value, expectedError)