Merge pull request #108 from OFFIS-DAI/feature-auto-port

Feature auto port

.github/workflows/test-mango.yml

@@ -13,12 +13,54 @@ permissions:
   contents: read
 
 jobs:
-  build:
+  build-mac:
+    runs-on: macOS-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+        cache: 'pip'
+        cache-dependency-path: '**/setup.py'
+    - name: Install dependencies
+      run: |
+        pip install virtualenv
+        virtualenv venv
+        source venv/bin/activate
+        pip3 install -U sphinx
+        pip3 install -r docs/requirements.txt
+        pip3 install -r requirements.txt
+        pip3 install -e .
+        brew install mosquitto
+        brew services start mosquitto
+        pip3 install pytest coverage ruff
+    - name: Lint with ruff
+      run: |
+        # stop the build if there are Python syntax errors or undefined names
+        source venv/bin/activate
+        ruff check .
+        ruff format --check .
+    - name: Doctests
+      run: |
+        source venv/bin/activate
+        make -C docs doctest
+    - name: Test+Coverage
+      run: |
+        source venv/bin/activate
+        coverage run -m pytest
+        coverage report
+
+  build-linux:
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
     steps:
     - uses: actions/checkout@v4
     - name: Set up Python

docs/source/agents-container.rst

@@ -28,6 +28,7 @@ a simple but fast network protocol.
         codec: Codec = None,
         clock: Clock = None,
         copy_internal_messages: bool = False,
+        auto_port=False,
         **kwargs: dict[str, Any],
     ) -> Container:
 
@@ -41,14 +42,14 @@ A simple container, that uses plain tcp for message exchange can be created as f
     from mango import create_tcp_container
 
     def get_simple_container():
-        container = create_tcp_container(addr=('localhost', 5555))
+        container = create_tcp_container(addr=('127.0.0.1', 5555))
         return container
 
     print(get_simple_container().addr)
 
 .. testoutput::
 
-    ('localhost', 5555)
+    ('127.0.0.1', 5555)
 
 The container type depends totally on the factory method you invoke. Every supported type has its own class backing
 the functionality.
