Add support for reducer filters

integration_test/test_filters.py

@@ -115,6 +115,45 @@ def repeat_s(filters: str) -> None:
     assert micro_received == len(EXPECTED_COUNTS)
 
 
+def postprocess() -> None:
+    instance = Instance({Operator.F_INIT: ["macro", "meso", "micro"]})
+
+    while instance.reuse_instance():
+        macro = instance.receive("macro")
+        meso = instance.receive("meso")
+        micro = instance.receive("micro")
+
+        print("Received:", macro.data, meso.data, micro.data)
+        if macro.data[1] == 0:
+            # meso and micro didn't run in this iteration, so we get an empty message
+            assert meso.data is None
+            assert micro.data is None
+        else:
+            assert meso.data[-1] == macro.data[-1] - 1
+            assert micro.data[-1] == 1
+
+
+def combined(filters: str) -> None:
+    is_padded = filters.split()[-1] == "pad"
+
+    instance = Instance({Operator.F_INIT: ["trigger"], Operator.S: ["in"]})
+
+    reused = 0
+    while instance.reuse_instance():
+        instance.receive("trigger")
+
+        last_value = None if reused == 0 else ["macro", reused, "meso", reused - 1]
+        for i in range(3):
+            msg = instance.receive("in")
+            if i and is_padded:
+                assert msg.data is None
+            else:
+                assert msg.data == last_value
+        reused += 1
+
+    assert reused == 4
+
+
 config = """
 ymmsl_version: v0.2
 models:
@@ -149,19 +188,35 @@ def repeat_s(filters: str) -> None:
         ports:
           f_init: macro meso micro
         implementation: pico
+      postprocess:
+        description: Postprocessing of final actor outputs
+        ports:
+          f_init: macro meso micro
+        implementation: postprocess
+      combined:
+        description: Receives meso.out with a combined reducer and repeater filter
+        ports:
+          f_init: trigger
+          s: in
+        implementation: combined
     conduits:
       macro.out:
       - meso.in
       - {filters} pico.macro
       - {filters} repeat_s.macro
+      - postprocess.macro
+      - combined.trigger
       meso.out:
       - micro.in
       - repeat pico.meso
       - repeat_s.meso
       - repeat repeat_s.repeated_meso
+      - last postprocess.meso
+      - last {combined_filter} combined.in
       micro.out:
       - pico.micro
       - repeat_s.micro
+      - last last postprocess.micro
 """
 
 
@@ -173,8 +228,14 @@ def test_repeater_filters(tmp_path, filters):
         "micro": ("python", micro),
         "repeat_s": ("python", repeat_s, filters),
         "pico": ("python", pico, filters),
+        "postprocess": ("python", postprocess),
+        "combined": ("python", combined, filters),
     }
-    run_manager_with_actors(config.format(filters=filters), tmp_path, actors)
+    run_manager_with_actors(
+        config.format(filters=filters, combined_filter=filters.split()[-1]),
+        tmp_path,
+        actors,
+    )
 
 
 @skip_if_python_only
@@ -186,8 +247,14 @@ def test_repeater_filters_cpp(tmp_path, filters):
         "micro": ("python", micro),
         "repeat_s": ("cpp", "conduit_filters_test", "repeat_s", filters),
         "pico": ("cpp", "conduit_filters_test", "pico", filters),
+        "postprocess": ("python", postprocess),
+        "combined": ("python", combined, filters),
     }
