kiwipy/rmq related modules into independent module by unkcpz · Pull Request #297 · aiidateam/plumpy

unkcpz · 2024-12-12T10:58:21Z

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).

codecov · 2024-12-14T13:40:44Z

Codecov Report

Attention: Patch coverage is 78.23640% with 116 lines in your changes missing coverage. Please review.

Please upload report for BASE (dev@f760b4a). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
src/plumpy/rmq/futures.py	46.99%	44 Missing ⚠️
src/plumpy/processes.py	71.92%	25 Missing ⚠️
src/plumpy/rmq/process_control.py	82.53%	18 Missing ⚠️
src/plumpy/controller.py	69.57%	7 Missing ⚠️
src/plumpy/message.py	94.02%	7 Missing ⚠️
src/plumpy/coordinator.py	73.69%	5 Missing ⚠️
src/plumpy/futures.py	70.59%	5 Missing ⚠️
src/plumpy/rmq/coordinator.py	91.12%	4 Missing ⚠️
src/plumpy/rmq/communications.py	92.86%	1 Missing ⚠️

Additional details and impacted files

@@          Coverage Diff           @@
##             dev     #297   +/-   ##
======================================
  Coverage       ?   89.46%           
======================================
  Files          ?       28           
  Lines          ?     3196           
  Branches       ?        0           
======================================
  Hits           ?     2859           
  Misses         ?      337           
  Partials       ?        0

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

tests/utils.py

…hange in kiwipy

unkcpz · 2025-02-24T09:30:25Z

Rebase with uv.lock, the lines changed are much less. It is ready for another review @agoscinski

unkcpz · 2025-02-24T09:32:17Z

tests/rmq/test_process_control.py

    )
-    communicator._loop.set_debug(True)
+    comm._loop.set_debug(True)
+    coordinator = RmqCoordinator(comm)


This is a typical example how I wrap the original rmq communicator into the Coordinator interface.

unkcpz · 2025-02-24T09:33:29Z

src/plumpy/coordinator.py

+    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: ...


I rename the interfaces to distinguish the coordinator from rmq communicator.

agoscinski

Need to pause review. Posting what I have so far now.

src/plumpy/broadcast_filter.py

agoscinski · 2025-02-26T13:28:47Z

src/plumpy/rmq/process_control.py

+
+# This class not conform with typing of ProcessController protocol.
+# Does't matter too much, since this controller is not directly used as the controller by downstream.
+class RemoteProcessController:


I would inherit from the protocol to make it explicit that we implement it

class RemoteProcessController(ProcessController):

