pc_rpc/Server: add built-in terminate

artiq/protocols/pc_rpc.py

@@ -240,7 +240,8 @@ class BestEffortClient:
     network errors are suppressed and connections are retried in the
     background.
 
-    RPC calls that failed because of network errors return ``None``.
+    RPC calls that failed because of network errors return ``None``. Other RPC
+    calls are blocking and return the correct value.
 
     :param firstcon_timeout: Timeout to use during the first (blocking)
         connection attempt at object initialization.
@@ -397,12 +398,19 @@ class Server(_AsyncioServer):
         exposed to the client. Keys are names identifying each object.
         Clients select one of these objects using its name upon connection.
     :param id_parameters: An optional human-readable string giving more
-        information about the parameters of the server.
+        information about the server.
+    :param builtin_terminate: If set, the server provides a built-in
+        ``terminate`` method that unblocks any tasks waiting on
+        ``wait_terminate``. This is useful to handle server termination
+        requests from clients.
     """
-    def __init__(self, targets, id_parameters=None):
+    def __init__(self, targets, id_parameters=None, builtin_terminate=False):
         _AsyncioServer.__init__(self)
         self.targets = targets
         self.id_parameters = id_parameters
+        self.builtin_terminate = builtin_terminate
+        if builtin_terminate:
+            self._terminate_request = asyncio.Event()
 
     @asyncio.coroutine
     def _handle_connection_cr(self, reader, writer):
@@ -448,9 +456,13 @@ class Server(_AsyncioServer):
                         obj = {"status": "ok", "ret": doc}
                     elif obj["action"] == "call":
                         logger.debug("calling %s", _PrettyPrintCall(obj))
-                        method = getattr(target, obj["name"])
+                        if self.builtin_terminate and obj["name"] == "terminate":
-                        ret = method(*obj["args"], **obj["kwargs"])
+                            self._terminate_request.set()
-                        obj = {"status": "ok", "ret": ret}
+                            obj = {"status": "ok", "ret": None}
+                        else:
+                            method = getattr(target, obj["name"])
+                            ret = method(*obj["args"], **obj["kwargs"])
+                            obj = {"status": "ok", "ret": ret}
                     else:
                         raise ValueError("Unknown action: {}"
                                          .format(obj["action"]))
@@ -462,18 +474,23 @@ class Server(_AsyncioServer):
         finally:
             writer.close()
 
+    @asyncio.coroutine
+    def wait_terminate(self):
+        yield from self._terminate_request.wait()
+
 
 def simple_server_loop(targets, host, port, id_parameters=None):
-    """Runs a server until an exception is raised (e.g. the user hits Ctrl-C).
+    """Runs a server until an exception is raised (e.g. the user hits Ctrl-C)
+    or termination is requested by a client.
 
     See ``Server`` for a description of the parameters.
     """
     loop = asyncio.get_event_loop()
     try:
-        server = Server(targets, id_parameters)
+        server = Server(targets, id_parameters, True)
         loop.run_until_complete(server.start(host, port))
         try:
-            loop.run_forever()
+            loop.run_until_complete(server.wait_terminate())
         finally:
             loop.run_until_complete(server.stop())
     finally: