SWDEV-489158: Fix for exit thread safety (#61)

* SWDEV-489158: Fix for exit thread safety

* Fixed exit thread logic

* Force CI to rerun

* Remove .vscode

* Fix thread safety bug

* Addressed some comments

* Formatting

---------

Co-authored-by: Giovanni Baraldi <gbaraldi@amd.com>

[ROCm/rocprofiler-sdk commit: f4984f9]

Author: Baraldi, Giovanni

samples/counter_collection/callback_client.cpp

@@ -59,6 +59,12 @@ start()
 
 namespace
 {
+struct tool_data_t
+{
+    std::mutex    mut{};
+    std::ostream* output_stream{nullptr};
+};
+
 rocprofiler_context_id_t&
 get_client_ctx()
 {
@@ -70,8 +76,8 @@ void
 record_callback(rocprofiler_dispatch_counting_service_data_t dispatch_data,
                 rocprofiler_record_counter_t*                record_data,
                 size_t                                       record_count,
-                rocprofiler_user_data_t                      user_data,
-                void*                                        callback_data_args)
+                rocprofiler_user_data_t /* user_data */,
+                void* callback_data_args)
 {
     std::stringstream ss;
     ss << "Dispatch_Id=" << dispatch_data.dispatch_info.dispatch_id
@@ -81,11 +87,11 @@ record_callback(rocprofiler_dispatch_counting_service_data_t dispatch_data,
         ss << "(Id: " << record_data[i].id << " Value [D]: " << record_data[i].counter_value
            << "),";
 
-    auto* output_stream = static_cast<std::ostream*>(callback_data_args);
-    if(!output_stream) throw std::runtime_error{"nullptr to output stream"};
-    *output_stream << "[" << __FUNCTION__ << "] " << ss.str() << "\n";
+    auto* tool = static_cast<tool_data_t*>(callback_data_args);
+    if(!tool || !tool->output_stream) throw std::runtime_error{"nullptr to output stream"};
 
-    (void) user_data;
+    auto _lk = std::unique_lock{tool->mut};
+    *tool->output_stream << "[" << __FUNCTION__ << "] " << ss.str() << "\n";
 }
 
 /**
@@ -197,12 +203,20 @@ tool_init(rocprofiler_client_finalize_t, void* user_data)
 void
 tool_fini(void* user_data)
 {
+    assert(user_data);
     std::clog << "In tool fini\n";
     rocprofiler_stop_context(get_client_ctx());
+    auto* tool_data = static_cast<tool_data_t*>(user_data);
+
+    {
+        auto  _lk           = std::unique_lock{tool_data->mut};
+        auto* output_stream = tool_data->output_stream;
 
-    auto* output_stream = static_cast<std::ostream*>(user_data);
-    *output_stream << std::flush;
-    if(output_stream != &std::cout && output_stream != &std::cerr) delete output_stream;
+        *output_stream << std::flush;
+        if(output_stream != &std::cout && output_stream != &std::cerr) delete output_stream;
+    }
+
+    delete tool_data;
 }
 }  // namespace
 
@@ -227,22 +241,23 @@ rocprofiler_configure(uint32_t                 version,
 
     std::clog << info.str() << std::endl;
 
-    std::ostream* output_stream = nullptr;
-    std::string   filename      = "counter_collection.log";
+    auto* tool_data = new tool_data_t{};
+
+    std::string filename = "counter_collection.log";
     if(auto* outfile = getenv("ROCPROFILER_SAMPLE_OUTPUT_FILE"); outfile) filename = outfile;
     if(filename == "stdout")
-        output_stream = &std::cout;
+        tool_data->output_stream = &std::cout;
     else if(filename == "stderr")
-        output_stream = &std::cerr;
+        tool_data->output_stream = &std::cerr;
     else
-        output_stream = new std::ofstream{filename};
+        tool_data->output_stream = new std::ofstream{filename};
 
     // create configure data
     static auto cfg =
         rocprofiler_tool_configure_result_t{sizeof(rocprofiler_tool_configure_result_t),
                                             &tool_init,
                                             &tool_fini,
-                                            static_cast<void*>(output_stream)};
+                                            static_cast<void*>(tool_data)};
 
     // return pointer to configure data
     return &cfg;

source/lib/output/format_path.cpp

@@ -43,6 +43,8 @@
 #include <cstring>
 #include <ctime>
 #include <fstream>
+#include <limits>
+#include <locale>
 #include <regex>
 #include <set>
 #include <sstream>
@@ -66,6 +68,17 @@ const auto env_regexes =
 //  - %q{USER}          Compatibility with NVIDIA
 //
 
+// https://gcc.gnu.org/bugzilla/show_bug.cgi?id=77704
+// NOLINTBEGIN
+[[maybe_unused]] volatile bool _initLocale = []() {
+    const std::ctype<char>& ct(std::use_facet<std::ctype<char>>(std::locale()));
+    for(size_t i = 0; i <= std::numeric_limits<unsigned char>::max(); i++)
+        ct.narrow(static_cast<char>(i), '\0');
+    ct.narrow(0, 0, 0, 0);
+    return true;
+}();
+// NOLINTEND
+
 std::string
 format_path_impl(std::string _fpath, const std::vector<output_key>& _keys)
 {

source/lib/rocprofiler-sdk/counters/sample_consumer.hpp

@@ -44,21 +44,23 @@ class consumer_thread_t
 
     void start()
     {
-        {
-            std::unique_lock<std::mutex> lk(mut);
-            if(valid.exchange(true)) return;
-        }
+        std::unique_lock<std::mutex> lk(mut);
+
+        if(valid.exchange(true)) return;
+        exited.store(false);
+
         consumer = std::thread{&consumer_thread_t::consumer_loop, this};
     }
 
     void exit()
     {
-        {
-            std::unique_lock<std::mutex> lk(mut);
-            if(!valid.exchange(false)) return;
-            cv.notify_one();
-        }
-        consumer.join();
+        std::unique_lock<std::mutex> lk(mut);
+
+        valid.store(false);
+        cv.notify_all();
+
+        if(!exited) cv.wait(lk, [&] { return exited.load(); });
+        if(consumer.joinable()) consumer.join();
     }
 
     void add(DataType&& params)
@@ -74,7 +76,7 @@ class consumer_thread_t
 
         buffer.at(write_ptr % buffer.size()) = std::move(params);
         write_ptr.fetch_add(1);
-        cv.notify_one();
+        cv.notify_all();
     }
 
 protected:
@@ -86,7 +88,12 @@ class consumer_thread_t
             {
                 std::unique_lock<std::mutex> lk(mut);
                 cv.wait(lk, [&] { return read_ptr != write_ptr || !valid; });
-                if(!valid && read_ptr == write_ptr) return;
+                if(!valid && read_ptr == write_ptr)
+                {
+                    exited.store(true);
+                    cv.notify_all();
+                    return;
+                }
             }
 
             auto retrieved = std::move(buffer.at(read_ptr % buffer.size()));
@@ -97,6 +104,7 @@ class consumer_thread_t
 
     consume_func_t             consume_fn;
     std::atomic<bool>          valid{false};
+    std::atomic<bool>          exited{true};
     std::mutex                 mut;
     std::atomic<size_t>        write_ptr{0};
     std::atomic<size_t>        read_ptr{0};

tests/rocprofv3/tracing-plus-counter-collection/CMakeLists.txt

@@ -30,7 +30,7 @@ set(cc-tracing-env "${PRELOAD_ENV}")
 
 set_tests_properties(
     rocprofv3-test-tracing-plus-counter-collection-execute
-    PROPERTIES TIMEOUT 45 LABELS "integration-tests;application-replay" ENVIRONMENT
+    PROPERTIES TIMEOUT 90 LABELS "integration-tests;application-replay" ENVIRONMENT
                "${cc-tracing-env}" FAIL_REGULAR_EXPRESSION
                "${ROCPROFILER_DEFAULT_FAIL_REGEX}")