Merge lp:~free.ekanayaka/landscape-client/method-call-defer into lp:~landscape/landscape-client/amp-trunk

[Sidnei da Silva]

[Free Ekanayaka (free.ekanayaka)]

Please note that this mechanism is mainly meant for unit-testing. In general our code will not really use this feature or rely on it, because we don't generally wait at all for remote calls that return deferreds. However when unit-testing it is often needed to wait for a remote call to complete, and this branch makes it easy.

[Duncan McGreggor (oubiwann)]

This is super-cool, Free -- so glad you did this :-)

[1] 1661 tests, all passing -- nice!

[2] This isn't part of this particular change, but there's a bare exception before the deferred check:

| try:
| result = getattr(self.factory.object, method)(*args, **kwargs)
| except Exception, error:
| raise MethodCallError("Remote exception %s" % str(error))

It'd be nice if expected exceptions were handled explicitly here.

[3] You'll want to get Thomas' feedback on this suggestion... but one thing you might want to consider is actually having receive_method_call always return a deferred (with return maybeDeferred(result)). That would make the method much simpler, but it would require changes elsewhere. Regardless, probably enough work to justify being done in a separate ticket/branch ;-)

+1 for merge!

review: Approve

[lp:~free.ekanayaka/landscape-client/method-call-defer]

216. By Free Ekanayaka on 2010-01-25

Use maybeDeferred to simplify receive_method_call logic (Duncan [3])

[Free Ekanayaka (free.ekanayaka)]

Thanks Duncan! It was awesome to get your Landscape "goodbye review" :)

[1]

Yay, thanks buildbot.

[2]

The thing is that this is a very general method call, so in principle any exception could be raised. However whatever gets raised is wrapped around a MethodCallError, so the remote caller can possibly cope with it.

[3]

Great idea! That indeed simplified the logic quite a bit. Thanks.

[Kapil Thangavelu (hazmat)]

Nice implementation, thorough test coverage. Looks good to me. +1

review: Approve

[Free Ekanayaka (free.ekanayaka)]

Thanks Kapil, that's merged in r185 of amp-trunk.

=== modified file 'landscape/lib/amp.py'
--- landscape/lib/amp.py	2010-01-20 08:33:12 +0000
+++ landscape/lib/amp.py	2010-01-25 08:30:27 +0000
@@ -1,8 +1,10 @@
 """Expose the methods of a remote object over AMP."""
 
-from twisted.internet.defer import Deferred
+from uuid import uuid4
+from twisted.internet.defer import Deferred, maybeDeferred
 from twisted.internet.protocol import ServerFactory, ClientFactory
 from twisted.protocols.amp import Argument, String, Command, AMP
+from twisted.python.failure import Failure
 
 from landscape.lib.bpickle import loads, dumps, dumps_table
 
@@ -35,11 +37,21 @@
                  ("args", MethodCallArgument()),
                  ("kwargs", MethodCallArgument())]
 
-    response = [("result", MethodCallArgument())]
+    response = [("result", MethodCallArgument()),
+                ("deferred", String(optional=True))]
 
     errors = {MethodCallError: "METHOD_CALL_ERROR"}
 
 
