Implement Limiting Max Child Processes Via an Exec Env Var API

src/Environment/Memory.php

@@ -0,0 +1,147 @@
+<?php
+declare(strict_types=1);
+
+namespace Neighborhoods\Kojo\Environment;
+
+use InvalidArgumentException;
+use LogicException;
+
+class Memory implements MemoryInterface
+{
+    protected $EnvironmentMaximumMemoryValue;
+    protected $EnvironmentMaximumMemoryInputUnit;
+    protected $WorkerMaximumMemoryValue;
+    protected $WorkerMaximumMemoryInputUnit;
+    protected $MaximumNumberOfWorkers;
+
+    public function getMaximumNumberOfWorkers(): int
+    {
+        if ($this->MaximumNumberOfWorkers === null) {
+            $environmentMaximumMemoryValue = $this->getEnvironmentMaximumMemoryValue();
+            $workerMaximumMemoryValue = $this->getWorkerMaximumMemoryValue();
+            $maximumNumberOfWorkers = (int)floor($environmentMaximumMemoryValue / $workerMaximumMemoryValue);
+            $this->MaximumNumberOfWorkers = $maximumNumberOfWorkers;
+        }
+
+        return $this->MaximumNumberOfWorkers;
+    }
+
+    public function setEnvironmentMaximumMemoryValue(int $EnvironmentMaximumMemoryValue): MemoryInterface
+    {
+        if ($this->EnvironmentMaximumMemoryValue !== null) {
+            throw new LogicException('Environment Maximum Memory Value is already set.');
+        }
+
+        $this->EnvironmentMaximumMemoryValue = $this->getBytes(
+            $EnvironmentMaximumMemoryValue,
+            $this->getEnvironmentMaximumMemoryInputUnit()
+        );
+
+        return $this;
+    }
+
+    protected function getEnvironmentMaximumMemoryValue(): int
+    {
+        if ($this->EnvironmentMaximumMemoryValue === null) {
+            throw new LogicException('Environment Maximum Memory Value has not been set.');
+        }
+
+        return $this->EnvironmentMaximumMemoryValue;
+    }
+
+    public function setEnvironmentMaximumMemoryInputUnit(string $EnvironmentMaximumMemoryUnit): MemoryInterface
+    {
+        if ($this->EnvironmentMaximumMemoryInputUnit !== null) {
+            throw new LogicException('Environment Maximum Memory Input Unit is already set.');
+        }
+
+        $this->assertValidMemoryUnit($EnvironmentMaximumMemoryUnit);
+        $this->EnvironmentMaximumMemoryInputUnit = $EnvironmentMaximumMemoryUnit;
+
+        return $this;
+    }
+
+    protected function getEnvironmentMaximumMemoryInputUnit(): string
+    {
+        if ($this->EnvironmentMaximumMemoryInputUnit === null) {
+            throw new LogicException('Environment Maximum Memory Input Unit has not been set.');
+        }
+
+        return $this->EnvironmentMaximumMemoryInputUnit;
+    }
+
+    public function setWorkerMaximumMemoryValue(int $WorkerMaximumMemoryValue): MemoryInterface
+    {
+        if ($this->WorkerMaximumMemoryValue !== null) {
+            throw new LogicException('Worker Maximum Memory Value is already set.');
+        }
+
+        $this->WorkerMaximumMemoryValue = $this->getBytes(
+            $WorkerMaximumMemoryValue,
+            $this->getWorkerMaximumMemoryInputUnit()
+        );
+
+        return $this;
+    }
+
+    public function getWorkerMaximumMemoryValue(): int
+    {
+        if ($this->WorkerMaximumMemoryValue === null) {
+            throw new LogicException('Worker Maximum Memory Value has not been set.');
+        }
+
+        return $this->WorkerMaximumMemoryValue;
+    }
+
+    public function setWorkerMaximumMemoryInputUnit(string $WorkerMaximumMemoryUnit): MemoryInterface
+    {
+        if ($this->WorkerMaximumMemoryInputUnit !== null) {
+            throw new LogicException('Worker Maximum Memory Input Unit is already set.');
+        }
+
+        $this->assertValidMemoryUnit($WorkerMaximumMemoryUnit);
+
+        $this->WorkerMaximumMemoryInputUnit = $WorkerMaximumMemoryUnit;
+
+        return $this;
+    }
+
+    protected function getWorkerMaximumMemoryInputUnit(): string
+    {
+        if ($this->WorkerMaximumMemoryInputUnit === null) {
+            throw new LogicException('Worker Maximum Memory Input Unit has not been set.');
+        }
+
+        return $this->WorkerMaximumMemoryInputUnit;
+    }
+
+    protected function getBytes(int $Value, string $Unit): int
+    {
+        switch ($Unit) {
+            /** @noinspection PhpMissingBreakStatementInspection */
+            case self::Gibibyte:
+                $Value *= 1024;
+            /** @noinspection PhpMissingBreakStatementInspection */
+            case self::Mebibyte:
+                $Value *= 1024;
+            case self::Kibibyte:
+                $Value *= 1024;
+        }
+
+        return $Value;
+    }
+
+    protected function assertValidMemoryUnit(string $MemoryUnit): MemoryInterface
+    {
+        switch ($MemoryUnit) {
+            case self::Kibibyte:
+            case self::Mebibyte:
+            case self::Gibibyte:
+                break;
+            default:
+                throw new InvalidArgumentException(sprintf('Memory Unit [%s] is not KiB, MiB, or GiB.', $MemoryUnit));
+        }
+
+        return $this;
+    }
+}

src/Environment/Memory/AwareTrait.php

@@ -0,0 +1,41 @@
+<?php
+declare(strict_types=1);
+
+namespace Neighborhoods\Kojo\Environment\Memory;
+
+use Neighborhoods\Kojo\Environment\MemoryInterface;
+
+/** @codeCoverageIgnore */
+trait AwareTrait
+{
+    protected $NeighborhoodsKojoEnvironmentMemory;
+
+    public function setEnvironmentMemory(MemoryInterface $environmentMemory): self
+    {
+        assert(!$this->hasEnvironmentMemory(),
+            new \LogicException('NeighborhoodsKojoEnvironmentMemory is already set.'));
+        $this->NeighborhoodsKojoEnvironmentMemory = $environmentMemory;
+
+        return $this;
+    }
+
+    protected function getEnvironmentMemory(): MemoryInterface
+    {
+        assert($this->hasEnvironmentMemory(), new \LogicException('NeighborhoodsKojoEnvironmentMemory is not set.'));
+
+        return $this->NeighborhoodsKojoEnvironmentMemory;
+    }
+
+    protected function hasEnvironmentMemory(): bool
+    {
+        return isset($this->NeighborhoodsKojoEnvironmentMemory);
+    }
+
+    protected function unsetEnvironmentMemory(): self
+    {
+        assert($this->hasEnvironmentMemory(), new \LogicException('NeighborhoodsKojoEnvironmentMemory is not set.'));
+        unset($this->NeighborhoodsKojoEnvironmentMemory);
+
+        return $this;
+    }
+}

src/Environment/MemoryInterface.php

@@ -0,0 +1,23 @@
+<?php
+declare(strict_types=1);
+
+namespace Neighborhoods\Kojo\Environment;
+
+interface MemoryInterface
+{
+    public const Gibibyte = 'GiB';
+    public const Mebibyte = 'MiB';
+    public const Kibibyte = 'KiB';
+
+    public function setEnvironmentMaximumMemoryValue(int $EnvironmentMaximumMemoryValue): MemoryInterface;
+
+    public function setWorkerMaximumMemoryInputUnit(string $WorkerMaximumMemoryUnit): MemoryInterface;
+
+    public function getMaximumNumberOfWorkers(): int;
+
+    public function setWorkerMaximumMemoryValue(int $WorkerMaximumMemoryValue): MemoryInterface;
+
+    public function setEnvironmentMaximumMemoryInputUnit(string $EnvironmentMaximumMemoryUnit): MemoryInterface;
+
+    public function getWorkerMaximumMemoryValue(): int;
+}

