Merge pull request #297 from unkcpz/rmq-out

Refactoring and  move kiwipy/rmq related modules as independent module.

The refactoring is targeting to decouple the dependencies of using kiwipy+rmq as the communicator for the process control. 
By forming a `Coordinator` protocol contract, the different type of rmq/kiwipy related codes are removed out from plumpy logic. The new contract also pave the way to make it clearly show how a new type coordinator can be implemented (future examples will be the `tatzelwurm` a task broker that has scheduler support and file based task broker require no background service). 
For the prototype of how a coordinator should look like, the `MockCoordinator` in `tests/utils` is the coordinator that store things in memory, and can serve as the lightweight ephemeral daemon without persistent functionality.

Another major change here is hand write the resolver of future by mimic how tho asyncio does for wrapping `concurrent.futures.Future` into `asyncio.Future`. I use the same way to convert `asyncio.Future` into `concurent.futures.Future` (which is the `kiwipy.Future` as alias).

docs/source/concepts.md

@@ -32,7 +32,7 @@ WorkChains support the use of logical constructs such as `If_` and `While_` to c
 
 A `Controller` can control processes throughout their lifetime, by sending and receiving messages. It can launch, pause, continue, kill and check status of the process.
 
-The {py:class}`~plumpy.process_comms.RemoteProcessThreadController` can communicate with the process over the thread communicator provided by {{kiwipy}} which can subscribe and send messages over the {{rabbitmq}} message broker.
+The {py:class}`~plumpy.rmq.process_control.RemoteProcessThreadController` can communicate with the process over the thread communicator provided by {{kiwipy}} which can subscribe and send messages over the {{rabbitmq}} message broker.
 
 The thread communicator runs on a independent thread (event loop) and so will not be blocked by sometimes long waiting times in the process event loop.
 Using RabbitMQ means that  even if the computer is terminated unexpectedly, messages are persisted and can be run once the computer restarts.

docs/source/nitpick-exceptions

@@ -23,7 +23,7 @@ py:class plumpy.base.state_machine.State
 py:class State
 py:class Process
 py:class plumpy.futures.CancellableAction
-py:class plumpy.communications.LoopCommunicator
+py:class plumpy.rmq.communications.LoopCommunicator
 py:class plumpy.persistence.PersistedPickle
 py:class plumpy.utils.AttributesFrozendict
 py:class plumpy.workchains._FunctionCall

docs/source/tutorial.ipynb

@@ -66,7 +66,7 @@
     "The {py:class}`~plumpy.workchains.WorkChain`\n",
     ": A subclass of `Process` that allows for running a process as a set of discrete steps (also known as instructions), with the ability to save the state of the process after each instruction has completed.\n",
     "\n",
-    "The process `Controller` (principally the {py:class}`~plumpy.process_comms.RemoteProcessThreadController`)\n",
+    "The process `Controller` (principally the {py:class}`~plumpy.rmq.process_control.RemoteProcessThreadController`)\n",
     ": To control the process or workchain throughout its lifetime."
    ]
   },

src/plumpy/__init__.py

@@ -4,18 +4,21 @@
 
 import logging
 
-from .communications import *
+# interfaces
+from .controller import ProcessController
+from .coordinator import Coordinator
 from .events import *
 from .exceptions import *
 from .futures import *
 from .loaders import *
+from .message import *
 from .mixins import *
 from .persistence import *
 from .ports import *
-from .process_comms import *
 from .process_listener import *
 from .process_states import *
 from .processes import *
+from .rmq import *
 from .utils import *
 from .workchains import *
 
@@ -27,14 +30,13 @@
     + futures.__all__
     + mixins.__all__
     + persistence.__all__
-    + communications.__all__
-    + process_comms.__all__
+    + message.__all__
     + process_listener.__all__
     + workchains.__all__
     + loaders.__all__
     + ports.__all__
     + process_states.__all__
-)
+) + ['ProcessController', 'Coordinator']
 
 
 # Do this se we don't get the "No handlers could be found..." warnings that will be produced

