Fix resource_limited_node exception safety (#1997)

Author: kboyarinov

include/oneapi/tbb/detail/_flow_graph_resource_limiting.h

@@ -21,7 +21,6 @@
 #error Do not #include this internal file directly; use public TBB headers instead.
 #endif
 
-#include <chrono>
 #include <unordered_map>
 #include <forward_list>
 #include <functional>
@@ -37,29 +36,21 @@ template <typename ResourceHandle>
 class resource_consumer_base;
 
 class request_id {
-    using clock_type = std::chrono::high_resolution_clock;
-    using time_point = clock_type::time_point;
-
-    time_point    m_time_point;
     std::uint64_t m_unique_integer;
 public:
     request_id(const std::uint64_t& unique_integer)
-        : m_time_point(clock_type::now())
-        , m_unique_integer(unique_integer)
+        : m_unique_integer(unique_integer)
     {}
 
-    struct hash {
-        std::hash<std::uint64_t> m_hash;
-
+    struct hash : protected std::hash<std::uint64_t> {
         std::size_t operator()(request_id id) const {
-            return m_hash(id.m_unique_integer);
+            return std::hash<std::uint64_t>::operator()(id.m_unique_integer);
         }
     };
 
-    struct equal {
+    struct equal : protected std::equal_to<std::uint64_t> {
         bool operator()(request_id lhs, request_id rhs) const {
-            return lhs.m_time_point == rhs.m_time_point &&
-                   lhs.m_unique_integer == rhs.m_unique_integer;
+            return std::equal_to<std::uint64_t>::operator()(lhs.m_unique_integer, rhs.m_unique_integer);
         }
     };
 }; // class request_id
@@ -495,9 +486,12 @@ class resource_limited_body_leaf
     void try_acquire_resources_and_execute(request_id id, request_data_type& req_data) {
         if (try_acquire_resources(id, req_data)) {
             // Access to all resources is granted
-            call_body(req_data.input_message, req_data.output_ports, req_data.handles);
-            release_resources(id, req_data);
-            remove_request(id);
+            try_call([&] {
+                call_body(req_data.input_message, req_data.output_ports, req_data.handles);
+            }).on_completion([&] {
+                release_resources(id, req_data);
+                remove_request(id);
+            });
         }
     }
 

test/tbb/test_resource_limited_node.cpp

@@ -25,6 +25,9 @@
 
 #include "tbb/flow_graph.h"
 
+//! \file test_resource_limited_node.cpp
+//! \brief Test for [preview] functionality
+
 using input_msg = conformance::message</*default_ctor = */true, /*copy_ctor = */true, /*copy_assign = */false>;
 using output_msg = conformance::message</*default_ctor = */false, /*copy_ctor = */false, /*copy_assign = */false>;
 
@@ -293,47 +296,78 @@ void test_root_genie() {
     // TODO: add fairness checks
 }
 
-void test_cancellation_with_active_requests() {
+void test_cancellation_with_active_requests(bool exception) {
     using namespace tbb::flow;
 
     int resource_value = 1;
     int input_value = 2;
     resource_limiter<int> limiter(resource_value);
 
-    using node_type = resource_limited_node<int, std::tuple<int>>;
+    using node_type = resource_limited_node<int, std::tuple<>>;
     using ports_type = typename node_type::output_ports_type;
+    
+#if TBB_USE_EXCEPTIONS
+    struct body_exception {};
+#endif
+    
+    tbb::task_group_context g2_context(tbb::task_group_context::isolated);
+    graph g1;
+    graph g2(g2_context);
 
-    graph g;
+    const std::size_t n_submissions = 100;
+    std::atomic<std::size_t> g2_node_body_counter{0};
 
-    node_type cancel_node(g, unlimited, std::tie(limiter),
-        [&](int input, ports_type& ports, int resource) {
+    node_type keep_using_node(g2, unlimited, std::tie(limiter),
+        [&](int input, ports_type&, int resource) {
             CHECK_MESSAGE(input == input_value, "Incorrect input");
             CHECK_MESSAGE(resource == resource_value, "Incorrect resource");
 
-            for (int i = 0; i < 100; ++i) {
-                std::get<0>(ports).try_put(input);
+            ++g2_node_body_counter;
+        });
+
+    node_type cancel_node(g1, unlimited, std::tie(limiter),
+        [&](int input, ports_type&, int resource) {
+            CHECK_MESSAGE(input == input_value, "Incorrect input");
+            CHECK_MESSAGE(resource == resource_value, "Incorrect resource");
+
+            for (std::size_t i = 0; i < n_submissions; ++i) {
+                keep_using_node.try_put(input);
             }
 
-            g.cancel();
+            if (exception) {
+#if TBB_USE_EXCEPTIONS
+                throw body_exception{};
+#else
+                CHECK_MESSAGE(false, "exception test was called when exceptions are not supported");
+#endif
+            } else {
+                g1.cancel();
+            }
 
-            for (int i = 0; i < 100; ++i) {
-                std::get<0>(ports).try_put(input);
+            for (std::size_t i = 0; i < n_submissions; ++i) {
+                keep_using_node.try_put(input);
             }
         });
 
-    std::atomic<std::size_t> num_bodies{0};
+    cancel_node.try_put(input_value);
 
-    node_type successor_node(g, unlimited, std::tie(limiter),
-        [&](int, ports_type&, int) {
-            ++num_bodies;
-        });
+#if TBB_USE_EXCEPTIONS
+    bool caught_exception = false;
+    try {
+        g1.wait_for_all();
+    } catch (body_exception) {
+        caught_exception = true;
+    }
 
-    make_edge(output_port<0>(cancel_node), successor_node);
+    CHECK_MESSAGE(exception == caught_exception, "Expected exception was not caught");
+#else
+    g1.wait_for_all();
+#endif
 
-    cancel_node.try_put(input_value);
-    g.wait_for_all();
-    CHECK_MESSAGE(num_bodies < tbb::this_task_arena::max_concurrency(),
-                  "Maximum number of node bodies exceeded");
+    g2.wait_for_all();
+    std::size_t expected_g2_body_calls = exception ? n_submissions : 2 * n_submissions;
+    CHECK_MESSAGE(g2_node_body_counter == expected_g2_body_calls,
+                  "Incorrect number of g2 node body calls");
 }
 
 //! \brief \ref interface
@@ -463,6 +497,9 @@ TEST_CASE("resource_limited_node and std::invoke") {
 
 //! \brief \ref error_guessing
 TEST_CASE("resource_limited_node cancellation with active requests") {
-    test_cancellation_with_active_requests();
+    test_cancellation_with_active_requests(/*exception =*/false);
+#if TBB_USE_EXCEPTIONS
+    test_cancellation_with_active_requests(/*exception = */true);
+#endif
 }
 #endif