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 <huanrwan@amd.com> Co-authored-by: Huanran Wang <huanran.wang@amd.com> Co-authored-by: Madsen, Jonathan <Jonathan.Madsen@amd.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
2025-08-25 08:57:41 -05:00
parent 748c9b74d9
commit b645010655
5 changed files with 142 additions and 0 deletions
@@ -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",