Repository: opensistemas-hub/osbrain
Pull request #112 (2017-04-11 10:44:31)

Allow agents and name servers to spawn child processes
This implements basic inception capabilities. There is no API defined for explicitly running agents and handling children processes (currently that would be the user's responsibility). That should probably be discussed in a new issue.

codecov[bot]: # [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/112?src=pr&el=h1) Report
> Merging [#112](https://codecov.io/gh/opensistemas-hub/osbrain/pull/112?src=pr&el=desc) into [master](https://codecov.io/gh/opensistemas-hub/osbrain/commit/c8bb10397235bb35715eeeee493e11fae20e5fdf?src=pr&el=desc) will **not change** coverage.
> The diff coverage is `75%`.

[![Impacted file tree graph](https://codecov.io/gh/opensistemas-hub/osbrain/pull/112/graphs/tree.svg?height=150&width=650&token=LfvMoNOrOc&src=pr)](https://codecov.io/gh/opensistemas-hub/osbrain/pull/112?src=pr&el=tree)

```diff
@@           Coverage Diff           @@
##           master     #112   +/-   ##
=======================================
  Coverage   97.43%   97.43%           
=======================================
  Files           8        8           
  Lines        1171     1171           
  Branches      173      173           
=======================================
  Hits         1141     1141           
  Misses         21       21           
  Partials        9        9
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/112?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/agent.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/112?src=pr&el=tree#diff-b3NicmFpbi9hZ2VudC5weQ==) | `95.77% <75%> (ø)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/112?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/112?src=pr&el=footer). Last update [c8bb103...f26a783](https://codecov.io/gh/opensistemas-hub/osbrain/pull/112?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
Peque: @Flood1993 It is because the `multiprocessing.Process` class actually [has a `daemon` flag](https://docs.python.org/3/library/multiprocessing.html#multiprocessing.Process.daemon) that we were unintentionally overriding... :confused:

That is why spawning new processes was raising as exception, as `.daemon` was not False.

Patch:
diff --git a/osbrain/__init__.py b/osbrain/__init__.py
index 1145c61..2b0515e 100644
--- a/osbrain/__init__.py
+++ b/osbrain/__init__.py
@@ -37,7 +37,7 @@ config['IPC_DIR'].mkdir(exist_ok=True, parents=True)
 __version__ = '0.4.2'
 
 from .agent import Agent, AgentProcess, run_agent
-from .nameserver import run_nameserver
+from .nameserver import NameServer, run_nameserver
 from .proxy import Proxy, NSProxy
 from .address import SocketAddress, AgentAddress
 from .logging import Logger, run_logger
diff --git a/osbrain/agent.py b/osbrain/agent.py
index 8dd5e21..4ea8527 100644
--- a/osbrain/agent.py
+++ b/osbrain/agent.py
@@ -1507,7 +1507,7 @@ class AgentProcess(multiprocessing.Process):
                  transport=None, base=Agent):
         super().__init__()
         self.name = name
-        self.daemon = None
+        self._daemon = None
         self.host, self.port = address_to_host_port(addr)
         if self.port is None:
             self.port = 0
@@ -1525,7 +1525,7 @@ class AgentProcess(multiprocessing.Process):
 
         try:
             ns = NSProxy(self.nsaddr)
-            self.daemon = Pyro4.Daemon(self.host, self.port)
+            self._daemon = Pyro4.Daemon(self.host, self.port)
         except Exception:
             self.queue.put(format_exception())
             return
@@ -1534,12 +1534,12 @@ class AgentProcess(multiprocessing.Process):
         self.agent = self.base(name=self.name, host=self.host,
                                serializer=self.serializer,
                                transport=self.transport)
-        uri = self.daemon.register(self.agent)
+        uri = self._daemon.register(self.agent)
         ns.register(self.name, uri)
         ns.release()
 
-        self.daemon.requestLoop(lambda: not self.shutdown_event.is_set())
-        self.daemon.unregister(self.agent)
+        self._daemon.requestLoop(lambda: not self.shutdown_event.is_set())
+        self._daemon.unregister(self.agent)
 
         self._teardown()
 
@@ -1556,7 +1556,7 @@ class AgentProcess(multiprocessing.Process):
                 raise
         finally:
             self.agent._killed = True
-            self.daemon.close()
+            self._daemon.close()
 
     def start(self):
         super().start()
@@ -1568,8 +1568,8 @@ class AgentProcess(multiprocessing.Process):
 
     def kill(self):
         self.shutdown_event.set()
-        if self.daemon:
-            self.daemon.shutdown()
+        if self._daemon:
+            self._daemon.shutdown()
 
     def sigint_handler(self, signal, frame):
         """
diff --git a/osbrain/nameserver.py b/osbrain/nameserver.py
index 8737c49..b7d78b9 100644
--- a/osbrain/nameserver.py
+++ b/osbrain/nameserver.py
@@ -56,16 +56,15 @@ class NameServer(Pyro4.naming.NameServer):
         self._pyroDaemon.shutdown()
 
 
-Pyro4.naming.NameServer = NameServer
-
-
 class NameServerProcess(multiprocessing.Process):
     """
     Name server class. Instances of a name server are system processes which
     can be run independently.
     """
-    def __init__(self, addr=None):
+    def __init__(self, addr=None, base=NameServer):
         super().__init__()
+        self._daemon = None
+        self.base = base
         if isinstance(addr, int):
             addr = '127.0.0.1:%s' % addr
         self.addr = addr
@@ -78,18 +77,19 @@ class NameServerProcess(multiprocessing.Process):
         # Capture SIGINT
 
         try:
-            self.daemon = Pyro4.naming.NameServerDaemon(self.host, self.port)
+            Pyro4.naming.NameServer = self.base
+            self._daemon = Pyro4.naming.NameServerDaemon(self.host, self.port)
         except Exception:
             self.queue.put(format_exception())
             return
         self.queue.put('STARTED')
-        self.uri = self.daemon.uriFor(self.daemon.nameserver)
+        self.uri = self._daemon.uriFor(self._daemon.nameserver)
         self.host = self.uri.host
         self.port = self.uri.port
         self.addr = SocketAddress(self.host, self.port)
-        internal_uri = self.daemon.uriFor(self.daemon.nameserver, nat=False)
+        internal_uri = self._daemon.uriFor(self._daemon.nameserver, nat=False)
         bcserver = None
-        hostip = self.daemon.sock.getsockname()[0]
+        hostip = self._daemon.sock.getsockname()[0]
         # Start broadcast responder
         bcserver = BroadcastServer(internal_uri)
         sys.stdout.write(
@@ -97,13 +97,13 @@ class NameServerProcess(multiprocessing.Process):
         sys.stdout.flush()
         bcserver.runInThread()
         sys.stdout.write(
-            "NS running on %s (%s)\n" % (self.daemon.locationStr, hostip))
+            "NS running on %s (%s)\n" % (self._daemon.locationStr, hostip))
         sys.stdout.write("URI = %s\n" % self.uri)
         sys.stdout.flush()
         try:
-            self.daemon.requestLoop(lambda: not self.shutdown_event.is_set())
+            self._daemon.requestLoop(lambda: not self.shutdown_event.is_set())
         finally:
-            self.daemon.close()
+            self._daemon.close()
             if bcserver is not None:
                 bcserver.close()
         sys.stdout.write("NS shut down.\n")
@@ -150,7 +150,7 @@ class NameServerProcess(multiprocessing.Process):
 
 
 def random_nameserver_process(host='127.0.0.1', port_start=10000,
-                              port_stop=20000, timeout=3.):
+                              port_stop=20000, timeout=3., base=NameServer):
     """
     Start a random NameServerProcess.
 
@@ -175,7 +175,7 @@ def random_nameserver_process(host='127.0.0.1', port_start=10000,
             # Bind to random port
             port = random.randrange(port_start, port_stop + 1)
             addr = SocketAddress(host, port)
-            nameserver = NameServerProcess(addr)
+            nameserver = NameServerProcess(addr, base=base)
             nameserver.start()
             return nameserver
         except RuntimeError as error:
@@ -184,7 +184,7 @@ def random_nameserver_process(host='127.0.0.1', port_start=10000,
             raise exception
 
 
-def run_nameserver(addr=None):
+def run_nameserver(addr=None, base=NameServer):
     """
     Ease the name server creation process.
 
@@ -202,7 +202,7 @@ def run_nameserver(addr=None):
         A proxy to the name server.
     """
     if not addr:
-        addr = random_nameserver_process().addr
+        addr = random_nameserver_process(base=base).addr
     else:
-        NameServerProcess(addr).start()
+        NameServerProcess(addr, base=base).start()
     return NSProxy(addr)


Test patch:
diff --git a/osbrain/tests/test_agent.py b/osbrain/tests/test_agent.py
index 62c952a..1fb2b55 100644
--- a/osbrain/tests/test_agent.py
+++ b/osbrain/tests/test_agent.py
@@ -1,6 +1,7 @@
 """
 Test file for agents.
 """
+import multiprocessing
 import time
 from uuid import uuid4
 from threading import Timer
@@ -536,3 +537,20 @@ def test_agent_stop(nsproxy):
     while agent.get_attr('running'):
         time.sleep(0.01)
     assert not agent.get_attr('running')
+
+
+def test_agent_spawn_process(nsproxy):
+    """
+    An agent should be able to spawn child processes.
+
+    It is a way to make sure agents are run as non-daemonic processes, which
+    are not allowed to have children.
+    """
+    class Spawner(Agent):
+        def spawn_process(self):
+            p = multiprocessing.Process()
+            p.start()
+            return True
+
+    agent = run_agent('a0', base=Spawner)
+    assert agent.spawn_process()
diff --git a/osbrain/tests/test_nameserver.py b/osbrain/tests/test_nameserver.py
index 7f334a6..4066c50 100644
--- a/osbrain/tests/test_nameserver.py
+++ b/osbrain/tests/test_nameserver.py
@@ -4,6 +4,7 @@ Test file for nameserver.
 import os
 import time
 import random
+import multiprocessing
 from threading import Timer
 
 import pytest
@@ -12,6 +13,7 @@ from osbrain import SocketAddress
 from osbrain import run_agent
 from osbrain import run_nameserver
 from osbrain import Agent
+from osbrain import NameServer
 from osbrain import NSProxy
 from osbrain.nameserver import NameServerProcess
 from osbrain.nameserver import random_nameserver_process
@@ -231,3 +233,35 @@ def test_nameserver_permissionerror():
         run_nameserver('127.0.0.1:22')
     assert 'PermissionError' in str(error.value)
     assert 'Permission denied' in str(error.value)
+
+
+def test_run_nameserver_base():
+    """
+    The `run_nameserver` function should accept a `base` parameter to specify
+    the base NameServer class.
+    """
+    class BobMarley(NameServer):
+        def get_up(self):
+            return 'stand up!'
+
+    ns = run_nameserver(base=BobMarley)
+    assert ns.get_up() == 'stand up!'
+    ns.shutdown()
+
+
+def test_nameserver_spawn_process(nsproxy):
+    """
+    A name server should be able to spawn child processes.
+
+    It is a way to make sure name servers are run as non-daemonic processes,
+    which are not allowed to have children.
+    """
+    class Spawner(NameServer):
+        def spawn_process(self):
+            p = multiprocessing.Process()
+            p.start()
+            return True
+
+    ns = run_nameserver(base=Spawner)
+    assert ns.spawn_process()
+    ns.shutdown()