From bb1a08c53f38c88d9e87b521580db36d2dbf8200 Mon Sep 17 00:00:00 2001 From: Huanran Wang Date: Wed, 2 Jul 2025 21:57:21 +0000 Subject: [PATCH 1/8] Using semaphore to sync with all peer processes in finalization stage [rocprofv3] Implement synchronization using POSIX semaphore in finalization --- source/lib/rocprofiler-sdk-tool/tool.cpp | 128 +++++++++++++++++++++++ 1 file changed, 128 insertions(+) diff --git a/source/lib/rocprofiler-sdk-tool/tool.cpp b/source/lib/rocprofiler-sdk-tool/tool.cpp index 7afcbd9f77..0cf37b9be4 100644 --- a/source/lib/rocprofiler-sdk-tool/tool.cpp +++ b/source/lib/rocprofiler-sdk-tool/tool.cpp @@ -100,6 +100,7 @@ #include #include +#include #include #include #include @@ -1534,6 +1535,132 @@ initialize_signal_handler(sigaction_func_t sigaction_func) } } +void +wait_peer_finished(const pid_t &pid, const pid_t &ppid) +{ + auto this_func = std::string_view{__FUNCTION__}; + + auto get_peer_pid = [&ppid]() { + auto fname = fmt::format("/proc/{}/task/{}/children", ppid, ppid); + auto ifs = std::ifstream{fname}; + auto peer_pid = std::vector{}; + while(ifs) + { + pid_t val = 0; + ifs >> val; + if(ifs && !ifs.eof() && val > 0) peer_pid.emplace_back(val); + } + return peer_pid; + }; + + auto _peer = get_peer_pid(); + + if(_peer.size() <= 1) + { + ROCP_INFO << fmt::format( + "[PPID={}][PID={}] has no peer process and no need to wait ", + ppid, + pid); + + // if no peer process no need to wait + return; + } + + ROCP_WARNING << fmt::format( + "[PPID={}][PID={}] rocprofv3 will wait for all {} peer processes under same parent finished to exit", + ppid, + pid, + _peer.size()); + + // Create a POSIX semaphore for synchronization + // Processes under the same parent share a same semaphore + sem_t *_sem = nullptr; + ROCP_INFO << fmt::format( + "[PPID={}][PID={}] Creating existing semaphore in {}", + ppid, + pid, + this_func); + + const std::string _sem_pid_group = "/finalization_process_sync_semaphore_pid_" + std::to_string(ppid); + _sem = sem_open(_sem_pid_group.c_str(), O_CREAT | O_EXCL, 0666, 0); + if(_sem == SEM_FAILED) + { + if(errno == EEXIST) + { + ROCP_INFO << fmt::format( + "[PPID={}][PID={}] Semaphore already exists in {}, opening existing semaphore", + ppid, + pid, + this_func); + _sem = sem_open(_sem_pid_group.c_str(), 0); + if(_sem == SEM_FAILED) + { + ROCP_WARNING << fmt::format( + "[PPID={}][PID={}] failed to open existing semaphore in {}", + ppid, + pid, + this_func); + } + } + else + { + ROCP_WARNING << fmt::format( + "[PPID={}][PID={}] failed to create semaphore in {}", + ppid, + pid, + this_func); + } + } + + // Post to semephore that this process has finished its work + if(sem_post(_sem) == -1) + { + ROCP_WARNING << fmt::format( + "[PPID={}][PID={}] failed to post to semaphore in {}", + ppid, + pid, + this_func); + } + + int _sem_val = 0; + do + { + if(sem_getvalue(_sem, &_sem_val) == -1) { + ROCP_WARNING << fmt::format( + "[PPID={}][PID={}] failed to get semaphore value in {}", + ppid, + pid, + this_func); + } + ROCP_TRACE << fmt::format( + "{} shows current sem_pid_group name: {} semaphore value: {}, peer size(): {}", + this_func, + _sem_pid_group, + _sem_val, + _peer.size()); + std::this_thread::sleep_for(std::chrono::milliseconds{100}); + } while(static_cast(_sem_val) < _peer.size()); + + // Clean up semaphore + if(sem_close(_sem) == -1) + { + ROCP_INFO << fmt::format( + "[PPID={}][PID={}] failed to close semaphore in {}", + ppid, + pid, + this_func); + } + + if(sem_unlink(_sem_pid_group.c_str()) == -1) + { + ROCP_WARNING << fmt::format( + "[PPID={}][PID={}] failed to unlink semaphore or it is already unlinked in {}", + ppid, + pid, + this_func); + } +} + void finalize_rocprofv3(std::string_view context) { @@ -2850,6 +2977,7 @@ rocprofv3_error_signal_handler(int signo, siginfo_t* info, void* ucontext) signo); finalize_rocprofv3(this_func); + wait_peer_finished(this_pid, this_ppid); ROCP_INFO << fmt::format( "[PPID={}][PID={}][TID={}][{}] rocprofv3 finalizing after signal {}... complete", From f05d5ed72acedc722e9dce0013517767eb7b0bad Mon Sep 17 00:00:00 2001 From: Huanran Wang Date: Mon, 7 Jul 2025 16:35:22 +0000 Subject: [PATCH 2/8] clang format code --- source/lib/rocprofiler-sdk-tool/tool.cpp | 76 ++++++++++-------------- 1 file changed, 32 insertions(+), 44 deletions(-) diff --git a/source/lib/rocprofiler-sdk-tool/tool.cpp b/source/lib/rocprofiler-sdk-tool/tool.cpp index 0cf37b9be4..eaf020c2de 100644 --- a/source/lib/rocprofiler-sdk-tool/tool.cpp +++ b/source/lib/rocprofiler-sdk-tool/tool.cpp @@ -106,7 +106,8 @@ #include #if defined(CODECOV) && CODECOV > 0 -extern "C" { +extern "C" +{ extern void __gcov_dump(void); } @@ -115,7 +116,8 @@ __gcov_dump(void); namespace common = ::rocprofiler::common; namespace tool = ::rocprofiler::tool; -extern "C" { +extern "C" +{ void rocprofv3_error_signal_handler(int signo, siginfo_t*, void*); } @@ -1536,7 +1538,7 @@ initialize_signal_handler(sigaction_func_t sigaction_func) } void -wait_peer_finished(const pid_t &pid, const pid_t &ppid) +wait_peer_finished(const pid_t& pid, const pid_t& ppid) { auto this_func = std::string_view{__FUNCTION__}; @@ -1558,34 +1560,30 @@ wait_peer_finished(const pid_t &pid, const pid_t &ppid) if(_peer.size() <= 1) { ROCP_INFO << fmt::format( - "[PPID={}][PID={}] has no peer process and no need to wait ", - ppid, - pid); + "[PPID={}][PID={}] has no peer process and no need to wait ", ppid, pid); // if no peer process no need to wait return; } - ROCP_WARNING << fmt::format( - "[PPID={}][PID={}] rocprofv3 will wait for all {} peer processes under same parent finished to exit", - ppid, - pid, - _peer.size()); + ROCP_WARNING << fmt::format("[PPID={}][PID={}] rocprofv3 will wait for all {} peer processes " + "under same parent finished to exit", + ppid, + pid, + _peer.size()); // Create a POSIX semaphore for synchronization // Processes under the same parent share a same semaphore - sem_t *_sem = nullptr; + sem_t* _sem = nullptr; ROCP_INFO << fmt::format( - "[PPID={}][PID={}] Creating existing semaphore in {}", - ppid, - pid, - this_func); + "[PPID={}][PID={}] Creating existing semaphore in {}", ppid, pid, this_func); - const std::string _sem_pid_group = "/finalization_process_sync_semaphore_pid_" + std::to_string(ppid); + const std::string _sem_pid_group = + "/finalization_process_sync_semaphore_pid_" + std::to_string(ppid); _sem = sem_open(_sem_pid_group.c_str(), O_CREAT | O_EXCL, 0666, 0); - if(_sem == SEM_FAILED) + if(_sem == SEM_FAILED) { - if(errno == EEXIST) + if(errno == EEXIST) { ROCP_INFO << fmt::format( "[PPID={}][PID={}] Semaphore already exists in {}, opening existing semaphore", @@ -1593,7 +1591,7 @@ wait_peer_finished(const pid_t &pid, const pid_t &ppid) pid, this_func); _sem = sem_open(_sem_pid_group.c_str(), 0); - if(_sem == SEM_FAILED) + if(_sem == SEM_FAILED) { ROCP_WARNING << fmt::format( "[PPID={}][PID={}] failed to open existing semaphore in {}", @@ -1605,38 +1603,30 @@ wait_peer_finished(const pid_t &pid, const pid_t &ppid) else { ROCP_WARNING << fmt::format( - "[PPID={}][PID={}] failed to create semaphore in {}", - ppid, - pid, - this_func); + "[PPID={}][PID={}] failed to create semaphore in {}", ppid, pid, this_func); } } // Post to semephore that this process has finished its work - if(sem_post(_sem) == -1) + if(sem_post(_sem) == -1) { ROCP_WARNING << fmt::format( - "[PPID={}][PID={}] failed to post to semaphore in {}", - ppid, - pid, - this_func); + "[PPID={}][PID={}] failed to post to semaphore in {}", ppid, pid, this_func); } int _sem_val = 0; do { - if(sem_getvalue(_sem, &_sem_val) == -1) { + if(sem_getvalue(_sem, &_sem_val) == -1) + { ROCP_WARNING << fmt::format( - "[PPID={}][PID={}] failed to get semaphore value in {}", - ppid, - pid, - this_func); + "[PPID={}][PID={}] failed to get semaphore value in {}", ppid, pid, this_func); } ROCP_TRACE << fmt::format( - "{} shows current sem_pid_group name: {} semaphore value: {}, peer size(): {}", - this_func, - _sem_pid_group, - _sem_val, + "{} shows current sem_pid_group name: {} semaphore value: {}, peer size(): {}", + this_func, + _sem_pid_group, + _sem_val, _peer.size()); std::this_thread::sleep_for(std::chrono::milliseconds{100}); } while(static_cast(_sem_val) < _peer.size()); @@ -1645,12 +1635,9 @@ wait_peer_finished(const pid_t &pid, const pid_t &ppid) if(sem_close(_sem) == -1) { ROCP_INFO << fmt::format( - "[PPID={}][PID={}] failed to close semaphore in {}", - ppid, - pid, - this_func); + "[PPID={}][PID={}] failed to close semaphore in {}", ppid, pid, this_func); } - + if(sem_unlink(_sem_pid_group.c_str()) == -1) { ROCP_WARNING << fmt::format( @@ -2804,7 +2791,8 @@ wait_pid(pid_t _pid, int _opts = 0) return _status; } -extern "C" { +extern "C" +{ void rocprofv3_set_main(main_func_t main_func) ROCPROFV3_INTERNAL_API; From f6e2147111e7eb059a758b35a8fbd5b73e6d92fe Mon Sep 17 00:00:00 2001 From: Huanran Wang Date: Mon, 7 Jul 2025 16:52:13 +0000 Subject: [PATCH 3/8] clang 11 format code --- source/lib/rocprofiler-sdk-tool/tool.cpp | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/source/lib/rocprofiler-sdk-tool/tool.cpp b/source/lib/rocprofiler-sdk-tool/tool.cpp index eaf020c2de..a15f9d1481 100644 --- a/source/lib/rocprofiler-sdk-tool/tool.cpp +++ b/source/lib/rocprofiler-sdk-tool/tool.cpp @@ -106,8 +106,7 @@ #include #if defined(CODECOV) && CODECOV > 0 -extern "C" -{ +extern "C" { extern void __gcov_dump(void); } @@ -116,8 +115,7 @@ __gcov_dump(void); namespace common = ::rocprofiler::common; namespace tool = ::rocprofiler::tool; -extern "C" -{ +extern "C" { void rocprofv3_error_signal_handler(int signo, siginfo_t*, void*); } @@ -2791,8 +2789,7 @@ wait_pid(pid_t _pid, int _opts = 0) return _status; } -extern "C" -{ +extern "C" { void rocprofv3_set_main(main_func_t main_func) ROCPROFV3_INTERNAL_API; From d008a0eeaa85711d4855630e867c7e50b1e776ad Mon Sep 17 00:00:00 2001 From: Huanran Wang Date: Mon, 14 Jul 2025 22:16:19 +0000 Subject: [PATCH 4/8] Add process sync option for rocprofv3 --- source/bin/rocprofv3.py | 13 +++++++++++++ source/docs/rocprofv3_input_schema.json | 4 ++++ source/lib/rocprofiler-sdk-tool/config.hpp | 2 ++ source/lib/rocprofiler-sdk-tool/tool.cpp | 2 +- 4 files changed, 20 insertions(+), 1 deletion(-) diff --git a/source/bin/rocprofv3.py b/source/bin/rocprofv3.py index bc6be7a75c..0cb7fecacf 100755 --- a/source/bin/rocprofv3.py +++ b/source/bin/rocprofv3.py @@ -705,6 +705,16 @@ def add_parser_bool_argument(gparser, *args, **kwargs): Note: glog still installs signal handlers which provide backtraces""", ) + add_parser_bool_argument( + advanced_options, + "--process-sync", + help="""Enables the process synchronization in the rocprofv3 tool finalization stage. + When --process-sync is set to true, + and rocprofv3 tool will force process to wait for its peer processes finishing write the trace data, + then they proceed. + Note: some workloads will teminate the process group when one of the process is finished""", + ) + advanced_options.add_argument( "--minimum-output-data", help="""Output files are generated only if output data size > minimum output data". @@ -1509,6 +1519,9 @@ def log_config(_env): if args.disable_signal_handlers is not None: update_env("ROCPROF_SIGNAL_HANDLERS", not args.disable_signal_handlers) + if args.process_sync is not None: + update_env("ROCPROF_PROCESS_SYNC", args.process_sync) + if args.minimum_output_data: update_env("ROCPROF_MINIMUM_OUTPUT_BYTES", args.minimum_output_data * 1024) diff --git a/source/docs/rocprofv3_input_schema.json b/source/docs/rocprofv3_input_schema.json index 4b697ee856..e507bf5e09 100644 --- a/source/docs/rocprofv3_input_schema.json +++ b/source/docs/rocprofv3_input_schema.json @@ -174,6 +174,10 @@ "type": "boolean", "description": "Disables the signal handlers in the rocprofv3 tool. When --disable-signal-handlers is set to true, and application has its signal handler on SIGSEGV or similar installed, then its signal handler will be used not the rocprofv3 signal handler. Note: glog still installs signal handlers which provide backtraces" }, + "process-sync":{ + "type": "boolean", + "description": "Enables the process synchronization in the rocprofv3 tool finalization stage. When --process-sync is set to true, rocprofv3 tool will force process to wait for its peer processes finishing write the trace data, then they proceed. Note: some workload will teminate the process group when one of the process is finished" + }, "pc_sampling_unit": { "type": "string", "description": "pc sampling unit" diff --git a/source/lib/rocprofiler-sdk-tool/config.hpp b/source/lib/rocprofiler-sdk-tool/config.hpp index 1bc1f826f3..f8e8f4f254 100644 --- a/source/lib/rocprofiler-sdk-tool/config.hpp +++ b/source/lib/rocprofiler-sdk-tool/config.hpp @@ -127,6 +127,7 @@ struct config : output_config bool advanced_thread_trace = get_env("ROCPROF_ADVANCED_THREAD_TRACE", false); bool att_serialize_all = get_env("ROCPROF_ATT_PARAM_SERIALIZE_ALL", false); bool enable_signal_handlers = get_env("ROCPROF_SIGNAL_HANDLERS", true); + bool enable_process_sync = get_env("ROCPROF_PROCESS_SYNC", true); bool selected_regions = get_env("ROCPROF_SELECTED_REGIONS", false); bool output_config_file = get_env("ROCPROF_OUTPUT_CONFIG_FILE", false); bool pc_sampling_host_trap = false; @@ -227,6 +228,7 @@ config::save(ArchiveT& ar) const CFG_SERIALIZE_MEMBER(truncate); CFG_SERIALIZE_MEMBER(minimum_output_bytes); CFG_SERIALIZE_MEMBER(enable_signal_handlers); + CFG_SERIALIZE_MEMBER(enable_process_sync); CFG_SERIALIZE_MEMBER(selected_regions); CFG_SERIALIZE_MEMBER(counter_groups_random_seed); diff --git a/source/lib/rocprofiler-sdk-tool/tool.cpp b/source/lib/rocprofiler-sdk-tool/tool.cpp index a15f9d1481..44c970fd18 100644 --- a/source/lib/rocprofiler-sdk-tool/tool.cpp +++ b/source/lib/rocprofiler-sdk-tool/tool.cpp @@ -2962,7 +2962,7 @@ rocprofv3_error_signal_handler(int signo, siginfo_t* info, void* ucontext) signo); finalize_rocprofv3(this_func); - wait_peer_finished(this_pid, this_ppid); + if(tool::get_config().enable_process_sync) wait_peer_finished(this_pid, this_ppid); ROCP_INFO << fmt::format( "[PPID={}][PID={}][TID={}][{}] rocprofv3 finalizing after signal {}... complete", From 6d97b1b8373c6bdcb216124c4426663295f6b3e7 Mon Sep 17 00:00:00 2001 From: Huanran Wang Date: Tue, 15 Jul 2025 21:54:33 +0000 Subject: [PATCH 5/8] Default value of process sync is false --- source/lib/rocprofiler-sdk-tool/config.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/lib/rocprofiler-sdk-tool/config.hpp b/source/lib/rocprofiler-sdk-tool/config.hpp index f8e8f4f254..506ccdac12 100644 --- a/source/lib/rocprofiler-sdk-tool/config.hpp +++ b/source/lib/rocprofiler-sdk-tool/config.hpp @@ -127,7 +127,7 @@ struct config : output_config bool advanced_thread_trace = get_env("ROCPROF_ADVANCED_THREAD_TRACE", false); bool att_serialize_all = get_env("ROCPROF_ATT_PARAM_SERIALIZE_ALL", false); bool enable_signal_handlers = get_env("ROCPROF_SIGNAL_HANDLERS", true); - bool enable_process_sync = get_env("ROCPROF_PROCESS_SYNC", true); + bool enable_process_sync = get_env("ROCPROF_PROCESS_SYNC", false); bool selected_regions = get_env("ROCPROF_SELECTED_REGIONS", false); bool output_config_file = get_env("ROCPROF_OUTPUT_CONFIG_FILE", false); bool pc_sampling_host_trap = false; From 9bb0ef42aa57a55819d123aac82573372f5743bf Mon Sep 17 00:00:00 2001 From: "Madsen, Jonathan" Date: Wed, 23 Jul 2025 16:14:07 -0500 Subject: [PATCH 6/8] Update source/lib/rocprofiler-sdk-tool/tool.cpp Apply suggestion by Copilot Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- source/lib/rocprofiler-sdk-tool/tool.cpp | 31 +++++++++++++++--------- 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/source/lib/rocprofiler-sdk-tool/tool.cpp b/source/lib/rocprofiler-sdk-tool/tool.cpp index 44c970fd18..a6653ca5ef 100644 --- a/source/lib/rocprofiler-sdk-tool/tool.cpp +++ b/source/lib/rocprofiler-sdk-tool/tool.cpp @@ -1612,22 +1612,29 @@ wait_peer_finished(const pid_t& pid, const pid_t& ppid) "[PPID={}][PID={}] failed to post to semaphore in {}", ppid, pid, this_func); } - int _sem_val = 0; - do + for(size_t i = 0; i < _peer.size(); ++i) { - if(sem_getvalue(_sem, &_sem_val) == -1) + ROCP_TRACE << fmt::format( + "{} waiting on semaphore for peer {}/{} in group: {}", + this_func, + i + 1, + _peer.size(), + _sem_pid_group); + if(sem_wait(_sem) == -1) { ROCP_WARNING << fmt::format( - "[PPID={}][PID={}] failed to get semaphore value in {}", ppid, pid, this_func); + "[PPID={}][PID={}] failed to wait on semaphore in {}", ppid, pid, this_func); } - ROCP_TRACE << fmt::format( - "{} shows current sem_pid_group name: {} semaphore value: {}, peer size(): {}", - this_func, - _sem_pid_group, - _sem_val, - _peer.size()); - std::this_thread::sleep_for(std::chrono::milliseconds{100}); - } while(static_cast(_sem_val) < _peer.size()); + else + { + ROCP_TRACE << fmt::format( + "{} successfully waited on semaphore for peer {}/{} in group: {}", + this_func, + i + 1, + _peer.size(), + _sem_pid_group); + } + } // Clean up semaphore if(sem_close(_sem) == -1) From 2295ee373a589f53f28934a8d6854c7b2d405839 Mon Sep 17 00:00:00 2001 From: Huanran Wang Date: Wed, 6 Aug 2025 20:13:09 +0000 Subject: [PATCH 7/8] update according to comments --- source/lib/rocprofiler-sdk-tool/helper.hpp | 45 ++++++++++ source/lib/rocprofiler-sdk-tool/tool.cpp | 96 ++++++---------------- 2 files changed, 72 insertions(+), 69 deletions(-) diff --git a/source/lib/rocprofiler-sdk-tool/helper.hpp b/source/lib/rocprofiler-sdk-tool/helper.hpp index a7f9272805..64e6dd5044 100644 --- a/source/lib/rocprofiler-sdk-tool/helper.hpp +++ b/source/lib/rocprofiler-sdk-tool/helper.hpp @@ -52,6 +52,7 @@ #include #include +#include #include #include #include @@ -157,3 +158,47 @@ convert_marker_tracing_kind(TracingKindT val) { return convert_marker_tracing_kind(val, std::make_index_sequence{}); } + +// RAII wrapper for semaphore to cleanup and +// sync worker processes in finalization process +struct SemaphoreGuard +{ + sem_t* sem = nullptr; + std::string name; + + SemaphoreGuard(const std::string& sem_name) + : name(sem_name) + {} + + ~SemaphoreGuard() + { + if(sem != nullptr) + { + if(sem_close(sem) == -1) + { + ROCP_WARNING << fmt::format("Failed to close semaphore in"); + } + + if(sem_unlink(name.c_str()) == -1) + { + ROCP_WARNING << fmt::format("Failed to unlink semaphore or it is already unlinked"); + } + } + } + + bool open_or_create() + { + // Try to create new semaphore + sem = sem_open(name.c_str(), O_CREAT | O_EXCL, 0666, 0); + if(sem != SEM_FAILED) return true; + + // If exists, open existing + if(errno == EEXIST) + { + sem = sem_open(name.c_str(), 0); + return (sem != SEM_FAILED); + } + + return false; + } +}; \ No newline at end of file diff --git a/source/lib/rocprofiler-sdk-tool/tool.cpp b/source/lib/rocprofiler-sdk-tool/tool.cpp index a6653ca5ef..0290dbebf4 100644 --- a/source/lib/rocprofiler-sdk-tool/tool.cpp +++ b/source/lib/rocprofiler-sdk-tool/tool.cpp @@ -100,7 +100,6 @@ #include #include -#include #include #include #include @@ -1553,9 +1552,9 @@ wait_peer_finished(const pid_t& pid, const pid_t& ppid) return peer_pid; }; - auto _peer = get_peer_pid(); + auto _peers_pid = get_peer_pid(); - if(_peer.size() <= 1) + if(_peers_pid.size() <= 1) { ROCP_INFO << fmt::format( "[PPID={}][PID={}] has no peer process and no need to wait ", ppid, pid); @@ -1568,89 +1567,48 @@ wait_peer_finished(const pid_t& pid, const pid_t& ppid) "under same parent finished to exit", ppid, pid, - _peer.size()); + _peers_pid.size()); // Create a POSIX semaphore for synchronization // Processes under the same parent share a same semaphore - sem_t* _sem = nullptr; ROCP_INFO << fmt::format( "[PPID={}][PID={}] Creating existing semaphore in {}", ppid, pid, this_func); - const std::string _sem_pid_group = - "/finalization_process_sync_semaphore_pid_" + std::to_string(ppid); - _sem = sem_open(_sem_pid_group.c_str(), O_CREAT | O_EXCL, 0666, 0); - if(_sem == SEM_FAILED) + const std::string _sem_name = "/rocprofv3_sync_pid_" + std::to_string(ppid); + auto guard = SemaphoreGuard{_sem_name}; + + if(!guard.open_or_create()) { - if(errno == EEXIST) - { - ROCP_INFO << fmt::format( - "[PPID={}][PID={}] Semaphore already exists in {}, opening existing semaphore", - ppid, - pid, - this_func); - _sem = sem_open(_sem_pid_group.c_str(), 0); - if(_sem == SEM_FAILED) - { - ROCP_WARNING << fmt::format( - "[PPID={}][PID={}] failed to open existing semaphore in {}", - ppid, - pid, - this_func); - } - } - else - { - ROCP_WARNING << fmt::format( - "[PPID={}][PID={}] failed to create semaphore in {}", ppid, pid, this_func); - } + ROCP_WARNING << fmt::format( + "[PPID={}][PID={}] Failed to initialize semaphore, skipping sync", ppid, pid); + return; } - // Post to semephore that this process has finished its work - if(sem_post(_sem) == -1) + // Signal completion and wait for peers + if(sem_post(guard.sem) == -1) { - ROCP_WARNING << fmt::format( - "[PPID={}][PID={}] failed to post to semaphore in {}", ppid, pid, this_func); + ROCP_WARNING << fmt::format("[PPID={}][PID={}] Failed to signal completion", ppid, pid); + return; } - for(size_t i = 0; i < _peer.size(); ++i) + // Wait for all peers to signal completion + int _sem_val = 0; + do { - ROCP_TRACE << fmt::format( - "{} waiting on semaphore for peer {}/{} in group: {}", - this_func, - i + 1, - _peer.size(), - _sem_pid_group); - if(sem_wait(_sem) == -1) + if(sem_getvalue(guard.sem, &_sem_val) == -1) { ROCP_WARNING << fmt::format( "[PPID={}][PID={}] failed to wait on semaphore in {}", ppid, pid, this_func); } - else - { - ROCP_TRACE << fmt::format( - "{} successfully waited on semaphore for peer {}/{} in group: {}", - this_func, - i + 1, - _peer.size(), - _sem_pid_group); - } - } - - // Clean up semaphore - if(sem_close(_sem) == -1) - { - ROCP_INFO << fmt::format( - "[PPID={}][PID={}] failed to close semaphore in {}", ppid, pid, this_func); - } - - if(sem_unlink(_sem_pid_group.c_str()) == -1) - { - ROCP_WARNING << fmt::format( - "[PPID={}][PID={}] failed to unlink semaphore or it is already unlinked in {}", - ppid, - pid, - this_func); - } + ROCP_TRACE << fmt::format( + "{} shows current sem_pid_group name: {} semaphore value: {}, peer size(): {}", + this_func, + _sem_name, + _sem_val, + _peers_pid.size()); + std::this_thread::yield(); + std::this_thread::sleep_for(std::chrono::milliseconds{100}); + } while(static_cast(_sem_val) < _peers_pid.size()); } void From e6761819a7c7b18529d6dfc43e955cae925120e9 Mon Sep 17 00:00:00 2001 From: Huanran Wang Date: Wed, 6 Aug 2025 20:39:35 +0000 Subject: [PATCH 8/8] add new line to helper.hpp --- source/lib/rocprofiler-sdk-tool/helper.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/lib/rocprofiler-sdk-tool/helper.hpp b/source/lib/rocprofiler-sdk-tool/helper.hpp index 64e6dd5044..df99f7a950 100644 --- a/source/lib/rocprofiler-sdk-tool/helper.hpp +++ b/source/lib/rocprofiler-sdk-tool/helper.hpp @@ -201,4 +201,4 @@ struct SemaphoreGuard return false; } -}; \ No newline at end of file +};