+class DeferredResponse(Command):
+    """Fire a L{Deferred} associated with an outstanding method call result."""
+
+    arguments = [("uuid", String()),
+                 ("result", MethodCallArgument(optional=True)),
+                 ("failure", String(optional=True))]
+    requiresAnswer = False
+
+
 class MethodCallServerProtocol(AMP):
     """Expose methods of a local object over AMP.
 
@@ -68,17 +80,109 @@
         if not method in self.methods:
             raise MethodCallError("Forbidden method '%s'" % method)
 
-        try:
-            result = getattr(self.factory.object, method)(*args, **kwargs)
-        except Exception, error:
-            raise MethodCallError("Remote exception %s" % str(error))
+        method_func = getattr(self.factory.object, method)
+        result = maybeDeferred(method_func, *args, **kwargs)
+
+        # If the Deferred was already fired, we can return its result
+        if result.called:
+            if isinstance(result.result, Failure):
+                failure = str(result.result.value)
+                result.addErrback(lambda error: None) # Stop propagating
+                raise MethodCallError(failure)
+            return {"result": self._check_result(result.result)}
+
+        uuid = str(uuid4())
+        result.addBoth(self.send_deferred_response, uuid)
+        return {"result": None, "deferred": uuid}
+
+    def _check_result(self, result):
+        """Check that the C{result} we're about to return is serializable.
+
+        @return: The C{result} itself if valid.
+        @raises: L{MethodCallError} if C{result} is not serializable.
+        """
         if not MethodCallArgument.check(result):
             raise MethodCallError("Non-serializable result")
-        return {"result": result}
+        return result
+
+    def send_deferred_response(self, result, uuid):
+        """Send a L{DeferredResponse} for the deferred with given C{uuid}.
+
+        This is called when the result of a L{Deferred} returned by an
+        object's method becomes available. A L{DeferredResponse} notifying
+        such result (either success or failure) is sent to the peer.
+        """
+        kwargs = {"uuid": uuid}
+        if isinstance(result, Failure):
+            kwargs["failure"] = str(result.value)
+        else:
+            kwargs["result"] = self._check_result(result)
+        self.callRemote(DeferredResponse, **kwargs)
 
 
 class MethodCallClientProtocol(AMP):
-    """Calls methods of a remote object over L{AMP}."""
+    """Calls methods of a remote object over L{AMP}.
+
+    @note: If the remote method returns a deferred, the associated local
+        deferred returned by L{send_method_call} will result in the same
+        callback value of the remote deferred.
+    @cvar timeout: A timeout for remote methods returning L{Deferred}s, if a
+        response for the deferred is not received within this amount of
+        seconds, the remote method call will errback with a L{MethodCallError}.
+    """
+    timeout = 60
+
+    def __init__(self):
+        AMP.__init__(self)
+        self._pending_responses = {}
+
+    @DeferredResponse.responder
+    def receive_deferred_response(self, uuid, result, failure):
+        """Receive the deferred L{MethodCall} response.
+
+        @param uuid: The id of the L{MethodCall} we're getting the result of.
+        @param result: The result of the associated deferred if successful.
+        @param failure: The failure message of the deferred if it failed.
+        """
+        self.fire_pending_response(uuid, result, failure)
+        return {}
+
+    def fire_pending_response(self, uuid, result, failure):
+        """Fire the L{Deferred} associated with a pending response.
+
+        @param uuid: The id of the L{MethodCall} we're getting the result of.
+        @param result: The result of the associated deferred if successful.
+        @param failure: The failure message of the deferred if it failed.
+        """
+        try:
+            deferred, call = self._pending_responses.pop(uuid)
+        except KeyError:
+            # Late response for a request that has timeout, just ignore it
+            return
+        if not call.called:
+            call.cancel()
+        if failure is None:
+            deferred.callback({"result": result})
+        else:
+            deferred.errback(MethodCallError(failure))
+
+    def handle_response(self, response):
+        """Handle a L{MethodCall} response.
+
+        If the response is tagged as deferred, it will be queued as pending,
+        and a L{Deferred} is returned, which will be fired as soon as the
+        final response becomes available, or the timeout is reached.
+        """
+        if response["deferred"]:
+            uuid = response["deferred"]
+            deferred = Deferred()
+            call = self.factory.reactor.callLater(self.timeout,
+                                                  self.fire_pending_response,
+                                                  uuid, None, "timeout")
+            self._pending_responses[uuid] = (deferred, call)
+            return deferred
+
+        return response
 
     def send_method_call(self, method, args=[], kwargs={}):
         """Send a L{MethodCall} command with the given arguments.
@@ -87,8 +191,12 @@
         @param args: The positional arguments to pass to the remote method.
         @param kwargs: The keyword arguments to pass to the remote method.
         """
-        return self.callRemote(MethodCall,
-                               method=method, args=args, kwargs=kwargs)
+        result = self.callRemote(MethodCall,
+                                 method=method, args=args, kwargs=kwargs)
+        # The result can be C{None} only if the requested command is a
+        # DeferredResponse, which has requiresAnswer set to False
+        if result is not None:
+            return result.addCallback(self.handle_response)
 
 
 class MethodCallServerFactory(ServerFactory):
@@ -111,16 +219,18 @@
 
     def __init__(self, reactor, notifier):
         """