src/plumpy/broadcast_filter.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+# type: ignore
+import re
+import typing
+
+
+class BroadcastFilter:
+    """A filter that can be used to limit the subjects and/or senders that will be received"""
+
+    def __init__(self, subscriber, subject=None, sender=None):
+        self._subscriber = subscriber
+        self._subject_filters = []
+        self._sender_filters = []
+        if subject is not None:
+            self.add_subject_filter(subject)
+        if sender is not None:
+            self.add_sender_filter(sender)
+
+    @property
+    def __name__(self):
+        return 'BroadcastFilter'
+
+    def __call__(self, body, sender=None, subject=None, correlation_id=None):
+        if self.is_filtered(sender, subject):
+            return None
+        return self._subscriber(body, sender, subject, correlation_id)
+
+    def is_filtered(self, sender, subject) -> bool:
+        if subject is not None and self._subject_filters and not any(check(subject) for check in self._subject_filters):
+            return True
+
+        if sender is not None and self._sender_filters and not any(check(sender) for check in self._sender_filters):
+            return True
+
+        return False
+
+    def add_subject_filter(self, subject_filter):
+        self._subject_filters.append(self._ensure_filter(subject_filter))
+
+    def add_sender_filter(self, sender_filter):
+        self._sender_filters.append(self._ensure_filter(sender_filter))
+
+    @classmethod
+    def _ensure_filter(cls, filter_value):
+        if isinstance(filter_value, str):
+            return re.compile(filter_value.replace('.', '[.]').replace('*', '.*')).match
+        if isinstance(filter_value, typing.Pattern):  # pylint: disable=isinstance-second-argument-not-valid-type
+            return filter_value.match
+
+        return lambda val: val == filter_value
+
+    @classmethod
+    def _make_regex(cls, filter_str):
+        """
+        :param filter_str: The filter string
+        :type filter_str: str
+        :return: The regular expression object
+        """
+        return re.compile(filter_str.replace('.', '[.]'))

src/plumpy/controller.py

@@ -0,0 +1,136 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any, Hashable, Optional, Protocol, Union, runtime_checkable
+
+from plumpy import loaders
+from plumpy.message import MessageType
+from plumpy.utils import PID_TYPE
+
+ProcessResult = Any
+ProcessStatus = Any
+
+
+@runtime_checkable
+class ProcessController(Protocol):
+    """
+    Control processes using coroutines that will send messages and wait
+    (in a non-blocking way) for their response
+    """
+
+    def get_status(self, pid: 'PID_TYPE') -> ProcessStatus:
+        """
+        Get the status of a process with the given PID
+        :param pid: the process id
+        :return: the status response from the process
+        """
+        ...
+
+    def pause_process(self, pid: 'PID_TYPE', msg_text: str | None = None) -> Any:
+        """
+        Pause the process
+
+        :param pid: the pid of the process to pause
+        :param msg: optional pause message
+        :return: True if paused, False otherwise
+        """
+        ...
+
+    def pause_all(self, msg_text: str | None) -> None:
+        """Pause all processes that are subscribed to the same coordinator
+
+        :param msg_text: an optional pause message text
+        """
+        ...
+
+    def play_process(self, pid: 'PID_TYPE') -> ProcessResult:
+        """Play the process
+
+        :param pid: the pid of the process to play
+        :return: True if played, False otherwise
+        """
+        ...
+
+    def play_all(self) -> None:
+        """Play all processes that are subscribed to the same coordinator"""
+
+    def kill_process(self, pid: 'PID_TYPE', msg_text: str | None = None) -> Any:
+        """Kill the process
+
+        :param pid: the pid of the process to kill
+        :param msg: optional kill message
+        :return: True if killed, False otherwise
+        """
+        ...
+
+    def kill_all(self, msg_text: Optional[str]) -> None:
+        """Kill all processes that are subscribed to the same coordinator
+
+        :param msg: an optional pause message
+        """
+        ...
+
+    def notify_msg(self, msg: MessageType, sender: Hashable | None = None, subject: str | None = None) -> None:
+        """
+        Notify all processes by broadcasting of a msg
+
+        :param msg: an optional pause message
+        """
+
+    def continue_process(
+        self, pid: 'PID_TYPE', tag: Optional[str] = None, nowait: bool = False, no_reply: bool = False
+    ) -> Union[None, PID_TYPE, ProcessResult]:
+        """Continue the process
+
+        :param _communicator: the communicator
+        :param pid: the pid of the process to continue
+        :param tag: the checkpoint tag to continue from
+        """
+        ...
+
+    def launch_process(
+        self,
+        process_class: str,
+        init_args: Optional[Sequence[Any]] = None,
+        init_kwargs: Optional[dict[str, Any]] = None,
+        persist: bool = False,
+        loader: Optional[loaders.ObjectLoader] = None,
+        nowait: bool = False,
+        no_reply: bool = False,
+    ) -> Union[None, PID_TYPE, ProcessResult]:
+        """Launch a process given the class and constructor arguments
+
+        :param process_class: the class of the process to launch
+        :param init_args: the constructor positional arguments
+        :param init_kwargs: the constructor keyword arguments
+        :param persist: should the process be persisted
+        :param loader: the classloader to use
+        :param nowait: if True, don't wait for the process to send a response, just return the pid
+        :param no_reply: if True, this call will be fire-and-forget, i.e. no return value
+        :return: the result of launching the process
+        """
+        ...
+
+    def execute_process(
+        self,
+        process_class: str,
+        init_args: Optional[Sequence[Any]] = None,
+        init_kwargs: Optional[dict[str, Any]] = None,
+        loader: Optional[loaders.ObjectLoader] = None,
+        nowait: bool = False,
+        no_reply: bool = False,
+    ) -> Union[None, PID_TYPE, ProcessResult]:
+        """Execute a process.  This call will first send a create task and then a continue task over
+        the communicator.  This means that if communicator messages are durable then the process
+        will run until the end even if this interpreter instance ceases to exist.
+
+        :param process_class: the process class to execute
+        :param init_args: the positional arguments to the class constructor
+        :param init_kwargs: the keyword arguments to the class constructor
+        :param loader: the class loader to use
+        :param nowait: if True, don't wait for the process to send a response
+        :param no_reply: if True, this call will be fire-and-forget, i.e. no return value
+        :return: the result of executing the process
+        """
+        ...