src/Environment/MemoryInterface.yml

@@ -0,0 +1,15 @@
+parameters:
+  Neighborhoods\Kojo\Environment\MemoryInterface.EnvironmentMaximumMemoryInputValue: '%env(ENVIRONMENT_MAXIMUM_MEMORY_INPUT_VALUE)%'
+  Neighborhoods\Kojo\Environment\MemoryInterface.EnvironmentMaximumMemoryInputUnit: '%env(ENVIRONMENT_MAXIMUM_MEMORY_INPUT_UNIT)%'
+  Neighborhoods\Kojo\Environment\MemoryInterface.WorkerMaximumMemoryValue: '%env(WORKER_MAXIMUM_MEMORY_INPUT_VALUE)%'
+  Neighborhoods\Kojo\Environment\MemoryInterface.WorkerMaximumMemoryUnit: '%env(WORKER_MAXIMUM_MEMORY_INPUT_UNIT)%'
+services:
+  Neighborhoods\Kojo\Environment\MemoryInterface:
+    class: Neighborhoods\Kojo\Environment\Memory
+    shared: true
+    public: false
+    calls:
+      - [setEnvironmentMaximumMemoryValue, ['%Neighborhoods\Kojo\Environment\MemoryInterface.EnvironmentMaximumMemoryInputValue%']]
+      - [setEnvironmentMaximumMemoryInputUnit, ['%Neighborhoods\Kojo\Environment\MemoryInterface.EnvironmentMaximumMemoryInputUnit%']]
+      - [setWorkerMaximumMemoryValue, ['%Neighborhoods\Kojo\Environment\MemoryInterface.WorkerMaximumMemoryValue%']]
+      - [setWorkerMaximumMemoryInputUnit, ['%Neighborhoods\Kojo\Environment\MemoryInterface.WorkerMaximumMemoryUnit%']]

src/Process/Job.php

@@ -8,6 +8,8 @@
 use Neighborhoods\Kojo\Process;
 use Neighborhoods\Kojo\Scheduler;
 use Neighborhoods\Kojo\Selector;
+use Neighborhoods\Kojo\Environment;
+use Throwable;
 
 class Job extends Forked implements JobInterface
 {
@@ -16,6 +18,7 @@ class Job extends Forked implements JobInterface
     use Scheduler\AwareTrait;
     use Selector\AwareTrait;
     use Process\Pool\Factory\AwareTrait;
+    use Environment\Memory\AwareTrait;
 
     protected function _run(): Forked
     {
@@ -26,12 +29,22 @@ protected function _run(): Forked
             $this->_getScheduler()->scheduleStaticJobs();
             $this->_getMaintainer()->updatePendingJobs();
             $this->_getMaintainer()->deleteCompletedJobs();
+            $this->applyMemoryLimit();
             $this->_getForeman()->workWorker();
-        } catch (\Throwable $throwable) {
+        } catch (Throwable $throwable) {
             $this->_getLogger()->critical($throwable->getMessage(), ['exception' => $throwable]);
             $this->_setOrReplaceExitCode(255);
         }
 
         return $this;
     }
+
+    protected function applyMemoryLimit(): JobInterface
+    {
+        $processMaximumMemoryValue = $this->getEnvironmentMemory()->getWorkerMaximumMemoryValue();
+        $this->_getLogger()->debug(sprintf('Applying memory limit of [%s] bytes.', $processMaximumMemoryValue));
+        ini_set('memory_limit', $processMaximumMemoryValue);
+
+        return $this;
+    }
 }

src/Process/Job.yml

@@ -15,5 +15,6 @@ services:
       - [setUuidMaximumInteger, [9999999999]]
       - [setProcessPoolFactory, ['@process.pool.factory-job']]
       - [setTitlePrefix, ['%process.title.prefix%']]
+      - [setEnvironmentMemory, ['@Neighborhoods\Kojo\Environment\MemoryInterface']]
   process.job:
-    alias: neighborhoods.kojo.process.job
+    alias: neighborhoods.kojo.process.job