-        @param reactor: The reactor used to schedule connection callbacks.
+        @param reactor: The reactor used by the created protocols to schedule
+            notifications and timeouts.
         @param notifier: A function that will be called when the connection is
             established. It will be passed the protocol instance as argument.
         """
-        self._reactor = reactor
+        self.reactor = reactor
         self._notifier = notifier
 
     def buildProtocol(self, addr):
         protocol = self.protocol()
-        self._reactor.callLater(0, self._notifier, protocol)
+        protocol.factory = self
+        self.reactor.callLater(0, self._notifier, protocol)
         return protocol
 
 
 
=== modified file 'landscape/lib/tests/test_amp.py'
--- landscape/lib/tests/test_amp.py	2010-01-20 08:33:12 +0000
+++ landscape/lib/tests/test_amp.py	2010-01-25 08:30:27 +0000
@@ -1,9 +1,11 @@
 from twisted.internet import reactor
+from twisted.internet.defer import Deferred
 from twisted.internet.protocol import ClientCreator
 
 from landscape.lib.amp import (
-    MethodCallError, MethodCallServerProtocol, MethodCallClientProtocol,
-    MethodCallServerFactory, RemoteObject, RemoteObjectCreator)
+    MethodCallError, MethodCallServerProtocol,
+    MethodCallClientProtocol, MethodCallServerFactory,
+    MethodCallClientFactory, RemoteObject, RemoteObjectCreator)
 from landscape.tests.helpers import LandscapeTest
 
 
@@ -60,6 +62,24 @@
     def translate(self, word):
         raise WordsException("Unknown word")
 
+    def google(self, word):
+        deferred = Deferred()
+        if word == "Landscape":
+            reactor.callLater(0.01, lambda: deferred.callback("Cool!"))
+        elif word == "Easy query":
+            deferred.callback("Done!")
+        elif word == "Weird stuff":
+            reactor.callLater(0.01, lambda: deferred.errback(Exception("bad")))
+        elif word == "Censored":
+            deferred.errback(Exception("very bad"))
+        elif word == "Long query":
+            # Do nothing, the deferred won't be fired at all
+            pass
+        elif word == "Slowish query":
+            # Fire the result after a while.
+            reactor.callLater(0.05, lambda: deferred.callback("Done!"))
+        return deferred
+
 
 class WordsProtocol(MethodCallServerProtocol):
 
@@ -72,7 +92,8 @@
                "multiply_alphabetically",
                "translate",
                "meaning_of_life",
-               "guess"]
+               "guess",
+               "google"]
 
 
 class MethodCallProtocolTest(LandscapeTest):
@@ -114,7 +135,8 @@
         result = self.protocol.send_method_call(method="empty",
                                                 args=[],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": None})
+        return self.assertSuccess(result, {"result": None,
+                                           "deferred": None})
 
     def test_with_return_value(self):
         """
@@ -124,7 +146,8 @@
         result = self.protocol.send_method_call(method="motd",
                                                 args=[],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "Words are cool"})
+        return self.assertSuccess(result, {"result": "Words are cool",
+                                           "deferred": None})
 
     def test_with_one_argument(self):
         """
@@ -134,7 +157,8 @@
         result = self.protocol.send_method_call(method="capitalize",
                                                 args=["john"],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "John"})
+        return self.assertSuccess(result, {"result": "John",
+                                           "deferred": None})
 
     def test_with_boolean_return_value(self):
         """
@@ -143,7 +167,8 @@
         result = self.protocol.send_method_call(method="is_short",
                                                 args=["hi"],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": True})
+        return self.assertSuccess(result, {"result": True,
+                                           "deferred": None})
 
     def test_with_many_arguments(self):
         """
@@ -152,7 +177,8 @@
         result = self.protocol.send_method_call(method="concatenate",
                                                 args=["You ", "rock"],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "You rock"})
+        return self.assertSuccess(result, {"result": "You rock",
+                                           "deferred": None})
 
     def test_with_default_arguments(self):
         """
@@ -162,7 +188,8 @@
         result = self.protocol.send_method_call(method="lower_case",
                                                 args=["OHH"],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "ohh"})
