Add support for reducer filters - #413
Merged
Merged
Conversation
IrisvdWerf
marked this pull request as ready for review
September 8, 2026 15:10
LourensVeen
requested changes
Sep 11, 2026
LourensVeen
left a comment
Contributor
There was a problem hiding this comment.
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 😄
LourensVeen
approved these changes
Sep 14, 2026
LourensVeen
left a comment
Contributor
There was a problem hiding this comment.
Looks good! Let's get this merged 😄
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR adds support for reducer filters ("LAST").
Notes:
Changes in this PR:
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

microwhich runs in the:macro:mesotimelineWe then determine the target timelines for any O_I or O_F port with LAST conduit filters:
O_I port
O_F port
:macro:micro:macro:macro::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 theCommunicator._reduced_countdict, keyed by the destinationcomponent.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)Milestone([0,0]), we replaceMilestone([0])by the last sent message and set the iteration count of that message to[0].Milestone([])is sent as is.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 thepre_receive()logic.