src/Process/Pool/Factory.yml

@@ -59,4 +59,4 @@ services:
       - [setProcessPoolStrategy, ['@process.pool.strategy-job']]
       - [setProcessCollection, ['@process.collection-empty']]
   process.pool.factory-empty:
-    alias: neighborhoods.kojo.process.pool.factory-empty
+    alias: neighborhoods.kojo.process.pool.factory-empty

src/Process/Pool/Logger.yml

@@ -19,4 +19,4 @@ services:
       - [setProcessPoolLoggerMessageFactory, ['@neighborhoods.kojo.process.pool.logger.message.factory']]
       - [setLevelFilterMask, ['%neighborhoods.kojo.process.pool.logger.level_filter_mask%']]
   process.pool.logger:
-    alias: neighborhoods.kojo.process.pool.logger
+    alias: neighborhoods.kojo.process.pool.logger

src/Process/Pool/Strategy.php

@@ -35,7 +35,7 @@ protected function _listenerProcessExited(ListenerInterface $listenerProcess): S
             while (
                 $listenerProcess->hasMessages()
                 && !$this->_getProcessPool()->isFull()
-                && $this->_getProcessPool()->canEnvironmentSustainAdditionProcesses()
+                && $this->canEnvironmentSustainAdditionalProcesses()
             ) {
                 $listenerProcess->processMessages();
             }
@@ -78,7 +78,7 @@ public function currentPendingChildExitsCompleted(): StrategyInterface
     protected function _jobProcessExited(JobInterface $jobProcess): Strategy
     {
         $this->_getProcessPool()->freeChildProcess($jobProcess->getProcessId());
-        if ($jobProcess->getExitCode() !== 0 && $this->_getProcessPool()->canEnvironmentSustainAdditionProcesses()) {
+        if ($jobProcess->getExitCode() !== 0 && $this->canEnvironmentSustainAdditionalProcesses()) {
             $typeCode = $jobProcess->getTypeCode();
             $replacementProcess = $this->_getProcessCollection()->getProcessPrototypeClone($typeCode);
             $replacementProcess->setThrottle($this->getChildProcessWaitThrottle());
@@ -97,7 +97,7 @@ protected function _jobProcessExited(JobInterface $jobProcess): Strategy
 
     public function receivedAlarm(): StrategyInterface
     {
-        if (!$this->_getProcessPool()->isFull() && $this->_getProcessPool()->canEnvironmentSustainAdditionProcesses()) {
+        if (!$this->_getProcessPool()->isFull() && $this->canEnvironmentSustainAdditionalProcesses()) {
             if ($this->_hasPausedListenerProcess()) {
                 $this->_unPauseListenerProcesses();
             } else {
@@ -131,7 +131,7 @@ public function initializePool(): StrategyInterface
                 }
             }
         }
-        if ($this->_hasFillProcessTypeCode() && $this->_getProcessPool()->canEnvironmentSustainAdditionProcesses()) {
+        if ($this->_hasFillProcessTypeCode() && $this->canEnvironmentSustainAdditionalProcesses()) {
             while (!$this->_getProcessPool()->isFull()) {
                 $fillProcessTypeCode = $this->_getFillProcessTypeCode();
                 $fillProcess = $this->_getProcessCollection()->getProcessPrototypeClone($fillProcessTypeCode);

src/Process/Pool/Strategy/Server.php

@@ -52,7 +52,7 @@ public function initializePool(): StrategyInterface
                 $this->_getProcessPool()->getProcess()->exit();
             }
         }
-        if ($this->_hasFillProcessTypeCode() && $this->_getProcessPool()->canEnvironmentSustainAdditionProcesses()) {
+        if ($this->_hasFillProcessTypeCode() && $this->canEnvironmentSustainAdditionalProcesses()) {
             while (!$this->_getProcessPool()->isFull()) {
                 $fillProcessTypeCode = $this->_getFillProcessTypeCode();
                 $fillProcess = $this->_getProcessCollection()->getProcessPrototypeClone($fillProcessTypeCode);