Merge pull request #10 from vrza/config-process

Run ConfigurationSource in a separate process

Author: vrza

README.adoc

@@ -43,7 +43,7 @@ Currently LRPM is not intended to run as PID 1, nor was it tested in this role.
 == Usage
 
 1. Implement the `ConfigurationSource`. This interface exposes a public method `loadConfiguration()`, that returns an associative array containing the configuration for all the workers (see details below).
-2. Implement the `Worker`. This interface exposes two public methods, `start()` and `cycle()`. The child process will call `start()`, passing the worker configuration as an argument, and then enter an endless loop, where it calls `cycle()`, and then checks for a shutdown condition (usually, the death of the LRPM supervisor process). Backoff should be implemented by the cycle() method -- be careful not to run a tight loop at times when cycles are not doing actual work!
+2. Implement the `Worker`. This interface exposes two public methods, `start()` and `cycle()`. The child process will call `start()`, passing the worker configuration as an argument, and then enter an endless loop, where it calls `cycle()`, and then checks for a shutdown condition (usually, the termination of the LRPM supervisor process). Backoff should be implemented at the end of the `cycle()` method, as we dispatch signal handlers right after `cycle()` returns. Be careful not to run a tight loop at times when cycles are not doing actual work!
 
 === Job configuration
 
@@ -110,17 +110,15 @@ Default value: 10
 
 === Signal handling
 
-LRPM supervisor process installs signal handlers for SIGCHLD, SIGTERM and SIGINT.
+LRPM supervisor process installs signal handlers for SIGCHLD, SIGTERM, SIGINT, SIGHUP and SIGUSR1.
 
-You can implement and install your own signal handlers inside your Worker implementation, but make sure that your Worker process shuts down cleanly after receiving SIGTERM, otherwise LRPM will consider it unresponsive and follow up with a SIGKILL.
+Worker process installs default signal handlers for SIGTERM and SIGINT. Signal handlers are dispatched between loop cycles, and these default handlers will terminate the Worker.
 
-Default SIGTERM and SIGINT handlers will terminate the Worker before the next loop cycle.
+You can implement and install your own signal handlers inside your Worker implementation, but make sure that your Worker process shuts down cleanly after receiving SIGTERM, otherwise the LRPM supervisor will consider it unresponsive and follow up with a SIGKILL.
 
 === Misc caveats
 
-Be aware that code in your `ConfigurationSource` class will run in the supervisor (parent) process, while your `Worker` classes will run in child processes.
-
-Sharing open sockets between parent and children through `fork(2)` is not safe. Worker processes should connect to wherever they need to connect to only after they have been spawned. For example, opening sockets in `ConfigurationSource`, in the parent process, passing them via references in the configuration associative array, and then using them in Worker processes is inherently unsafe.
+Be aware that code in your entry point will run in the supervisor (parent) process, while your `Worker` classes will run in child processes. The entry point should do no more than set up the autoloader and run the `ProcessManager`. Sharing open sockets between parent and children through `fork(2)` is not safe. Worker processes should connect to wherever they need to connect to only after they have been spawned.
 
 == Operating LRPM
 
@@ -146,11 +144,11 @@ opcache.file_cache_only=0
 
 PHP-LRPM keeps metadata in an associative array. For efficient lookups by PID, a separate index is maintained.
 
-This functionality was offloaded to a generic library https://github.com/vrza/array-with-secondary-keys[Array with Secondary Keys], that wraps a hash map and maintains secondary indexes (similar to how secondary keys in an SQL database work). Implementing this particular collection lead to creation of https://github.com/vrza/cardinal-collections[Cardinal Collections], a PHP toolkit for building collections.
+This functionality was offloaded to a generic library https://github.com/vrza/array-with-secondary-keys[Array with Secondary Keys], that wraps a hash map and maintains secondary indexes (similar to how secondary keys in an SQL database work). Implementing this particular collection lead to the creation of https://github.com/vrza/cardinal-collections[Cardinal Collections], a PHP toolkit for building collections.
 
 ==== Implement receiving, handling and responding to control messages
 
-Included is the `lrpmctl` tool, which uses the https://github.com/vrza/php-tipc[tipc] library to exchange messages with a running instance of LRPM. Some examples of messages include getting the `status` of running processes (see screenshot above), and requesting a `restart` of a process.
+Included is the `lrpmctl` tool, which uses the https://github.com/vrza/php-tipc[tipc] library to exchange messages with a running instance of LRPM over a Unix domain socket connection. Some examples of messages include getting the `status` of all workers (see screenshot above), and requesting a `restart` of a worker process.
 
 ==== Make sure unresponsive processes get terminated
 
@@ -160,6 +158,10 @@ Wait for children to terminate after sending SIGTERM, follow up with SIGKILL if
 
 Implemented blocking shutdown loop that makes sure all children are terminated on shutdown, including processes that may be unresponsive.
 
+==== Configuration process
+
+Made `ConfigurationSource` run in a process separate from the supervisor. This is to prevent `Worker` processes inheriting sockets opened by `ConfigurationSource` code (e.g. persistent database connections). The supervisor process and the config process are using the tipc library to exchange messages over a Unix domain socket connection.
+
 == Some name ideas that were considered
 
 * Palermo

README.in.adoc

@@ -43,7 +43,7 @@ Currently LRPM is not intended to run as PID 1, nor was it tested in this role.
 == Usage
 
 1. Implement the `ConfigurationSource`. This interface exposes a public method `loadConfiguration()`, that returns an associative array containing the configuration for all the workers (see details below).
-2. Implement the `Worker`. This interface exposes two public methods, `start()` and `cycle()`. The child process will call `start()`, passing the worker configuration as an argument, and then enter an endless loop, where it calls `cycle()`, and then checks for a shutdown condition (usually, the death of the LRPM supervisor process). Backoff should be implemented by the cycle() method -- be careful not to run a tight loop at times when cycles are not doing actual work!
+2. Implement the `Worker`. This interface exposes two public methods, `start()` and `cycle()`. The child process will call `start()`, passing the worker configuration as an argument, and then enter an endless loop, where it calls `cycle()`, and then checks for a shutdown condition (usually, the termination of the LRPM supervisor process). Backoff should be implemented at the end of the `cycle()` method, as we dispatch signal handlers right after `cycle()` returns. Be careful not to run a tight loop at times when cycles are not doing actual work!
 
 === Job configuration
 
@@ -73,17 +73,15 @@ See `example.php` for a full running example with more details.
 
 === Signal handling
 
-LRPM supervisor process installs signal handlers for SIGCHLD, SIGTERM and SIGINT.
+LRPM supervisor process installs signal handlers for SIGCHLD, SIGTERM, SIGINT, SIGHUP and SIGUSR1.
 
-You can implement and install your own signal handlers inside your Worker implementation, but make sure that your Worker process shuts down cleanly after receiving SIGTERM, otherwise LRPM will consider it unresponsive and follow up with a SIGKILL.
+Worker process installs default signal handlers for SIGTERM and SIGINT. Signal handlers are dispatched between loop cycles, and these default handlers will terminate the Worker.
 
-Default SIGTERM and SIGINT handlers will terminate the Worker before the next loop cycle.
+You can implement and install your own signal handlers inside your Worker implementation, but make sure that your Worker process shuts down cleanly after receiving SIGTERM, otherwise the LRPM supervisor will consider it unresponsive and follow up with a SIGKILL.
 
 === Misc caveats
 
-Be aware that code in your `ConfigurationSource` class will run in the supervisor (parent) process, while your `Worker` classes will run in child processes.
-
-Sharing open sockets between parent and children through `fork(2)` is not safe. Worker processes should connect to wherever they need to connect to only after they have been spawned. For example, opening sockets in `ConfigurationSource`, in the parent process, passing them via references in the configuration associative array, and then using them in Worker processes is inherently unsafe.
+Be aware that code in your entry point will run in the supervisor (parent) process, while your `Worker` classes will run in child processes. The entry point should do no more than set up the autoloader and run the `ProcessManager`. Sharing open sockets between parent and children through `fork(2)` is not safe. Worker processes should connect to wherever they need to connect to only after they have been spawned.
 
 == Operating LRPM
 
@@ -109,11 +107,11 @@ opcache.file_cache_only=0
 
 PHP-LRPM keeps metadata in an associative array. For efficient lookups by PID, a separate index is maintained.
 
-This functionality was offloaded to a generic library https://github.com/vrza/array-with-secondary-keys[Array with Secondary Keys], that wraps a hash map and maintains secondary indexes (similar to how secondary keys in an SQL database work). Implementing this particular collection lead to creation of https://github.com/vrza/cardinal-collections[Cardinal Collections], a PHP toolkit for building collections.
+This functionality was offloaded to a generic library https://github.com/vrza/array-with-secondary-keys[Array with Secondary Keys], that wraps a hash map and maintains secondary indexes (similar to how secondary keys in an SQL database work). Implementing this particular collection lead to the creation of https://github.com/vrza/cardinal-collections[Cardinal Collections], a PHP toolkit for building collections.
 
 ==== Implement receiving, handling and responding to control messages
 
-Included is the `lrpmctl` tool, which uses the https://github.com/vrza/php-tipc[tipc] library to exchange messages with a running instance of LRPM. Some examples of messages include getting the `status` of running processes (see screenshot above), and requesting a `restart` of a process.
+Included is the `lrpmctl` tool, which uses the https://github.com/vrza/php-tipc[tipc] library to exchange messages with a running instance of LRPM over a Unix domain socket connection. Some examples of messages include getting the `status` of all workers (see screenshot above), and requesting a `restart` of a worker process.
 
 ==== Make sure unresponsive processes get terminated
 
@@ -123,6 +121,10 @@ Wait for children to terminate after sending SIGTERM, follow up with SIGKILL if
 
 Implemented blocking shutdown loop that makes sure all children are terminated on shutdown, including processes that may be unresponsive.
 
+==== Configuration process
+
+Made `ConfigurationSource` run in a process separate from the supervisor. This is to prevent `Worker` processes inheriting sockets opened by `ConfigurationSource` code (e.g. persistent database connections). The supervisor process and the config process are using the tipc library to exchange messages over a Unix domain socket connection.
+
 == Some name ideas that were considered
 
 * Palermo

bin/lrpmctl

@@ -56,8 +56,12 @@ probeForAutoloader($progname);
 use TIPC\UnixSocketStreamClient;
 use TextTableFormatter\Table;
 use TerminalPalette\AnsiSeq;
+use PHPLRPM\IPCUtilities;
 
-$socket = findLrpmUnixSocket();
+$socket = UnixSocketStreamClient::findSocketPath('control', IPCUtilities::getSocketDirs());
+if (is_null($socket)) {
+    exitNoConnection();
+}
 $client = new UnixSocketStreamClient($socket);
 if ($client->connect() === false) {
     exitNoConnection();
@@ -91,32 +95,6 @@ function probeForAutoloader($programName): void
     exit(EXIT_AUTOLOADER_NOT_FOUND);
 }
 
