Introduced worker registry for better graceful coordinator shutdown (#20)

intuibase · web-flow · commit 3678b46e8d6c · 2026-02-17T12:46:20.000+01:00
diff --git a/prod/native/libcommon/code/AgentGlobals.cpp b/prod/native/libcommon/code/AgentGlobals.cpp
@@ -59,12 +59,13 @@ AgentGlobals::AgentGlobals(std::shared_ptr<LoggerInterface> logger,
     opAmp_(std::make_shared<opentelemetry::php::transport::OpAmp>(logger_, config_, httpTransportAsync_, resourceDetector_)),
     sharedMemory_(std::make_shared<opentelemetry::php::SharedMemoryState>()),
     requestScope_(std::make_shared<opentelemetry::php::RequestScope>(logger_, bridge_, sapi_, sharedMemory_, dependencyAutoLoaderGuard_, inferredSpans_, config_, [hs = hooksStorage_]() { hs->clear(); }, [this]() { return getPeriodicTaskExecutor();}, [this]() { return coordinatorConfigProvider_->triggerUpdateIfChanged(); })),
-    messagesDispatcher_(std::make_shared<opentelemetry::php::coordinator::CoordinatorMessagesDispatcher>(logger_, httpTransportAsync_)),
+    workerRegistry_(std::make_shared<opentelemetry::php::coordinator::WorkerRegistry>(logger_)),
+    messagesDispatcher_(std::make_shared<opentelemetry::php::coordinator::CoordinatorMessagesDispatcher>(logger_, httpTransportAsync_, workerRegistry_)),
     coordinatorConfigProvider_(std::make_shared<opentelemetry::php::coordinator::CoordinatorConfigurationProvider>(logger_, opAmp_)),
-    coordinatorProcess_(std::make_shared<opentelemetry::php::coordinator::CoordinatorProcess>(logger_, messagesDispatcher_, coordinatorConfigProvider_))
+    coordinatorProcess_(std::make_shared<opentelemetry::php::coordinator::CoordinatorProcess>(logger_, messagesDispatcher_, coordinatorConfigProvider_, workerRegistry_))
     {
-        forkableRegistry_->registerForkable(httpTransportAsync_);
-        forkableRegistry_->registerForkable(opAmp_);
+        // forkableRegistry_->registerForkable(httpTransportAsync_);
+        // forkableRegistry_->registerForkable(opAmp_);
 
         configManager_->attachLogger(logger_);
 
diff --git a/prod/native/libcommon/code/AgentGlobals.h b/prod/native/libcommon/code/AgentGlobals.h
@@ -2,6 +2,7 @@
 
 #include "PhpSapi.h"
 #include "Logger.h"
+#include "coordinator/WorkerRegistry.h"
 #include <functional>
 #include <memory>
 
@@ -39,6 +40,7 @@ namespace coordinator {
 class CoordinatorMessagesDispatcher;
 class CoordinatorProcess;
 class CoordinatorConfigurationProvider;
+class WorkerRegistry;
 } // namespace coordinator
 namespace transport {
 class CurlSender;
@@ -83,6 +85,7 @@ class AgentGlobals {
     std::shared_ptr<opentelemetry::php::transport::OpAmp> opAmp_;
     std::shared_ptr<SharedMemoryState> sharedMemory_;
     std::shared_ptr<RequestScope> requestScope_;
+    std::shared_ptr<coordinator::WorkerRegistry> workerRegistry_;
     std::shared_ptr<coordinator::CoordinatorMessagesDispatcher> messagesDispatcher_;
     std::shared_ptr<coordinator::CoordinatorConfigurationProvider> coordinatorConfigProvider_;
     std::shared_ptr<coordinator::CoordinatorProcess> coordinatorProcess_;
diff --git a/prod/native/libcommon/code/coordinator/CoordinatorConfigurationProvider.h b/prod/native/libcommon/code/coordinator/CoordinatorConfigurationProvider.h
@@ -4,7 +4,6 @@
 #include "transport/OpAmp.h"
 
 #include <memory>
-#include <span>
 #include <string>
 #include <utility>
 
diff --git a/prod/native/libcommon/code/coordinator/CoordinatorMessagesDispatcher.cpp b/prod/native/libcommon/code/coordinator/CoordinatorMessagesDispatcher.cpp
@@ -1,4 +1,5 @@
 #include "CoordinatorMessagesDispatcher.h"
+#include "LoggerInterface.h"
 #include "coordinator/proto/CoordinatorCommands.pb.h"
 
 namespace opentelemetry::php::coordinator {
@@ -60,6 +61,17 @@ void CoordinatorMessagesDispatcher::processRecievedMessage(const std::span<const
             httpTransport_->enqueue(p.endpoint_hash(), buf);
             break;
         }
+        case coordinator::CoordinatorCommand::WORKER_STARTED: {
+            ELOG_DEBUG(logger_, COORDINATOR, "CoordinatorMessagesDispatcher: Worker started");
+            workerRegistry_->registerWorker(command.worker_started().process_id(), command.worker_started().parent_process_id());
+            break;
+        }
+        case coordinator::CoordinatorCommand::WORKER_IS_GOING_TO_SHUTDOWN: {
+            ELOG_DEBUG(logger_, COORDINATOR, "CoordinatorMessagesDispatcher: Worker pid: {} ppid: {} is going to shutdown", command.worker_is_going_to_shutdown().process_id(), command.worker_is_going_to_shutdown().parent_process_id());
+            workerRegistry_->unregisterWorker(command.worker_is_going_to_shutdown().process_id());
+            break;
+        }
+
         default:
             ELOG_WARNING(logger_, COORDINATOR, "CoordinatorMessagesDispatcher: Unknown CoordinatorCommand type={}", static_cast<int>(command.type()));
             break;
diff --git a/prod/native/libcommon/code/coordinator/CoordinatorMessagesDispatcher.h b/prod/native/libcommon/code/coordinator/CoordinatorMessagesDispatcher.h
@@ -4,6 +4,7 @@
 
 #include "LoggerInterface.h"
 #include "transport/HttpTransportAsyncInterface.h"
+#include "WorkerRegistry.h"
 
 #include <memory>
 #include <span>
@@ -14,7 +15,7 @@ namespace opentelemetry::php::coordinator {
 
 class CoordinatorMessagesDispatcher {
 public:
-    CoordinatorMessagesDispatcher(std::shared_ptr<LoggerInterface> logger, std::shared_ptr<transport::HttpTransportAsyncInterface> httpTransport) : logger_(std::move(logger)), httpTransport_(std::move(httpTransport)) {
+    CoordinatorMessagesDispatcher(std::shared_ptr<LoggerInterface> logger, std::shared_ptr<transport::HttpTransportAsyncInterface> httpTransport, std::shared_ptr<WorkerRegistry> workerRegistry) : logger_(std::move(logger)), httpTransport_(std::move(httpTransport)), workerRegistry_(std::move(workerRegistry)) {
     }
 
     ~CoordinatorMessagesDispatcher() = default;
@@ -24,6 +25,7 @@ class CoordinatorMessagesDispatcher {
 private:
     std::shared_ptr<LoggerInterface> logger_;
     std::shared_ptr<transport::HttpTransportAsyncInterface> httpTransport_;
+    std::shared_ptr<WorkerRegistry> workerRegistry_;
 };
 
 }
diff --git a/prod/native/libcommon/code/coordinator/CoordinatorProcess.cpp b/prod/native/libcommon/code/coordinator/CoordinatorProcess.cpp
@@ -13,7 +13,7 @@ void CoordinatorProcess::coordinatorLoop() {
         unsigned int priority = 0;
 
         try {
-            if (commandQueue_->timed_receive(buffer, maxMqPayloadSize, receivedSize, priority, std::chrono::steady_clock::now() + std::chrono::milliseconds(10))) {
+            if (commandQueue_->timed_receive(buffer, maxMqPayloadSize, receivedSize, priority, std::chrono::steady_clock::now() + std::chrono::milliseconds(100))) {
                 processor_.processReceivedChunk(reinterpret_cast<const CoordinatorPayload *>(buffer), receivedSize);
             }
         } catch (std::exception &ex) {
@@ -28,10 +28,16 @@ void CoordinatorProcess::setupPeriodicTasks() {
     periodicTaskExecutor_ = std::make_unique<PeriodicTaskExecutor>(std::vector<PeriodicTaskExecutor::task_t>{[this](PeriodicTaskExecutor::time_point_t now) {
         // Check parent process is alive
         if (getppid() != parentProcessId_) {
-            ELOG_DEBUG(logger_, COORDINATOR, "CoordinatorProcess: parent process has exited, shutting down coordinator process");
-            working_ = false;
+            ELOG_DEBUG(logger_, COORDINATOR, "CoordinatorProcess: parent process has exited. Checking if workers are still alive.");
+            workerRegistry_->verifyWorkersAlive();
+            if (workerRegistry_->getWorkerCount() > 0) {
+                ELOG_DEBUG(logger_, COORDINATOR, "CoordinatorProcess: there are still {} alive workers, continuing work", workerRegistry_->getWorkerCount());
+            } else {
+                working_ = false;
+            }
         }
 
+
         static auto lastCleanupTime = std::chrono::steady_clock::now();
         if (now - lastCleanupTime >= cleanUpLostMessagesInterval) {
             processor_.cleanupAbandonedMessages(now, std::chrono::seconds(10));
diff --git a/prod/native/libcommon/code/coordinator/CoordinatorProcess.h b/prod/native/libcommon/code/coordinator/CoordinatorProcess.h
@@ -12,11 +12,12 @@
 
 #include <atomic>
 #include <chrono>
-#include <functional>
 #include <memory>
 #include <string>
 
 #include "CoordinatorConfigurationProvider.h"
+#include "coordinator/WorkerRegistrar.h"
+#include "coordinator/WorkerRegistry.h"
 
 namespace opentelemetry::php::coordinator {
 
@@ -29,17 +30,19 @@ constexpr static std::chrono::minutes cleanUpLostMessagesInterval(1);
 class CoordinatorProcess : public boost::noncopyable, public ForkableInterface {
 
 public:
-    CoordinatorProcess(std::shared_ptr<LoggerInterface> logger, std::shared_ptr<CoordinatorMessagesDispatcher> messagesDispatcher, std::shared_ptr<CoordinatorConfigurationProvider> configProvider) : logger_(std::move(logger)), messagesDispatcher_(std::move(messagesDispatcher)), configProvider_(std::move(configProvider)) {
+    CoordinatorProcess(std::shared_ptr<LoggerInterface> logger, std::shared_ptr<CoordinatorMessagesDispatcher> messagesDispatcher, std::shared_ptr<CoordinatorConfigurationProvider> configProvider, std::shared_ptr<WorkerRegistry> workerRegistry) : logger_(std::move(logger)), messagesDispatcher_(std::move(messagesDispatcher)), configProvider_(std::move(configProvider)), workerRegistry_(std::move(workerRegistry)) {
     }
     ~CoordinatorProcess() {
     }
 
     void prefork() final {
         periodicTaskExecutor_->prefork();
+        workerRegistrar_->prefork();
     }
 
     void postfork([[maybe_unused]] bool child) final {
         periodicTaskExecutor_->postfork(child);
+        workerRegistrar_->postfork(child);
     }
 
     // returns true in scope of forked CoordinatorProcess
@@ -83,6 +86,8 @@ class CoordinatorProcess : public boost::noncopyable, public ForkableInterface {
     CoordinatorTelemetrySignalsSender coordinatorSender_{logger_, [this](const std::string &payload) { return processor_.sendPayload(payload); }};
     std::shared_ptr<CoordinatorMessagesDispatcher> messagesDispatcher_;
     std::shared_ptr<CoordinatorConfigurationProvider> configProvider_;
+    std::unique_ptr<WorkerRegistrar> workerRegistrar_{std::make_unique<WorkerRegistrar>(logger_, [this](const std::string &payload) { return processor_.sendPayload(payload); })};
+    std::shared_ptr<WorkerRegistry> workerRegistry_;
 
     int processId_ = 0;
     int parentProcessId_ = 0;
diff --git a/prod/native/libcommon/code/coordinator/CoordinatorTelemetrySignalsSender.h b/prod/native/libcommon/code/coordinator/CoordinatorTelemetrySignalsSender.h
@@ -20,9 +20,9 @@ class CoordinatorTelemetrySignalsSender : public transport::HttpTransportAsyncIn
 
     ~CoordinatorTelemetrySignalsSender() = default;
 
-    void initializeConnection(std::string endpointUrl, std::size_t endpointHash, std::string contentType, enpointHeaders_t const &endpointHeaders, std::chrono::milliseconds timeout, std::size_t maxRetries, std::chrono::milliseconds retryDelay, opentelemetry::php::transport::HttpEndpointSSLOptions sslOptions);
-    void enqueue(std::size_t endpointHash, std::span<std::byte> payload, responseCallback_t callback = {});
-    void updateRetryDelay(size_t endpointHash, std::chrono::milliseconds retryDelay) {
+    void initializeConnection(std::string endpointUrl, std::size_t endpointHash, std::string contentType, enpointHeaders_t const &endpointHeaders, std::chrono::milliseconds timeout, std::size_t maxRetries, std::chrono::milliseconds retryDelay, opentelemetry::php::transport::HttpEndpointSSLOptions sslOptions) override;
+    void enqueue(std::size_t endpointHash, std::span<std::byte> payload, responseCallback_t callback = {}) override;
+    void updateRetryDelay(size_t endpointHash, std::chrono::milliseconds retryDelay) override {
     }
 
 private:
diff --git a/prod/native/libcommon/code/coordinator/WorkerRegistrar.cpp b/prod/native/libcommon/code/coordinator/WorkerRegistrar.cpp
@@ -0,0 +1,40 @@
+#include "WorkerRegistrar.h"
+
+#include "os/OsUtils.h"
+#include "coordinator/proto/CoordinatorCommands.pb.h"
+
+namespace opentelemetry::php::coordinator {
+
+void WorkerRegistrar::registerWorker() {
+    coordinator::WorkerStartedCommand command;
+    command.set_process_id(osutils::getCurrentProcessId());
+    command.set_parent_process_id(osutils::getParentProcessId());
+
+    coordinator::CoordinatorCommand coordCommand;
+    coordCommand.set_type(coordinator::CoordinatorCommand::WORKER_STARTED);
+    *coordCommand.mutable_worker_started() = command;
+
+    if (!sendPayload_(coordCommand.SerializeAsString())) {
+        ELOG_DEBUG(logger_, COORDINATOR, "WorkerRegistrar: failed to send worker registration message to coordinator process");
+    } else {
+        ELOG_DEBUG(logger_, COORDINATOR, "WorkerRegistrar: sent worker registration message to coordinator process");
+    }
+}
+
+void WorkerRegistrar::unregisterWorker() {
+    coordinator::WorkerIsGoingToShutdownCommand command;
+    command.set_process_id(osutils::getCurrentProcessId());
+    command.set_parent_process_id(osutils::getParentProcessId());
+
+    coordinator::CoordinatorCommand coordCommand;
+    coordCommand.set_type(coordinator::CoordinatorCommand::WORKER_IS_GOING_TO_SHUTDOWN);
+    *coordCommand.mutable_worker_is_going_to_shutdown() = command;
+
+    if (!sendPayload_(coordCommand.SerializeAsString())) {
+        ELOG_DEBUG(logger_, COORDINATOR, "WorkerRegistrar: failed to send worker unregister message");
+    } else {
+        ELOG_DEBUG(logger_, COORDINATOR, "WorkerRegistrar: sent worker unregistration message to coordinator process");
+    }
+}
+
+}
diff --git a/prod/native/libcommon/code/coordinator/WorkerRegistrar.h b/prod/native/libcommon/code/coordinator/WorkerRegistrar.h
@@ -0,0 +1,40 @@
+#pragma once
+
+
+#include "LoggerInterface.h"
+#include "ForkableInterface.h"
+#include <functional>
+#include <memory>
+
+namespace opentelemetry::php::coordinator {
+
+class WorkerRegistrar : public ForkableInterface {
+public:
+    using sendPayload_t = std::function<bool(std::string const &payload)>;
+
+    WorkerRegistrar(std::shared_ptr<LoggerInterface> logger, sendPayload_t sendPayload) : logger_(std::move(logger)), sendPayload_(std::move(sendPayload)) {
+    }
+
+    ~WorkerRegistrar() {
+        unregisterWorker();
+    }
+
+    void prefork() final {
+    }
+
+    void postfork([[maybe_unused]] bool child) final {
+        if (child) {
+            registerWorker();
+        }
+    }
+
+private:
+    void registerWorker();
+    void unregisterWorker();
+
+    std::shared_ptr<LoggerInterface> logger_;
+    sendPayload_t sendPayload_;
+
+};
+
+}
diff --git a/prod/native/libcommon/code/coordinator/WorkerRegistry.h b/prod/native/libcommon/code/coordinator/WorkerRegistry.h
@@ -0,0 +1,54 @@
+#pragma once
+
+#include "LoggerInterface.h"
+#include <cstddef>
+#include <memory>
+#include <unordered_set>
+#include <sys/types.h>
+#include <signal.h>
+#include <errno.h>
+
+namespace opentelemetry::php::coordinator {
+
+class WorkerRegistry {
+public:
+    WorkerRegistry(std::shared_ptr<LoggerInterface> logger) : logger_(std::move(logger)) {
+    }
+
+    void registerWorker(pid_t processId, pid_t parentProcessId) {
+        ELOG_DEBUG(logger_, COORDINATOR, "WorkerRegistry: registering worker with process id {} and parent process id {}", processId, parentProcessId);
+        workers_.emplace(processId);
+    }
+
+    void unregisterWorker(pid_t processId) {
+        ELOG_DEBUG(logger_, COORDINATOR, "WorkerRegistry: removing worker with process id {}", processId);
+        workers_.erase(processId);
+    }
+
+    bool hasWorker(pid_t processId) const {
+        return workers_.find(processId) != workers_.end();
+    }
+
+    void verifyWorkersAlive() {
+        for (auto it = workers_.begin(); it != workers_.end();) {
+            pid_t pid = *it;
+            if (kill(pid, 0) == -1 && errno == ESRCH) {
+                ELOG_DEBUG(logger_, COORDINATOR, "WorkerRegistry: worker with process id {} is not alive, removing from registry", pid);
+                it = workers_.erase(it);
+            } else {
+                ++it;
+            }
+        }
+    }
+
+    std::size_t getWorkerCount() const {
+        return workers_.size();
+    }
+
+private:
+    std::shared_ptr<LoggerInterface> logger_;
+    std::unordered_set<pid_t> workers_;
+
+};
+
+}
diff --git a/prod/native/libphpbridge/code/PhpBridge.cpp b/prod/native/libphpbridge/code/PhpBridge.cpp
@@ -54,6 +54,9 @@ std::string_view PhpBridge::getPhpSapiName() const {
 }
 
 void PhpBridge::compileAndExecuteFile(std::string_view fileName) const {
+    if (fileName.empty()) {
+        throw std::invalid_argument("File name provided for compilation and execution cannot be empty");
+    }
 
 #if PHP_VERSION_ID < 80100
     AutoZval fn{fileName};