src/plumpy/coordinator.py

@@ -0,0 +1,56 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+from re import Pattern
+from typing import TYPE_CHECKING, Any, Callable, Hashable, Protocol, runtime_checkable
+
+if TYPE_CHECKING:
+    ID_TYPE = Hashable
+    Receiver = Callable[..., Any]
+
+
+@runtime_checkable
+class Coordinator(Protocol):
+    def hook_rpc_receiver(
+        self,
+        receiver: 'Receiver',
+        identifier: 'ID_TYPE | None' = None,
+    ) -> Any: ...
+
+    def hook_broadcast_receiver(
+        self,
+        receiver: 'Receiver',
+        subject_filters: list[Hashable | Pattern[str]] | None = None,
+        sender_filters: list[Hashable | Pattern[str]] | None = None,
+        identifier: 'ID_TYPE | None' = None,
+    ) -> Any: ...
+
+    def hook_task_receiver(
+        self,
+        receiver: 'Receiver',
+        identifier: 'ID_TYPE | None' = None,
+    ) -> 'ID_TYPE': ...
+
+    def unhook_rpc_receiver(self, identifier: 'ID_TYPE | None') -> None: ...
+
+    def unhook_broadcast_receiver(self, identifier: 'ID_TYPE | None') -> None: ...
+
+    def unhook_task_receiver(self, identifier: 'ID_TYPE') -> None: ...
+
+    def rpc_send(
+        self,
+        recipient_id: Hashable,
+        msg: Any,
+    ) -> Any: ...
+
+    def broadcast_send(
+        self,
+        body: Any | None,
+        sender: 'ID_TYPE | None' = None,
+        subject: str | None = None,
+        correlation_id: 'ID_TYPE | None' = None,
+    ) -> Any: ...
+
+    def task_send(self, task: Any, no_reply: bool = False) -> Any: ...
+
+    def close(self) -> None: ...

src/plumpy/exceptions.py

@@ -1,16 +1,23 @@
 # -*- coding: utf-8 -*-
 from typing import Optional
 
-__all__ = ['ClosedError', 'InvalidStateError', 'KilledError', 'PersistenceError', 'UnsuccessfulResult']
+__all__ = [
+    'ClosedError',
+    'CoordinatorConnectionError',
+    'CoordinatorTimeoutError',
+    'InvalidStateError',
+    'KilledError',
+    'PersistenceError',
+    'UnsuccessfulResult',
+]
 
 
 class KilledError(Exception):
     """The process was killed."""
 
 
 class InvalidStateError(Exception):
-    """
-    Raised when an operation is attempted that requires the process to be in a state
+    """Raised when an operation is attempted that requires the process to be in a state
     that is different from the current state
     """
 
@@ -33,3 +40,19 @@ class PersistenceError(Exception):
 
 class ClosedError(Exception):
     """Raised when an mutable operation is attempted on a closed process"""
+
+
+class TaskRejectedError(Exception):
+    """A task was rejected by the coordinacor"""
+
+
+class CoordinatorCommunicationError(Exception):
+    """Generic coordinator communication error"""
+
+
+class CoordinatorConnectionError(ConnectionError):
+    """Raised when coordinator cannot be connected"""
+
+
+class CoordinatorTimeoutError(TimeoutError):
+    """Raised when communicate with coordinator timeout"""