-function findLrpmUnixSocket()
-{
-    $socketDirs = [
-        '/run/php-lrpm',
-        '/run/user/' . posix_geteuid() . '/php-lrpm'
-    ];
-    $socketFileName = 'control';
-    foreach ($socketDirs as $dir) {
-        $candidate = $dir . '/' . $socketFileName;
-        if (is_writable($candidate)) {
-            return $candidate;
-        }
-    }
-    fwrite(STDERR, "Could not find lrpm Unix domain socket" . PHP_EOL);
-    fwrite(
-        STDERR,
-        "Tried: " . implode(
-            ', ',
-            array_map(function ($dir) use ($socketFileName) {
-                return $dir . '/' . $socketFileName;
-            }, $socketDirs)
-        ) . PHP_EOL
-    );
-    exitNoConnection();
-}
-
 function exitNoConnection(): void
 {
     fwrite(STDERR, 'Could not connect to lrpm' . PHP_EOL);

composer.json

@@ -33,11 +33,12 @@
         "ext-json": "*",
         "ext-pcntl": "*",
         "ext-posix": "*",
+        "ext-sockets": "*",
         "vrza/array-with-secondary-keys": "dev-main",
         "vrza/cardinal-collections": "dev-main",
         "vrza/php-tipc": "dev-main",
-        "vrza/text-table-formatter": "dev-main",
-        "vrza/terminal-palette": "dev-main"
+        "vrza/terminal-palette": "dev-main",
+        "vrza/text-table-formatter": "dev-main"
     },
     "autoload": {
         "psr-4": {

example.php

@@ -5,6 +5,5 @@
 use PHPLRPM\Test\MockConfigurationSource;
 use PHPLRPM\ProcessManager;
 
-$configurationSource = new MockConfigurationSource();
-$processManager = new ProcessManager($configurationSource);
+$processManager = new ProcessManager(MockConfigurationSource::class);
 $processManager->run();

src/ConfigurationClient.php

@@ -0,0 +1,55 @@
+<?php
+
+namespace PHPLRPM;
+
+use TIPC\UnixSocketStreamClient;
+
+class ConfigurationClient
+{
+    private $configSocket;
+
+    public function findConfigSocket()
+    {
+        $this->configSocket = UnixSocketStreamClient::findSocketPath(
+            ConfigurationService::SOCKET_FILE_NAME,
+            IPCUtilities::getSocketDirs()
+        );
+        if (is_null($this->configSocket)) {
+            throw new RuntimeException("Supervisor could not find config process Unix domain socket");
+        }
+    }
+
+    /**
+     * Creates a configuration client and contacts the configuration
+     * process with a request to poll for latest configuration.
+     *
+     * @return array containing latest configuration
+     * @throws ConfigurationPollException
+     */
+    public function pollConfiguration($signalHandlers): array
+    {
+        $recvBufSize = 8 * 1024 * 1024;
+        $client = new UnixSocketStreamClient($this->configSocket, $recvBufSize);
+        if ($client->connect() === false) {
+            throw new ConfigurationPollException("Could not connect to socket {$this->configSocket}");
+        }
+        if ($client->sendMessage(ConfigurationService::REQ_POLL_CONFIG_SOURCE) === false) {
+            $client->disconnect();
+            throw new ConfigurationPollException("Could not send config query message over socket {$this->configSocket}");
+        }
+        $signals = array_keys($signalHandlers);
+        pcntl_sigprocmask(SIG_BLOCK, $signals);
+        if (empty($response = $client->receiveMessage())) {
+            pcntl_sigprocmask(SIG_UNBLOCK, $signals);
+            $client->disconnect();
+            throw new ConfigurationPollException("Failed to read config response from {$this->configSocket}");
+        }
+        pcntl_sigprocmask(SIG_UNBLOCK, $signals);
+        $client->disconnect();
+        if ($response === ConfigurationService::RESP_ERROR_CONFIG_SOURCE) {
+            throw new ConfigurationPollException("Config process at {$this->configSocket} responded with an error message");
+        }
+        return Serialization::deserialize($response);
+    }
+
+}

src/ConfigurationPollException.php

@@ -0,0 +1,10 @@
+<?php
+
+namespace PHPLRPM;
+
+use Exception;
+
+class ConfigurationPollException extends Exception
+{
+
+}

src/ConfigurationProcessManager.php

@@ -0,0 +1,79 @@
+<?php
+
+namespace PHPLRPM;
+
+class ConfigurationProcessManager
+{
+    private const CONFIG_PROCESS_MAX_BACKOFF_SECONDS = 300;
+    private const CONFIG_PROCESS_MAX_RETRIES = 5;
+    private const CONFIG_PROCESS_MIN_RUN_TIME_SECONDS = 40;
+    private const CONFIG_PROCESS_TERM_TIMEOUT_SECONDS = 5;
+
+    private $configProcessId;
+    private $configProcessRetries = 0;
+    private $configProcessLastStart = 0;
+    private $configProcessRestartAt = 0;
+
+    public function getPID(): ?int
+    {
+        return $this->configProcessId;
+    }
+
+    public function setPID($pid): void
+    {
+        $this->configProcessId = $pid;
+    }
+
+    public function handleTerminatedConfigProcess(): void
+    {
+        $this->configProcessId = null;
+        $this->scheduleRestartWithBackoff();
+    }
+
+    private function scheduleRestartWithBackoff()
+    {
+        $now = time();
+        if ($now - $this->configProcessLastStart >= self::CONFIG_PROCESS_MIN_RUN_TIME_SECONDS) {
+            $this->configProcessRetries = 0;
+        } else {
+            $backoff = min(2 ** $this->configProcessRetries, self::CONFIG_PROCESS_MAX_BACKOFF_SECONDS);
+            fwrite(STDERR, "==> Backing off on config process spawn (retry: " . $this->configProcessRetries . ", seconds: $backoff)" . PHP_EOL);
+            $this->configProcessRestartAt = $now + $backoff;
+            $this->configProcessRetries++;
+        }
+    }
+
+    public function shouldRetryStartingConfigProcess(): ?bool
+    {
+        if (!is_null($this->configProcessId)) {
+            return false;
+        }
+        if ($this->configProcessRetries > self::CONFIG_PROCESS_MAX_RETRIES) {
+            fwrite(STDERR, '==> Config process failed after ' . self::CONFIG_PROCESS_MAX_RETRIES . ' retries, giving up' . PHP_EOL);
+            return null;
+        }
+        $now = time();
+        if ($this->configProcessRestartAt < $now) {
+            $this->configProcessLastStart = $now;
+            return true;
+        }
+        return false;
+    }
+
+    public function stopConfigurationProcess(): void
+    {
+        if (is_null($this->configProcessId)) {
+            return;
+        }
+        posix_kill($this->configProcessId, SIGTERM);
+        $remaining = self::CONFIG_PROCESS_TERM_TIMEOUT_SECONDS;
+        while (!is_null($this->configProcessId) && $remaining > 0) {
+            $remaining = sleep($remaining);
+            pcntl_signal_dispatch();
+        }
+        if (!is_null($this->configProcessId)) {
+            posix_kill($this->configProcessId, SIGKILL);
+        }
+    }
+
+}