Skip to content

Add support for reducer filters - #413

Merged
IrisvdWerf merged 12 commits into
developfrom
feature/reducer-filters
Sep 14, 2026
Merged

IrisvdWerf merged 12 commits into
developfrom
feature/reducer-filters

Conversation

@maarten-ic

@maarten-ic maarten-ic commented Sep 3, 2026 •

Copy link
Copy Markdown
Collaborator

This PR adds support for reducer filters ("LAST").

Notes:


Changes in this PR:

  • Support LAST conduit filters, see below for the algorithm.
  • Set MMPMessage.message_count negative and disable corresponding consistency checks when LAST filters are applied. This prevents checkpointing (see Checkpointing support when using a "last" conduit filter #411) and muscle_manager will raise an exception when trying to use checkpoints in a simulation with reducer conduit filters.
  • Add a check in the TimelineManager that the iteration counts of F_INIT messages match the timeline of a component.

LAST conduit filters algorithm:

  • In Communicator.prepare_conduit_filters: determine the target timeline of messages sent to a peer port with last filters applied. As an example: take component micro which runs in the :macro:meso timeline
    image

    We then determine the target timelines for any O_I or O_F port with LAST conduit filters:

    # LAST filters Target timeline
    O_I port

    O_F port
    1 :macro:micro :macro
    2 :macro :
    3 : N/A

    The actual name of the timeline isn't important for the remainder of the algorithm, so we just store the "depth" of the timeline (len(timeline) in Python). This corresponds to the number of elements in the IterationCount for messages on that timeline. This number is stored in the Communicator._reduced_count dict, keyed by the destination component.port. I'm not too happy with the name of this variable, so feel free to update if you have a better idea!

    N.B. Since one outgoing port can be connected (through multicasting) with many Conduits (each potentially having different amounts of LAST filters), these "reduced_counts" are keyed by peer port (<component>.<port>).

  • When sending a message (Communicator.send_message) to a peer port with LAST filters applied, we will store the message in the reducer_cache (keyed by Endpoint reference to include both the peer portname and the slot)

    • When handling milestones we check if we need to send those cached messages instead of the milestone. When multiple reducer filters are applied we may need to discard the milestone altogether. For example, if we have two LAST filters applied to the O_I port from our example above, then we discard Milestone([0,0]), we replace Milestone([0]) by the last sent message and set the iteration count of that message to [0]. Milestone([]) is sent as is.
    • This handling of milestone also happens in Communicator.send_message (Communicator.apply_reduce_filters() to be precise), which was much easier to implement than trying to catch it before sending the milestones in the pre_receive() logic.

@IrisvdWerf
IrisvdWerf marked this pull request as ready for review September 8, 2026 15:10

@LourensVeen LourensVeen left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks good! I have some requests, but this is close to ready.

And looking at that first comment again, _outgoing_timeline_length would have my preference, but feel free to suggest something even better 😄

Comment thread src/python/libmuscle/communicator.py Outdated
Comment thread src/python/libmuscle/test/test_communicator_conduit_filters.py
Comment thread integration_test/test_filters.py

@LourensVeen LourensVeen left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks good! Let's get this merged 😄

@IrisvdWerf
IrisvdWerf merged commit 27e3bb0 into develop Sep 14, 2026
15 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants