From b645010655c815e50e13ce5fb728857c039b151d Mon Sep 17 00:00:00 2001 From: "systems-assistant[bot]" <221163467+systems-assistant[bot]@users.noreply.github.com> Date: Mon, 25 Aug 2025 08:57:41 -0500 Subject: [PATCH] Using semaphore to sync with all peer processes in finalization stage (#169) * Using semaphore to sync with all peer processes in finalization stage [rocprofv3] Implement synchronization using POSIX semaphore in finalization * clang format code * clang 11 format code * Add process sync option for rocprofv3 * Default value of process sync is false * Update source/lib/rocprofiler-sdk-tool/tool.cpp Apply suggestion by Copilot Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * update according to comments * add new line to helper.hpp --------- Co-authored-by: Huanran Wang Co-authored-by: Huanran Wang Co-authored-by: Madsen, Jonathan Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../rocprofiler-sdk/source/bin/rocprofv3.py | 13 ++++ .../source/docs/rocprofv3_input_schema.json | 4 + .../lib/rocprofiler-sdk-tool/config.hpp | 2 + .../lib/rocprofiler-sdk-tool/helper.hpp | 45 +++++++++++ .../source/lib/rocprofiler-sdk-tool/tool.cpp | 78 +++++++++++++++++++ 5 files changed, 142 insertions(+) diff --git a/projects/rocprofiler-sdk/source/bin/rocprofv3.py b/projects/rocprofiler-sdk/source/bin/rocprofv3.py index bc6be7a75c..0cb7fecacf 100755 --- a/projects/rocprofiler-sdk/source/bin/rocprofv3.py +++ b/projects/rocprofiler-sdk/source/bin/rocprofv3.py @@ -705,6 +705,16 @@ For MPI applications (or other job launchers such as SLURM), place rocprofv3 ins 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 run(app_args, args, **kwargs): 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/projects/rocprofiler-sdk/source/docs/rocprofv3_input_schema.json b/projects/rocprofiler-sdk/source/docs/rocprofv3_input_schema.json index 4b697ee856..e507bf5e09 100644 --- a/projects/rocprofiler-sdk/source/docs/rocprofv3_input_schema.json +++ b/projects/rocprofiler-sdk/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/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/config.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/config.hpp index 1bc1f826f3..506ccdac12 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/config.hpp +++ b/projects/rocprofiler-sdk/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", 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; @@ -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/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/helper.hpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/helper.hpp index a7f9272805..df99f7a950 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/helper.hpp +++ b/projects/rocprofiler-sdk/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; + } +}; diff --git a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/tool.cpp b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/tool.cpp index 7afcbd9f77..0290dbebf4 100644 --- a/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/tool.cpp +++ b/projects/rocprofiler-sdk/source/lib/rocprofiler-sdk-tool/tool.cpp @@ -1534,6 +1534,83 @@ 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 _peers_pid = get_peer_pid(); + + if(_peers_pid.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, + _peers_pid.size()); + + // Create a POSIX semaphore for synchronization + // Processes under the same parent share a same semaphore + ROCP_INFO << fmt::format( + "[PPID={}][PID={}] Creating existing semaphore in {}", ppid, pid, this_func); + + const std::string _sem_name = "/rocprofv3_sync_pid_" + std::to_string(ppid); + auto guard = SemaphoreGuard{_sem_name}; + + if(!guard.open_or_create()) + { + ROCP_WARNING << fmt::format( + "[PPID={}][PID={}] Failed to initialize semaphore, skipping sync", ppid, pid); + return; + } + + // Signal completion and wait for peers + if(sem_post(guard.sem) == -1) + { + ROCP_WARNING << fmt::format("[PPID={}][PID={}] Failed to signal completion", ppid, pid); + return; + } + + // Wait for all peers to signal completion + int _sem_val = 0; + do + { + if(sem_getvalue(guard.sem, &_sem_val) == -1) + { + ROCP_WARNING << fmt::format( + "[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_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 finalize_rocprofv3(std::string_view context) { @@ -2850,6 +2927,7 @@ rocprofv3_error_signal_handler(int signo, siginfo_t* info, void* ucontext) signo); finalize_rocprofv3(this_func); + 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",