Decouple broker and coordinator interface

[unkcpz]

• all FIXME notes resolved
• MockCoordinator -> InMemoryCoordinator and an exmaple of submit without rmq.
• runner with dedicated thread (need more experiments)
• Which change in plumpy/rmq-out resolve the test_disconnect timeout?
• BroadcastFilter absorbed into add_broadcast_subscriber API.
• broker and coordinator are duplicate abstraction, remove broker this should be done after, in this change, it bring coordinator layer first to not break the broker abstraction introduced (which is not real abstraction and relatively useless).
• Move all kiwipy/rmq modules into broker/rabbitmq and make aiida-core not directly (but through broker/coordinator interface) dep on it.

[agoscinski]

First round of comments.

Note for me

Runner -> Broker -> Communicator
                 -> Coordinator -> Communicator
                 -> ProcessController

[agoscinski]

We probably should expose types in plumpy to external libraries (see code in plumpy) to use them for type hints here

[agoscinski]

Note for me: This has been moved from src/aiida/manage/manager.py

[agoscinski]

Note for me: keyword arguments have been made explicit

[agoscinski]

why this are marked for internal use only? Do you want to make them part of the class? Right now they are just defined in the function

[agoscinski]

Note for me: this is making the keyword arguments explicit. I think the logic is more or less the same

[agoscinski]

I played around and this seems to work

# copy of runner.close adapted
assert not runner._closed
runner._loop.call_soon_threadsafe(runner._loop.stop)
thread.join()
if not runner._loop.is_running():
    runner._loop.close()
reset_event_loop_policy()
runner._closed = True

at least for this toy example

import asyncio
import threading
import time

def start_event_loop(loop):
    """Run an asyncio event loop in a separate thread."""
    asyncio.set_event_loop(loop)
    loop.run_forever()

# Create a new asyncio event loop
loop = asyncio.new_event_loop()

# Start the loop in a new thread
thread = threading.Thread(target=start_event_loop, args=(loop,), daemon=True)
thread.start()

# Function to run async tasks from the main thread
def run_async_task(coro):
    """Submit a coroutine to the event loop running in another thread."""
    return asyncio.run_coroutine_threadsafe(coro, loop)

# Example async function
async def say_hello():
    print(f"Hello from thread: {threading.current_thread().name}")
    await asyncio.sleep(2)
    print(f"Goodbye from thread: {threading.current_thread().name}")

# Submit coroutine to the running event loop
future = run_async_task(say_hello())

# Wait for the result
future.result()

# Give time to process (optional)
time.sleep(1)

# Stop the event loop and join the thread (if needed)
loop.call_soon_threadsafe(loop.stop)
thread.join()
loop.close()

If I don't use call_soon_threadsafe it hangs. Also I cannot close the loop before stopping it.

If this works there still remains the question how to integrate this into the runner as the runner needs to be aware about the thread it is running on. One cannot close the loop before joining the thread

[agoscinski]

Note for me: the runner does not know the communicator/coordinatior anymore but the broker. The broker knows the coordinator.

[agoscinski]

I actually like is None checks in general more since they are explicit. This also turns returns also to the default case if wrong types are put in loop = [] or [5] # -> loop will be [5]. I don't think it matters in this case but I wanted to mention this.

[agoscinski]

Can we give it a more meaningful name?

Suggested change

	U = TypeVar('U', bound=kiwipy.Communicator)
	CommunicatorImpl = TypeVar('CommunicatorImpl', bound=kiwipy.Communicator)

[agoscinski]

Right now we keep two communicator

Broker -> Communicator
       -> Coordinator -> Communicator

It could happen easily in the code that two communicators get out of sync (so reference to a different one). If we expose the Communicator in the Coordinator we could simplify it to

Broker -> Coordinator -> Communicator

and avoid this issue of two communicators. I am not sure how easy it is to replace the communicator in the coordinator, since no one is managing the old communicator anymore. So we would need to close the existing coordinator and raise an error if it cannot be closed.