@@ -66,7 +67,7 @@ asynchronous context manager, which we provide by invoking :meth:`mango.activate
     from mango import create_tcp_container, activate
 
     async def start_container():
-        container = create_tcp_container(addr=('localhost', 5555))
+        container = create_tcp_container(addr=('127.0.0.1', 5555))
 
         async with activate(container) as c:
             print("The container is activated now!")
@@ -105,7 +106,7 @@ Note that, custom agents that inherit from the ``Agent`` class have to call ``su
         pass
 
     async def create_and_register_agent():
-        container = create_tcp_container(addr=('localhost', 5555))
+        container = create_tcp_container(addr=('127.0.0.1', 5555))
 
         agent = container.register(MyAgent(), suggested_aid="CustomAgent")
         return agent

docs/source/codecs.rst

@@ -113,8 +113,8 @@ All that is left to do now is to pass our codec to the container. This is done d
 
         # codecs can be passed directly to the container
         # if no codec is passed a new instance of JSON() is created
-        sending_container = create_tcp_container(addr=("localhost", 5556), codec=codec)
-        receiving_container = create_tcp_container(addr=("localhost", 5555), codec=codec)
+        sending_container = create_tcp_container(addr=("127.0.0.1", 5556), codec=codec)
+        receiving_container = create_tcp_container(addr=("127.0.0.1", 5555), codec=codec)
         receiving_agent = receiving_container.register(SimpleReceivingAgent())
 
         async with activate(sending_container, receiving_container):

docs/source/getting_started.rst

@@ -56,12 +56,12 @@ found in :doc:`Agents and container <agents-container>`
 
     # Containers have to be created using a factory method
     # Other container types are available through create_mqtt_container and create_ec_container
-    container = create_tcp_container(addr=('localhost', 5555))
+    container = create_tcp_container(addr=('127.0.0.1', 5555))
     print(container.addr)
 
 .. testoutput::
 
-    ('localhost', 5555)
+    ('127.0.0.1', 5555)
 
 
 This is how a tcp container is created. While container creation, it is possible to set the codec, the address information (depending on the type)
@@ -103,12 +103,12 @@ The following script will create a RepeatingAgent, register it, and let it run w
         async with activate(first_container) as container:
             await asyncio.sleep(duration)
 
-    asyncio.run(run_container_and_agent(addr=('localhost', 5555), duration=0.05))
+    asyncio.run(run_container_and_agent(addr=('127.0.0.1', 5555), duration=0.05))
 
 .. testoutput::
 
     Creating a RepeatingAgent. At this point self.addr=None
-    The agent has been registered to a container: AgentAddress(protocol_addr=('localhost', 5555), aid='agent0')!
+    The agent has been registered to a container: AgentAddress(protocol_addr=('127.0.0.1', 5555), aid='agent0')!
     All containers have been activated!
 
 In this example no messages are sent, nor does the Agent do anything, but the call order of the hook-in functions is clearly visible.
@@ -142,7 +142,7 @@ to another agent:
         async with activate(first_container) as container:
             await first_hello_agent.greet(second_hello_agent.addr)
 
-    asyncio.run(run_container_and_agent(addr=('localhost', 5555), duration=0.05))
+    asyncio.run(run_container_and_agent(addr=('127.0.0.1', 5555), duration=0.05))
 
 .. testoutput::
 
@@ -198,13 +198,13 @@ a RepeatingAgent and let them run.
             await asyncio.sleep(.1)
 
     asyncio.run(run_container_and_two_agents(
-        first_addr=('localhost', 5555), second_addr=('localhost', 5556))
+        first_addr=('127.0.0.1', 5555), second_addr=('127.0.0.1', 5556))
     )
 
 .. testoutput::
 
     Creating a RepeatingAgent. At this point self.addr=None
-    The agent has been registered to a container: AgentAddress(protocol_addr=('localhost', 5555), aid='agent0')!
+    The agent has been registered to a container: AgentAddress(protocol_addr=('127.0.0.1', 5555), aid='agent0')!
     All containers have been activated!
     Received a message with the following content: Hello world!!
 

docs/source/role-api.rst

@@ -175,8 +175,8 @@ determine the message dispatch order (lower number = earlier execution, default=
 
     async def show_handle_sub():
         my_composed_agent = agent_composed_of(MyRole())
-        async with run_with_tcp(1, my_composed_agent) as cl:
-            await cl[0].send_message(Ping(), my_composed_agent.addr)
+        async with run_with_tcp(1, my_composed_agent) as container:
+            await container.send_message(Ping(), my_composed_agent.addr)
 
     asyncio.run(show_handle_sub())
 

docs/source/scheduling.rst

@@ -196,8 +196,8 @@ In the following a simple example is shown.
   from mango import DistributedClockAgent, DistributedClockManager, create_tcp_container, activate, ExternalClock
 
   async def main():
-    container_man = create_tcp_container(("localhost", 1555), clock=ExternalClock())
-    container_ag = create_tcp_container(("localhost", 1556), clock=ExternalClock())
+    container_man = create_tcp_container(("127.0.0.1", 1555), clock=ExternalClock())
+    container_ag = create_tcp_container(("127.0.0.1", 1556), clock=ExternalClock())
 
     clock_agent = container_ag.register(DistributedClockAgent())
     clock_manager = container_man.register(DistributedClockManager(

docs/source/tutorial.rst

@@ -83,15 +83,15 @@ For this tutorial we will cover the tcp container.
 
     from mango import create_tcp_container
 
-    PV_CONTAINER_ADDRESS = ("localhost", 5555)
+    PV_CONTAINER_ADDRESS = ("127.0.0.1", 5555)
 
     pv_container = create_tcp_container(addr=PV_CONTAINER_ADDRESS)
 
     print(pv_container.addr)
 
 .. testoutput::
 
-    ('localhost', 5555)
+    ('127.0.0.1', 5555)
 
 Now we can create our agents. Agents always live inside a container and therefore need to be registered to the container.
 
@@ -110,7 +110,7 @@ Now we can create our agents. Agents always live inside a container and therefor
 
     Hello I am a PV agent!
     Hello I am a PV agent!
-    AgentAddress(protocol_addr=('localhost', 5555), aid='agent1')
+    AgentAddress(protocol_addr=('127.0.0.1', 5555), aid='agent1')
 
 For now, our agents and containers are purely passive entities. First, we need to activate the container to start
 the tcp server and its internal asynchronous behavior. In mango this can be done with :meth:`mango.activate` and the `async with` syntax.
@@ -143,7 +143,7 @@ is wrapped in the :class:`mango.AgentAddress` class and can be retrieved with :m
 
     Hello I am a PV agent!
     Hello I am a PV agent!
-    Received message with content: Hello, this is a simple message. and meta {'sender_id': 'agent2', 'sender_addr': ('localhost', 5555), 'receiver_id': 'agent3', 'network_protocol': 'tcp', 'priority': 0}.
+    Received message with content: Hello, this is a simple message. and meta {'sender_id': 'agent2', 'sender_addr': ('127.0.0.1', 5555), 'receiver_id': 'agent3', 'network_protocol': 'tcp', 'priority': 0}.
 
 
 *********************************
@@ -369,8 +369,8 @@ Lastly, we call all relevant instantiations and the run function within our main
 
     from mango import create_tcp_container, activate, Performatives
 
-    PV_CONTAINER_ADDRESS = ("localhost", 5555)
-    CONTROLLER_CONTAINER_ADDRESS = ("localhost", 5556)
+    PV_CONTAINER_ADDRESS = ("127.0.0.1", 5555)
+    CONTROLLER_CONTAINER_ADDRESS = ("127.0.0.1", 5556)
     PV_FEED_IN = {
         'PV Agent 0': 2.0,
         'PV Agent 1': 1.0,
@@ -464,8 +464,8 @@ Next, we need to create a codec, make our message objects known to it, and pass
 
     from mango import JSON
 
-    PV_CONTAINER_ADDRESS = ("localhost", 5555)
-    CONTROLLER_CONTAINER_ADDRESS = ("localhost", 5556)
+    PV_CONTAINER_ADDRESS = ("127.0.0.1", 5555)
+    CONTROLLER_CONTAINER_ADDRESS = ("127.0.0.1", 5556)
 
     my_codec = JSON()
     my_codec.add_serializer(*AskFeedInMsg.__serializer__())
@@ -675,8 +675,8 @@ also run tasks at specific times. For a full overview we refer to the documentat
 
     from mango import sender_addr, Role, RoleAgent, JSON, create_tcp_container, json_serializable, agent_composed_of
 
-    PV_CONTAINER_ADDRESS = ("localhost", 5555)
-    CONTROLLER_CONTAINER_ADDRESS = ("localhost", 5556)
+    PV_CONTAINER_ADDRESS = ("127.0.0.1", 5555)
+    CONTROLLER_CONTAINER_ADDRESS = ("127.0.0.1", 5556)
     PV_FEED_IN = {
         "PV Agent 0": 2.0,
         "PV Agent 1": 1.0,

mango/__init__.py

@@ -20,7 +20,6 @@
 from .messages.codecs import (
     json_serializable,
     JSON,
-    FastJSON,
     PROTOBUF,
     SerializationError,
 )

mango/container/factory.py

@@ -61,6 +61,7 @@ def create_tcp(
     codec: Codec = None,
     clock: Clock = None,
     copy_internal_messages: bool = False,
+    auto_port=False,
     **kwargs: dict[str, Any],
 ) -> Container:
     """
@@ -81,7 +82,7 @@ def create_tcp(
 
     # initialize TCPContainer
     return TCPContainer(
-        addr=addr,
+        addr=(addr[0], 0) if auto_port else addr,
         codec=codec,
         clock=clock,
         copy_internal_messages=copy_internal_messages,

mango/container/mp.py

@@ -2,7 +2,7 @@
 import logging
 import os
 from dataclasses import dataclass
-from multiprocessing import Event, Process
+from multiprocessing import get_context
 from multiprocessing.synchronize import Event as MultiprocessingEvent
 
 import dill  # noqa F401 # do not remove! Necessary for the auto loaded pickle reg extensions
@@ -102,16 +102,21 @@ def create_agent_process_environment(
     :type process_initialized_event: Event
     """
     asyncio.set_event_loop(asyncio.new_event_loop())
+    container_data.codec = dill.loads(container_data.codec)
+    agent_creator = dill.loads(agent_creator)
+    mirror_container_creator = dill.loads(mirror_container_creator)
 
     async def start_agent_loop():
         container = mirror_container_creator(
             container_data,
             asyncio.get_event_loop(),
-            message_pipe,
+            message_pipe.dup(),
             main_queue,
-            event_pipe,
+            event_pipe.dup(),
             terminate_event,
         )
+        message_pipe.close()
+        event_pipe.close()
         if asyncio.iscoroutinefunction(agent_creator):
             await agent_creator(container)
         else:
@@ -354,11 +359,12 @@ def __init__(
         self._active = False
         self._container = container
         self._mp_enabled = False
+        self._ctx = get_context("spawn")
 
     def _init_mp(self):
         # For agent multiprocessing support
         self._agent_processes = []
-        self._terminate_sub_processes = Event()
+        self._terminate_sub_processes = self._ctx.Event()
         self._pid_to_message_pipe = {}
         self._pid_to_pipe = {}
         self._pid_to_aids = {}
@@ -439,21 +445,21 @@ def create_agent_process(self, agent_creator, container, mirror_container_creato
             self._init_mp()
             self._active = True
 
-        from_pipe_message, to_pipe_message = aioduplex()
-        from_pipe, to_pipe = aioduplex()
-        process_initialized = Event()
+        from_pipe_message, to_pipe_message = aioduplex(self._ctx)
+        from_pipe, to_pipe = aioduplex(self._ctx)
+        process_initialized = self._ctx.Event()
         with to_pipe.detach() as to_pipe, to_pipe_message.detach() as to_pipe_message:
-            agent_process = Process(
+            agent_process = self._ctx.Process(
                 target=create_agent_process_environment,
                 args=(
                     ContainerData(
                         addr=container.addr,
-                        codec=container.codec,
+                        codec=dill.dumps(container.codec),
                         clock=container.clock,
                         kwargs=container._kwargs,
                     ),
-                    agent_creator,
-                    mirror_container_creator,
+                    dill.dumps(agent_creator),
+                    dill.dumps(mirror_container_creator),
                     to_pipe_message,
                     self._main_queue,
                     to_pipe,
@@ -465,19 +471,25 @@ def create_agent_process(self, agent_creator, container, mirror_container_creato
             agent_process.daemon = True
             agent_process.start()
 
-        self._pid_to_message_pipe[agent_process.pid] = from_pipe_message
-        self._pid_to_pipe[agent_process.pid] = from_pipe
+        from_pipe_message_dup = from_pipe_message.dup()
+        from_pipe_dup = from_pipe.dup()
+        from_pipe_message.close()
+        from_pipe.close()
+        self._pid_to_message_pipe[agent_process.pid] = from_pipe_message_dup
+        self._pid_to_pipe[agent_process.pid] = from_pipe_dup
         self._handle_process_events_tasks.append(
-            asyncio.create_task(self._handle_process_events(from_pipe))
+            asyncio.create_task(self._handle_process_events(from_pipe_dup))
         )
         self._handle_sp_messages_tasks.append(
-            asyncio.create_task(self._handle_process_message(from_pipe_message))
+            asyncio.create_task(self._handle_process_message(from_pipe_message_dup))
         )
 
         async def wait_for_process_initialized():
             while not process_initialized.is_set():
                 await asyncio.sleep(WAIT_STEP)
 
+            await asyncio.sleep(0)
+
         return AgentProcessHandle(
             asyncio.create_task(wait_for_process_initialized()), agent_process.pid
         )

mango/container/tcp.py

@@ -198,6 +198,8 @@ async def start(self):
             self.addr[0],
             self.addr[1],
         )
+        # if 0 is specified
+        self.addr = (self.addr[0], self.server.sockets[0]._sock.getsockname()[1])
         await super().start()
 
     async def send_message(