It defeats part of the idea Protocol that you do not need inheritance, but from this file it is not clear that this is implemented. Still you get the advantage (or disadvantage, depending how one sees it) of moving the check that the class is correctly implemented to static type checking phase and not runtime is for abstract classes. (EDIT: The same is True for abstract classes I really don't get the benifit)

Further reading: https://peps.python.org/pep-0544/#explicitly-declaring-implementation

Agree, I inherit for both, these two controllers do not need to be generic. Make a lot sense to just explicitly declaring the implementation.

agoscinski · 2025-02-26T13:37:29Z

tests/utils.py

 Snapshot = collections.namedtuple('Snapshot', ['state', 'bundle', 'outputs'])


+class MockCoordinator:


I think we should have a test that this actually implements the protocol. By inheriting from the Protocol the type checker would pick it up but I am not sure if the type checker is running on the tests.

Just give it a try. For MockCoordinator, sure, didn't see the power of protocol. When it comes to RmqCoordinator where I want to have a generic typing support for the inner communicator (in aiida, it is LoopCommunicator, in plumpy, there are RmqCommunicator and RmqThreadCommunicator).
I am sure there should be ways to workaround this:

tests/rmq/__init__.py:19: in <module> class RmqCoordinator(Generic[U], Coordinator): ../../.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/abc.py:106: in __new__ cls = super().__new__(mcls, name, bases, namespace, **kwargs) E TypeError: Cannot create a consistent method resolution E order (MRO) for bases Generic, Coordinator

I believe this can make our life a bit uneasy when it comes to define more generic types that need to work with static type checking.

It is a valid point that the test is better to add, so I add runtime_checkable to Coordinator and have two tests for MockCoordinator and RmqCoordinator respectively. (4242057)

Okay, that was my skill issue. What I should do is:

class RmqCoordinator(Coordinator, Generic[U]): ...

Was the inheritance order important or what was the issue something else?

The problem was Coordinator as a protocol is already inherited from Generic. Looks it again, I still don't think it is a good way to put two parent class here.

A note for my self.
In the current situation, there is no actually benefit to use protocol over abc class. I can just argue it is a bit more "flexible", with the cost of losing the explicity. The advantage only shows up when there is another implementation that has its own way to use its own "Coordinator" then making RmqCoordinator to work with that case without inherit can be only possible with protocol. But we never need to get there.

Therefore, the style will be using protocol but explicitly inherit from it for subclass.

agoscinski · 2025-02-26T15:28:16Z

src/plumpy/rmq/futures.py

+    return new_future
+
+
+# XXX: this required in aiida-core, see if really need this unwrap.


Did you change the logic somewhere that it does not need to be unwrapped? I am actually not sure anymore why we have these nested Futures.

The only place that uses this unwrap is in aiida_core/src/aiida/engine/processes/control.py. This piece of shitty workaround was actually add by me :( aiidateam/aiida-core@cd0d15c what a shame.

I said the workaround "should and can be removed". I'll give it a look on the aiida-core side after this PR.

agoscinski

Added comments for future changes in the dev branch to PR #319

subscriber -> receiver to distinguish interfaces from RMQ communicator.

Explicitly declare protocol implementations

unkcpz · 2025-02-27T10:52:46Z

It is rebased and merged to dev branch. Many many thanks for the review @agoscinski!

unkcpz marked this pull request as draft December 12, 2024 10:58

unkcpz mentioned this pull request Dec 12, 2024

Defer import of aio_pika #295

Closed

unkcpz force-pushed the rmq-out branch 5 times, most recently from fc52fcd to c0a8bbd Compare December 14, 2024 23:47

unkcpz changed the base branch from dev to master December 15, 2024 00:22

unkcpz changed the base branch from master to dev December 15, 2024 00:22

unkcpz force-pushed the rmq-out branch 9 times, most recently from 4960836 to 47ac8e4 Compare December 17, 2024 20:28

unkcpz changed the title ~~Move kiwipy/rmq related modules into a specific module~~ kiwipy/rmq related modules into independent module Dec 17, 2024

unkcpz force-pushed the rmq-out branch 7 times, most recently from 85fc72a to da644ac Compare December 17, 2024 21:53

unkcpz commented Dec 17, 2024

View reviewed changes

tests/utils.py Outdated Show resolved Hide resolved

unkcpz mentioned this pull request Dec 17, 2024

Defer the import of plumpy exception thus aiopika aiidateam/aiida-core#6661

Closed

unkcpz marked this pull request as ready for review December 17, 2024 22:52

unkcpz added 7 commits February 24, 2025 10:27

Remove kiwipy/rmq dependencies of process module

2d9fdb9

Interface change from communicator -> coordinator

9d6655e

Remove unnecessary task_send ab from RemoteProcessControl interface

f8cc8ed

Interface for ProcessController

34d3842

RmqCoordinator example to show how using interface can avoid making c…

bf99d23

…hange in kiwipy

broadcast subscriber has versatile filters

15b267b

Generic typing for Coordinator

23d954c

unkcpz force-pushed the rmq-out branch 2 times, most recently from 49991fe to f698369 Compare February 24, 2025 09:29

unkcpz commented Feb 24, 2025

View reviewed changes

agoscinski requested changes Feb 26, 2025

View reviewed changes

unkcpz force-pushed the rmq-out branch from 4242057 to fc3dc88 Compare February 26, 2025 21:14

agoscinski mentioned this pull request Feb 27, 2025

dev -> main #319

Closed

agoscinski approved these changes Feb 27, 2025

View reviewed changes

unkcpz added 7 commits February 27, 2025 11:48

Adopt new message protocol and changes required for aiida-core support

46bd2d3

Simpler create_task_threadsafe implementation

8e23a88

Remove RmqCoordinator to tests/util only

f3f3095

Export plumpy.futures.Future

b577f5e

Remove first unnecessary _comm argument to subscriber

b0877db

Protocol fulfill for RemoteProcessThreadController

4b0267c

Rename interfaces of Coordinator

c4c50e0

subscriber -> receiver to distinguish interfaces from RMQ communicator.

unkcpz force-pushed the rmq-out branch from fc3dc88 to 14c6348 Compare February 27, 2025 10:49

unkcpz added 2 commits February 27, 2025 11:49

ralex - T -> CommT in generic type at communications.py

b1e4e1e

misc: pre-commit / uv.lock

14c6348

Explicitly declare protocol implementations

unkcpz merged commit a4b8752 into aiidateam:dev Feb 27, 2025
7 checks passed

unkcpz deleted the rmq-out branch February 27, 2025 10:53

		Snapshot = collections.namedtuple('Snapshot', ['state', 'bundle', 'outputs'])


		class MockCoordinator:

		return new_future


		# XXX: this required in aiida-core, see if really need this unwrap.

Conversation

unkcpz commented Dec 12, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov bot commented Dec 14, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

unkcpz commented Feb 24, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agoscinski left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

agoscinski Feb 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agoscinski Feb 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agoscinski left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

unkcpz commented Feb 27, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

unkcpz commented Dec 12, 2024 •

edited

Loading

codecov bot commented Dec 14, 2024 •

edited

Loading

agoscinski Feb 26, 2025 •

edited

Loading

agoscinski Feb 26, 2025 •

edited

Loading

agoscinski left a comment •

edited

Loading