-    run_manager_with_actors(config.format(filters=filters), tmp_path, actors)
+    run_manager_with_actors(
+        config.format(filters=filters, combined_filter=filters.split()[-1]),
+        tmp_path,
+        actors,
+    )
 
 
 checkpoint_config = """

src/cpp/libmuscle/communicator.cpp

@@ -96,8 +96,9 @@ std::vector<std::string> Communicator::get_locations() const {
 }
 
 void Communicator::set_peer_info(PeerInfo const & peer_info) {
+    timeline_ = manager_.get_timeline();
     peer_info_ = peer_info;
-    timeline_manager_ = std::make_unique<TimelineManager>(port_manager_);
+    timeline_manager_ = std::make_unique<TimelineManager>(port_manager_, timeline_.get());
     prepare_conduit_filters_();
 }
 
@@ -161,8 +162,16 @@ void Communicator::send_message(
 
         if (message.has_next_timestamp())
             mpp_message.next_timestamp = message.next_timestamp();
-
-        auto message_bytes = mpp_message.encoded();
+
+        std::vector<char> message_bytes;
+        auto peer_port = recv_endpoint.kernel + recv_endpoint.port;
+        if (outgoing_timeline_length_.count(peer_port) > 0) {
+            message_bytes = apply_reduce_filters_(peer_port, std::move(mpp_message));
+            if (message_bytes.empty())
+                continue;
+        } else {
+            message_bytes = mpp_message.encoded();
+        }
         profile_event.message_size = message_bytes.size();
         server_.deposit(recv_endpoint.ref(), std::move(message_bytes));
     }
@@ -177,6 +186,57 @@ void Communicator::send_message(
         port.set_closed(slot);
 }
 
+std::vector<char> Communicator::apply_reduce_filters_(
+        ymmsl::Reference const & peer_port, MPPMessage && message) {
+    message.message_number = -1;  // GH#411: Disabled checkpointing for reducer filter
+
+    auto reduced_count = outgoing_timeline_length_.at(peer_port);
+
+    if (!is_milestone(message.data)) {
+        // Reduce the message iteration count to match with the timeline we send to
+        message.iteration.resize(reduced_count);
+        auto it = reducer_cache_.find(message.receiver);
+        if (it != reducer_cache_.end())
+            reducer_cache_.erase(it);  // Remove existing entry
+        reducer_cache_.emplace(message.receiver, message);
+        log_debug("Message for ", message.receiver, " stored in cache");
+        return {};
+    }
+
+    // Decide whether to send the milestone, ignore it, or send a cached message.
+    std::size_t n_milestone = message.iteration.size();
+    if (n_milestone < reduced_count) {
+        // Milestone from ancestor timeline: send it
+        return message.encoded();
+    } else if (n_milestone == reduced_count) {
+        // This is the target timeline after reduce filters applied: we need to
+        // send the cached message (or make up an empty one) and discard the
+        // milestone:
+        auto it = reducer_cache_.find(message.receiver);
+        if (it == reducer_cache_.end()) {
+            log_info(
+                    "No cached message available to send because this instance did ",
+                    "not run. Sending an empty message to ", message.receiver, " instead.");
+            return MPPMessage(
+                    message.sender, message.receiver, message.port_length,
+                    message.timestamp, message.next_timestamp, message.settings_overlay,
+                    message.message_number, Data(), message.iteration
+                    ).encoded();
+        }
+
+        assert(it->second.iteration == message.iteration);        
+        log_debug("Sending cached message to ", message.receiver);
+        auto encoded = it->second.encoded();
+        reducer_cache_.erase(it);
+        return encoded;
+    } else {
+        log_debug(
+                "Ignored milestone for ", message.receiver,
+                " because of LAST filters.");
+        return {};
+    }
+}
+
 Communicator::FInitCacheType Communicator::pre_receive() {
     assert(timeline_manager_);
     auto finished_iteration = timeline_manager_->start_reuse_iteration();
@@ -376,7 +436,10 @@ MPPMessage Communicator::receive_message_(
     }
 
     int expected_message_number = port.get_num_messages(slot);
-    if (expected_message_number != mpp_message.message_number) {
+    if (
+        mpp_message.message_number >= 0  // GH#411: negative for reducer filters
+        && expected_message_number != mpp_message.message_number
+    ) {
         if (expected_message_number - 1 == mpp_message.message_number and
                 port.is_resuming(slot)) {
             log_debug("Discarding received message on ", port_and_slot,
@@ -560,7 +623,21 @@ void Communicator::prepare_conduit_filters_() {
         }
     }
 
-    // TODO: reducer filters
+    // Reducer filters
+    for (auto op : {Operator::O_I, Operator::O_F}) {
+        for (Port const & port : port_manager_.get_connected_ports(op, {})) {
+            for (auto & peer_port : peer_info_.get().get_peer_ports(port.name)) {
+                auto & filters = peer_info_.get().get_filters_for_receiver(peer_port);
+                // Count the reducer filters, receiving component handles repeaters
+                auto n_reducers = std::count_if(
+                        filters.begin(), filters.end(), ::ymmsl::is_reducer);
+                if (n_reducers > 0) {
+                    std::size_t reduced_count = timeline_.get().size() + port.timeline.size() - n_reducers;
+                    outgoing_timeline_length_.emplace(peer_port, reduced_count);
+                }
+            }
+        }
+    }
 }
 
 

src/cpp/libmuscle/communicator.hpp

@@ -170,6 +170,19 @@ class Communicator {
                 Optional<int> slot = {}
                 );
 
+        /** Apply reduce filters to a message sent on a conduit with reduce filters.
+         * 
+         * User-provided messages (through instance.send()) will be stored (overwriting any
+         * existing message). For milestones this method decides if the milestone should be
+         * sent, or a cached message, or nothing at all.
+         * 
+         * @param peer_port Peer port (component + port) to send to.
+         * @param message MPPMessage to be checked.
+         * @return The encoded MPPMessage to send, or an empty vector if we do not need to send anything.
+         */
+        std::vector<char> apply_reduce_filters_(
+                ymmsl::Reference const & peer_port, MPPMessage && message);
+
         ymmsl::Reference instance_id_() const;
         MPPClient & get_client_(ymmsl::Reference const & instance);
 
@@ -242,10 +255,26 @@ class Communicator {
         Optional<PeerInfo> peer_info_;
         double receive_timeout_;
         std::unique_ptr<TimelineManager> timeline_manager_;
+        Optional<ymmsl::Timeline> timeline_;
 
         PortManager::PortReferences pre_receive_ports_;
         std::unordered_map<std::string, std::vector<::ymmsl::ConduitFilter>> repeat_filters_;
         MPPCacheType message_cache_;
+
+        /** Size of IterationCount, after applying the reducer filters, per peer port.
+         *
+         * Keys are references to peer ports: ``component + port``. The outgoing
+         * timeline length is the size of the IterationCount after applying the
+         * reducer filters and determines in which (parent) timeline these messages
+         * are sent.
+         * If our timeline is ":macro:micro" then:
+         * - outgoing_timeline_length = 0: send on the root (":") timeline
+         * - outgoing_timeline_length = 1: send on the ":macro" timeline
+         * - outgoing_timeline_length = 2: send on the ":macro:micro" timeline
+         */
+        std::unordered_map<::ymmsl::Reference, std::size_t> outgoing_timeline_length_;
+        /** Message cache for reducer filters */
+        MPPCacheType reducer_cache_;
 };
 
 } }

src/cpp/libmuscle/tests/mocks/mock_timeline_manager.hpp

@@ -31,7 +31,8 @@ class MockTimelineManager : public MockClass<MockTimelineManager> {
             init_from_return_value();
         }
 
-        explicit MockTimelineManager(PortManager const & port_manager) {
+        explicit MockTimelineManager(
+                PortManager const & port_manager, ::ymmsl::Timeline const & timeline) {
             init_from_return_value();
             constructor(&port_manager);
         }