Merge branch 'yan-976-constrain-resource-overusage'

Signed-off-by: Rodrigo Tobar <rtobar@icrar.org>

daliuge-engine/dlg/rpc.py

@@ -275,13 +275,28 @@ class DropProxy(object):
     """
 
     def __init__(self, rpc_client, hostname, port, sessionId, uid):
-        self.rpc_client = ZeroRPCClient()
+        # The current version of multiprocessing support creates an RPCClient
+        # per DropProxy, disregarding the rpc_client parameter given here.
+        # This uses too many resources though, but is only needed if the NM is
+        # instructed to use multiprocessing support. To avoid this resource
+        # over-usage we then detect if the given rpc_client (an instance of
+        # NodeManagerBase) has been started with multiprocessing support (which
+        # is confusingly bound to there being a *thread* pool too) and only if
+        # we detect the situation we create our own RPCClient; otherwise we use
+        # the given rpc_client as is.
+        if hasattr(rpc_client, "_threadpool") and rpc_client._threadpool:
+            self.rpc_client = ZeroRPCClient()
+            self._own_rpc_client = True
+        else:
+            self.rpc_client = rpc_client
+            self._own_rpc_client = False
         self.hostname = hostname
         self.port = port
         self.session_id = sessionId
         self.uid = uid
         logger.debug("Created %r", self)
-        self.rpc_client.start()
+        if self._own_rpc_client:
+            self.rpc_client.start()
 
     def handleEvent(self, evt):
         pass
@@ -304,4 +319,5 @@ def __repr__(self):
         )
 
     def __del__(self):
-        self.rpc_client.shutdown()
+        if self._own_rpc_client:
+            self.rpc_client.shutdown()