Fix for overwrite_node to continue_node accounting (#1729)

Co-authored-by: Konstantin Boyarinov <[email protected]>

Author: vossmjp

include/oneapi/tbb/flow_graph.h

@@ -329,6 +329,7 @@ class receiver {
     virtual graph& graph_reference() const = 0;
 
     template<typename TT, typename M> friend class successor_cache;
+    template< typename TTT > friend class overwrite_node;
     virtual bool is_continue_receiver() { return false; }
 
     // TODO revamp: reconsider the inheritance and move node priority out of receiver
@@ -3318,6 +3319,12 @@ class overwrite_node : public graph_node, public receiver<T>, public sender<T> {
         spin_mutex::scoped_lock l( my_mutex );
         if (my_buffer_is_valid && is_graph_active( my_graph )) {
             // We have a valid value that must be forwarded immediately.
+            if (s.is_continue_receiver()) {
+                // try_put can never fail, since continue_receivers always accept
+                my_successors.register_successor( s );
+                s.try_put( my_buffer );
+                return true;
+            }
             bool ret = s.try_put( my_buffer );
             if ( ret ) {
                 // We add the successor that accepted our put

test/common/graph_utils.h

@@ -1,5 +1,5 @@
 /*
-    Copyright (c) 2005-2024 Intel Corporation
+    Copyright (c) 2005-2025 Intel Corporation
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
@@ -696,6 +696,47 @@ void test_reserving_nodes() {
     CHECK(end_receiver.my_count == 2 * N);
 }
 
+template<typename BufferNode>
+void test_nested_make_edge_single_item_buffer_to_continue_receiver() {
+    tbb::flow::graph g;
+
+    using msg_t = tbb::flow::continue_msg;
+    using cnode_t = tbb::flow::continue_node<msg_t>;
+
+    std::atomic<int> count(0);
+
+    // make a single item buffer and fill it
+    BufferNode b{g};
+    b.try_put(msg_t{});
+
+    cnode_t execute_one_time{g,
+        [&](const msg_t& m) {
+            ++count;
+            return m;
+        }};
+
+    cnode_t edge_adder{g, 
+        [&](const msg_t& m) {
+            // should increment predecessor count on execute_one_time
+            // should NOT cause execute_one_time to immediately execute
+            // since it has 2 predecessors, edge_adder and b
+            tbb::flow::make_edge(b, execute_one_time);
+            return m;
+        }};
+
+    tbb::flow::make_edge(edge_adder, execute_one_time);
+
+    // execute_one should execute
+    edge_adder.try_put(msg_t{});
+    g.wait_for_all();
+
+    // execute_one should NOT execute, it has 2 predecessors and
+    // has seen a total of 3 messages
+    execute_one_time.try_put(msg_t{});
+    g.wait_for_all();
+    CHECK_MESSAGE ((count == 1), "node should only execute once");
+}
+
 namespace lightweight_testing {
 
 typedef std::tuple<int, int> output_tuple_type;

test/tbb/test_overwrite_node.cpp

@@ -1,5 +1,5 @@
 /*
-    Copyright (c) 2005-2024 Intel Corporation
+    Copyright (c) 2005-2025 Intel Corporation
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
@@ -423,3 +423,13 @@ TEST_CASE("test overwrite_node try_put_and_wait") {
     test_overwrite_node_try_put_and_wait();
 }
 #endif
+
+//! Test for nested make_edge from predecessor of continue_node that adds edge to overwrite_node
+//! \brief \ref error_guessing
+TEST_CASE("Nested make_edge to continue_node") {
+    tbb::flow::graph g;
+
+    using msg_t = tbb::flow::continue_msg;
+    using buffer_t = tbb::flow::overwrite_node<msg_t>;
+    test_nested_make_edge_single_item_buffer_to_continue_receiver<buffer_t>();
+}

test/tbb/test_write_once_node.cpp

@@ -1,5 +1,5 @@
 /*
-    Copyright (c) 2005-2024 Intel Corporation
+    Copyright (c) 2005-2025 Intel Corporation
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
@@ -380,3 +380,11 @@ TEST_CASE("test write_once_node try_put_and_wait") {
     test_try_put_and_wait();
 }
 #endif
+
+//! Test for nested make_edge from predecessor of continue_node that adds edge to write_once_node
+//! \brief \ref error_guessing
+TEST_CASE("Nested make_edge write_once_node to continue_node") {
+    using msg_t = tbb::flow::continue_msg;
+    using buffer_t = tbb::flow::write_once_node<msg_t>;
+    test_nested_make_edge_single_item_buffer_to_continue_receiver<buffer_t>();
+}