pc_rpc.Server: Marshal exception on unencodable return value

This effectively just moves the pyon.encode() calls into the
try/except block in _process_action().

This way, the client acutally gets an exception if the return
value of a method call is not PYON-serializable instead of the
connection just being dropped, and the exception being swallowed
on the server side.

This is very useful to debug things in practice, as
 1) "… is not PYON-serializable" immediately points physicists who
    just use sipyco as a black box to what the error is, whereas a
    plain ConnectionResetError is rather inscrutable, and
 2) the exception message includes a string representation of the
    intended return value, further making it clear what is going on.

sipyco/pc_rpc.py

@@ -534,32 +534,39 @@ async def _process_action(self, target, obj):
                         },
                         "Terminate the server.")
                 logger.debug("RPC docs for %s: %s", target, doc)
-                return {"status": "ok", "ret": doc}
+                return doc
             elif obj["action"] == "call":
                 logger.debug("calling %s", _PrettyPrintCall(obj))
                 if (self.builtin_terminate and obj["name"] ==
                         "terminate"):
                     self._terminate_request.set()
-                    return {"status": "ok", "ret": None}
+                    return None
                 else:
                     method = getattr(target, obj["name"])
                     ret = method(*obj["args"], **obj["kwargs"])
                     if inspect.iscoroutine(ret):
                         ret = await ret
-                    return {"status": "ok", "ret": ret}
+                    return ret
             else:
                 raise ValueError("Unknown action: {}"
                                  .format(obj["action"]))
+        finally:
+            if self._noparallel is not None:
+                self._noparallel.release()
+
+    async def _process_and_pyonize(self, target, obj):
+        try:
+            return pyon.encode({
+                "status": "ok",
+                "ret": await self._process_action(target, obj)
+            })
         except (asyncio.CancelledError, SystemExit):
             raise
         except:
-            return {
+            return pyon.encode({
                 "status": "failed",
                 "exception": current_exc_packed()
-            }
-        finally:
-            if self._noparallel is not None:
-                self._noparallel.release()
+            })
 
     async def _handle_connection_cr(self, reader, writer):
         try:
@@ -595,8 +602,9 @@ async def _handle_connection_cr(self, reader, writer):
                 line = await reader.readline()
                 if not line:
                     break
-                reply = await self._process_action(target, pyon.decode(line.decode()))
-                writer.write((pyon.encode(reply) + "\n").encode())
+                reply = await self._process_and_pyonize(target,
+                    pyon.decode(line.decode()))
+                writer.write((reply + "\n").encode())
         except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError):
             # May happens on Windows when client disconnects
             pass

sipyco/test/test_pc_rpc.py

@@ -82,6 +82,8 @@ async def _asyncio_echo(self, target):
             self.assertEqual(test_object, test_object_back)
             test_object_back = await remote.async_echo(test_object)
             self.assertEqual(test_object, test_object_back)
+            with self.assertRaises(TypeError):
+                await remote.return_unserializable()
             with self.assertRaises(AttributeError):
                 await remote.non_existing_method
             await remote.terminate()
@@ -145,6 +147,10 @@ async def async_echo(self, x):
         await asyncio.sleep(0.01)
         return x
 
+    def return_unserializable(self):
+        # Arbitrary classes can't be PYON-serialized.
+        return Echo()
+
 
 def run_server():
     loop = asyncio.new_event_loop()