Fix limiter_node decrementer

include/oneapi/tbb/flow_graph.h

@@ -1886,6 +1886,7 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
     size_t my_threshold;
     size_t my_count; // number of successful puts
     size_t my_tries; // number of active put attempts
+    size_t my_future_decrement; // number of active decrement 
     reservable_predecessor_cache< T, spin_mutex > my_predecessors;
     spin_mutex my_mutex;
     broadcast_cache< T > my_successors;
@@ -1894,9 +1895,15 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
     threshold_regulator< limiter_node<T, DecrementType>, DecrementType > decrement;
 
     graph_task* decrement_counter( long long delta ) {
+        if (delta > 0 && size_t(delta) > my_threshold ){
+            delta = my_threshold;
+        }
         {
             spin_mutex::scoped_lock lock(my_mutex);
             if( delta > 0 && size_t(delta) > my_count ) {
+                if( my_tries > 0 ){
+                    my_future_decrement += (size_t(delta) - my_count);
+                }
                 my_count = 0;
             }
             else if( delta < 0 && size_t(-delta) > my_threshold - my_count ) {
@@ -1941,6 +1948,15 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
                 {
                     spin_mutex::scoped_lock lock(my_mutex);
                     ++my_count;
+                    if(my_future_decrement){
+                        if(my_count > my_future_decrement){
+                            my_count -= my_future_decrement;
+                            my_future_decrement = 0;
+                        }else{
+                            my_future_decrement -= my_count;
+                            my_count = 0;
+                        }
+                    }
                     --my_tries;
                     my_predecessors.try_consume();
                     if ( check_conditions() ) {
@@ -1988,8 +2004,8 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
 public:
     //! Constructor
     limiter_node(graph &g, size_t threshold)
-        : graph_node(g), my_threshold(threshold), my_count(0), my_tries(0), my_predecessors(this)
-        , my_successors(this), decrement(this)
+        : graph_node(g), my_threshold(threshold), my_count(0), my_tries(0), my_future_decrement(0),
+        my_predecessors(this) , my_successors(this), decrement(this)
     {
         initialize();
     }
@@ -2071,7 +2087,6 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
         }
 
         graph_task* rtask = my_successors.try_put_task(t);
-
         if ( !rtask ) {  // try_put_task failed.
             spin_mutex::scoped_lock lock(my_mutex);
             --my_tries;
@@ -2085,8 +2100,17 @@ class limiter_node : public graph_node, public receiver< T >, public sender< T >
         else {
             spin_mutex::scoped_lock lock(my_mutex);
             ++my_count;
+            if(my_future_decrement){
+                if(my_count > my_future_decrement){
+                    my_count -= my_future_decrement;
+                    my_future_decrement = 0;
+                }else{
+                    my_future_decrement -= my_count;
+                    my_count = 0;
+                }
+            }
             --my_tries;
-             }
+        }
         return rtask;
     }
 

test/tbb/test_limiter_node.cpp

@@ -517,6 +517,35 @@ void test_deduction_guides() {
 }
 #endif
 
+void test_decrement_while_try_put_task() {
+    constexpr int threshold = 50000;
+
+    tbb::flow::graph graph{};
+    std::atomic<int> processed;
+    tbb::flow::input_node<int> input{ graph, [&](tbb::flow_control & fc) -> int {
+        static int i = {};
+        if (i++ >= threshold) fc.stop();
+        return i;
+    }};
+    tbb::flow::limiter_node<int, int> blockingNode{ graph, 1 };
+    tbb::flow::multifunction_node<int, std::tuple<int>> processing{ graph, tbb::flow::serial,
+        [&](const int & value, typename decltype(processing)::output_ports_type & out) {
+            if (value != threshold)
+                std::get<0>(out).try_put(1);
+            processed.store(value);
+        }};
+
+    tbb::flow::make_edge(input, blockingNode);
+    tbb::flow::make_edge(blockingNode, processing);
+    tbb::flow::make_edge(processing, blockingNode.decrementer());
+
+    input.activate();
+
+    graph.wait_for_all();
+    CHECK_MESSAGE(processed.load() == threshold, "decrementer terminate flow graph work");
+}
+
+
 //! Test puts on limiter_node with decrements and varying parallelism levels
 //! \brief \ref error_guessing
 TEST_CASE("Serial and parallel tests") {
@@ -537,6 +566,12 @@ TEST_CASE("Test continue_msg reception") {
     test_continue_msg_reception();
 }
 
+//! Test put message on decrementer port does not stop message flow
+//! \brief \ref error_guessing
+TEST_CASE("Test try_put to decrementer while try_put to limiter_node") {
+    test_decrement_while_try_put_task();
+}
+
 //! Test multifunction_node connected to limiter_node
 //! \brief \ref error_guessing
 TEST_CASE("Multifunction connected to limiter") {