+        return self.assertSuccess(result, {"result": "ohh",
+                                           "deferred": None})
 
     def test_with_overriden_default_arguments(self):
         """
@@ -173,7 +200,8 @@
         result = self.protocol.send_method_call(method="lower_case",
                                                 args=["OHH"],
                                                 kwargs={"index": 2})
-        return self.assertSuccess(result, {"result": "OHh"})
+        return self.assertSuccess(result, {"result": "OHh",
+                                           "deferred": None})
 
     def test_with_dictionary_arguments(self):
         """
@@ -183,7 +211,8 @@
                                                        "alphabetically",
                                                 args=[{"foo": 2, "bar": 3}],
                                                 kwargs={})
-        return self.assertSuccess(result, {"result": "barbarbarfoofoo"})
+        return self.assertSuccess(result, {"result": "barbarbarfoofoo",
+                                           "deferred": None})
 
     def test_with_non_serializable_return_value(self):
         """
@@ -211,17 +240,19 @@
     def setUp(self):
         super(RemoteObjectTest, self).setUp()
         socket = self.mktemp()
-        factory = MethodCallServerFactory(Words())
-        factory.protocol = WordsProtocol
-        self.port = reactor.listenUNIX(socket, factory)
+        server_factory = MethodCallServerFactory(Words())
+        server_factory.protocol = WordsProtocol
+        self.port = reactor.listenUNIX(socket, server_factory)
 
         def set_remote(protocol):
             self.protocol = protocol
             self.words = RemoteObject(protocol)
 
-        connector = ClientCreator(reactor, MethodCallClientProtocol)
-        connected = connector.connectUNIX(socket)
-        return connected.addCallback(set_remote)
+        connected = Deferred()
+        connected.addCallback(set_remote)
+        client_factory = MethodCallClientFactory(reactor, connected.callback)
+        reactor.connectUNIX(socket, client_factory)
+        return connected
 
     def tearDown(self):
         self.protocol.transport.loseConnection()
@@ -319,6 +350,62 @@
         result = self.words.guess("word", "cool", value=4)
         return self.assertSuccess(result, "Guessed!")
 
+    def test_with_success_full_deferred(self):
+        """
+        If the target object method returns a L{Deferred}, it is handled
+        transparently.
+        """
+        result = self.words.google("Landscape")
+        return self.assertSuccess(result, "Cool!")
+
+    def test_with_failing_deferred(self):
+        """
+        If the target object method returns a failing L{Deferred}, a
+        L{MethodCallError} is raised.
+        """
+        result = self.words.google("Weird stuff")
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_already_callback_deferred(self):
+        """
+        The target object method can return an already fired L{Deferred}.
+        """
+        result = self.words.google("Easy query")
+        return self.assertSuccess(result, "Done!")
+
+    def test_with_already_errback_deferred(self):
+        """
+        If the target object method can return an already failed L{Deferred}.
+        """
+        result = self.words.google("Censored")
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_deferred_timeout(self):
+        """
+        If the peer protocol doesn't send a response for a deferred within
+        the given timeout, the method call fails.
+        """
+        self.protocol.timeout = 0.1
+        result = self.words.google("Long query")
+        return self.assertFailure(result, MethodCallError)
+
+    def test_with_late_response(self):
+        """
+        If the peer protocol sends a late response for a request that has
+        already timeout, that response is ignored.
+        """
+        self.protocol.timeout = 0.01
+        result = self.words.google("Slowish query")
+        self.assertFailure(result, MethodCallError)
+
+        def assert_late_response_is_handled(ignored):
+            deferred = Deferred()
+            # We wait a bit to be sure that the late response gets delivered
+            reactor.callLater(0.1, lambda: deferred.callback(None))
+            return deferred
+
+        return result.addCallback(assert_late_response_is_handled)
+
 
 class RemoteObjectCreatorTest(LandscapeTest):