ROCm · ammarwa · Aug 25, 2025 · Jul 2, 2025 · Jul 7, 2025 · Jul 7, 2025
@@ -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)
 

@@ -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"

@@ -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);

@@ -52,6 +52,7 @@
 #include <glog/logging.h>
 
 #include <cxxabi.h>
+#include <semaphore.h>
 #include <sys/syscall.h>
 #include <sys/types.h>
 #include <unistd.h>
@@ -157,3 +158,47 @@ convert_marker_tracing_kind(TracingKindT val)
 {
     return convert_marker_tracing_kind(val, std::make_index_sequence<MARKER_API_LAST>{});
 }
+
+// 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;
+    }
+};
@@ -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<pid_t>{};
+        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<unsigned long>(_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",