KOJO-3 KOJO-2 | bugfixes etc.

- update example worker
- added environmental parameters to be set by client
- set alarm to 5s
- set max child processes to 10
- refactored isFull to (isFull and canEnvironmentSustainAdditionProcesses) for testing the state of the environment since that state can (and likely will) change between samples so it causes an unsafe test when combined with addChildProcess
- updated tests

example/Environment/Parameters.yml

@@ -0,0 +1,8 @@
+parameters:
+  neighborhoods.kojo.environment.parameters.redis_port: '%env(REDIS_PORT)%'
+  neighborhoods.kojo.environment.parameters.redis_host: '%env(REDIS_HOST)%'
+  neighborhoods.kojo.environment.parameters.database_user_name: '%env(DATABASE_USERNAME)%'
+  neighborhoods.kojo.environment.parameters.database_password: '%env(DATABASE_PASSWORD)%'
+  neighborhoods.kojo.environment.parameters.database_adapter: '%env(DATABASE_ADAPTER)%'
+  neighborhoods.kojo.environment.parameters.database_host: '%env(DATABASE_HOST)%'
+  neighborhoods.kojo.environment.parameters.database_name: '%env(DATABASE_NAME)%'

example/Worker.php

@@ -13,6 +13,11 @@ class Worker
 
     public function work()
     {
+//        $newJobScheduler = $this->_getApiV1WorkerService()->getNewJobScheduler();
+//        $newJobScheduler->setJobTypeCode('type_code_1')
+//                        ->setWorkAtDateTime(new \DateTime('now'))
+//                        ->save()
+//                        ->getJobId();
         $this->_getApiV1WorkerService()->requestCompleteSuccess()->applyRequest();
 
         return $this;

src/Db/PDO/Builder.yml

@@ -4,9 +4,9 @@ services:
     public: false
     shared: false
     calls:
-      - [setPassword, ['%env(DATABASE_PASSWORD)%']]
-      - [setUserName, ['%env(DATABASE_USERNAME)%']]
-      - [setDataSourceName, ['%env(DATABASE_ADAPTER)%:dbname=%env(DATABASE_NAME)%;host=%env(DATABASE_HOST)%']]
+      - [setPassword, ['%neighborhoods.kojo.environment.parameters.database_password%']]
+      - [setUserName, ['%neighborhoods.kojo.environment.parameters.database_user_name%']]
+      - [setDataSourceName, ['%neighborhoods.kojo.environment.parameters.database_adapter%:dbname=%neighborhoods.kojo.environment.parameters.database_name%;host=%neighborhoods.kojo.environment.parameters.database_host%']]
   db.pdo.builder:
     alias: neighborhoods.kojo.db.pdo.builder
     public: false

src/Environment/Parameters.yml

@@ -1,7 +1,8 @@
 parameters:
-  env(REDIS_PORT): 6379
-  env(REDIS_HOST): 'redis'
-  env(DATABASE_USERNAME): ''
-  env(DATABASE_PASSWORD): ''
-  env(DATABASE_ADAPTER): ''
-  env(DATABASE_HOST): ''
+  neighborhoods.kojo.environment.parameters.redis_port:
+  neighborhoods.kojo.environment.parameters.redis_host:
+  neighborhoods.kojo.environment.parameters.database_user_name:
+  neighborhoods.kojo.environment.parameters.database_password:
+  neighborhoods.kojo.environment.parameters.database_adapter:
+  neighborhoods.kojo.environment.parameters.database_host:
+  neighborhoods.kojo.environment.parameters.database_name:

src/Process/Pool.php

@@ -79,16 +79,21 @@ public function getCountOfChildProcesses(): int
 
     public function addChildProcess(ProcessInterface $childProcess): PoolInterface
     {
-        $this->_getProcessSignal()->incrementWaitCount();
-        if ($this->isFull()) {
-            throw new \LogicException('Process pool is full.');
-        }else {
-            $childProcess->start();
-            $this->_childProcesses[$childProcess->getProcessId()] = $childProcess;
-            $message = "Forked process[{$childProcess->getProcessId()}][{$childProcess->getTypeCode()}].";
-            $this->_getLogger()->info($message);
+        try{
+            $this->_getProcessSignal()->incrementWaitCount();
+            if ($this->isFull()) {
+                throw new \LogicException('Process pool is full.');
+            }else {
+                $childProcess->start();
+                $this->_childProcesses[$childProcess->getProcessId()] = $childProcess;
+                $message = "Forked process[{$childProcess->getProcessId()}][{$childProcess->getTypeCode()}].";
+                $this->_getLogger()->info($message);
+            }
+            $this->_getProcessSignal()->decrementWaitCount();
+        }catch(\Throwable $throwable){
+            $this->_getProcessSignal()->decrementWaitCount();
+            throw $throwable;
         }
-        $this->_getProcessSignal()->decrementWaitCount();
 
         return $this;
     }

src/Process/Pool/Factory.yml

@@ -1,7 +1,7 @@
 parameters:
-  process_pool_strategy.max_child_processes: 20
+  process_pool_strategy.max_child_processes: 10
   process_pool_strategy.child_process_wait_throttle: 1
-  process_pool_strategy.max_alarm_time: 60
+  process_pool_strategy.max_alarm_time: 5
   process_pool_strategy-server.max_child_processes: 1
   process_pool_strategy-server.child_process_wait_throttle: 1
   process_pool_strategy-server.max_alarm_time: 0

src/Process/Pool/Strategy.php

@@ -30,7 +30,11 @@ protected function _listenerProcessExited(ListenerInterface $listenerProcess): S
         if ($listenerProcess->getExitCode() !== 0) {
             $this->_pauseListenerProcess($listenerProcess);
         }else {
-            while (!$this->_getProcessPool()->isFull() && $listenerProcess->hasMessages()) {
+            while (
+                $listenerProcess->hasMessages()
+                && !$this->_getProcessPool()->isFull()
+                && $this->_getProcessPool()->canEnvironmentSustainAdditionProcesses()
+            ) {
                 $listenerProcess->processMessages();
             }
 
@@ -67,7 +71,7 @@ public function currentPendingChildExitsCompleted(): StrategyInterface
     protected function _jobProcessExited(JobInterface $jobProcess): Strategy
     {
         $this->_getProcessPool()->freeChildProcess($jobProcess->getProcessId());
-        if ($jobProcess->getExitCode() !== 0) {
+        if ($jobProcess->getExitCode() !== 0 && $this->_getProcessPool()->canEnvironmentSustainAdditionProcesses()) {
             $typeCode = $jobProcess->getTypeCode();
             $replacementProcess = $this->_getProcessCollection()->getProcessPrototypeClone($typeCode);
             $replacementProcess->setThrottle($this->getChildProcessWaitThrottle());
@@ -82,7 +86,7 @@ protected function _jobProcessExited(JobInterface $jobProcess): Strategy
 
     public function receivedAlarm(): StrategyInterface
     {
-        if (!$this->_getProcessPool()->isFull()) {
+        if (!$this->_getProcessPool()->isFull() && $this->_getProcessPool()->canEnvironmentSustainAdditionProcesses()) {
             if ($this->_hasPausedListenerProcess()) {
                 $this->_unPauseListenerProcesses();
             }else {
@@ -105,7 +109,7 @@ public function initializePool(): StrategyInterface
         foreach ($this->_getProcessCollection() as $process) {
             $this->_getProcessPool()->addChildProcess($process);
         }
-        if ($this->_hasFillProcessTypeCode()) {
+        if ($this->_hasFillProcessTypeCode() && $this->_getProcessPool()->canEnvironmentSustainAdditionProcesses()) {
             while (!$this->_getProcessPool()->isFull()) {
                 $fillProcessTypeCode = $this->_getFillProcessTypeCode();
                 $fillProcess = $this->_getProcessCollection()->getProcessPrototypeClone($fillProcessTypeCode);

src/Process/PoolAbstract.php

@@ -43,19 +43,17 @@ public function setAlarm(int $seconds): PoolInterface
 
     public function isEmpty(): bool
     {
-        return (bool)($this->getCountOfChildProcesses() === 0);
+        return ($this->getCountOfChildProcesses() === 0);
     }
 
     public function isFull(): bool
     {
-        if ((float)current(sys_getloadavg()) > $this->_getProcessPoolStrategy()->getMaximumLoadAverage()) {
-            $isFull = true;
-        }else {
-            $maxChildProcesses = $this->_getProcessPoolStrategy()->getMaxChildProcesses();
-            $isFull = (bool)($this->getCountOfChildProcesses() >= $maxChildProcesses);
-        }
+        return ($this->getCountOfChildProcesses() >= $this->_getProcessPoolStrategy()->getMaxChildProcesses());
+    }
 
-        return $isFull;
+    public function canEnvironmentSustainAdditionProcesses(): bool
+    {
+        return ((float)current(sys_getloadavg()) <= $this->_getProcessPoolStrategy()->getMaximumLoadAverage());
     }
 
     protected function _initialize(): PoolInterface

src/Process/PoolInterface.php

@@ -36,4 +36,6 @@ public function getCountOfChildProcesses(): int;
     public function setProcess(ProcessInterface $process);
 
     public function getProcess(): ProcessInterface;
+
+    public function canEnvironmentSustainAdditionProcesses(): bool;
 }

src/Redis/Factory.yml

@@ -5,8 +5,8 @@ services:
     shared: true
     calls:
       - [addOption, [!php/const \Redis::OPT_READ_TIMEOUT, '-1']]
-      - [setHost, ['%env(REDIS_HOST)%']]
-      - [setPort, ['%env(REDIS_PORT)%']]
+      - [setHost, ['%neighborhoods.kojo.environment.parameters.redis_host%']]
+      - [setPort, ['%neighborhoods.kojo.environment.parameters.redis_port%']]
   redis.factory:
     alias: neighborhoods.kojo.redis.factory
     public: false

tests/Application/Environment/Parameters.yml

@@ -0,0 +1,8 @@
+parameters:
+  neighborhoods.kojo.environment.parameters.redis_port: '%env(REDIS_PORT)%'
+  neighborhoods.kojo.environment.parameters.redis_host: '%env(REDIS_HOST)%'
+  neighborhoods.kojo.environment.parameters.database_user_name: '%env(DATABASE_USERNAME)%'
+  neighborhoods.kojo.environment.parameters.database_password: '%env(DATABASE_PASSWORD)%'
+  neighborhoods.kojo.environment.parameters.database_adapter: '%env(DATABASE_ADAPTER)%'
+  neighborhoods.kojo.environment.parameters.database_host: '%env(DATABASE_HOST)%'
+  neighborhoods.kojo.environment.parameters.database_name: '%env(DATABASE_NAME)%'

tests/BehaviorSmokeTest/dataSet1/fixtures/1_job_types.yml

@@ -11,7 +11,7 @@ kojo_job_type:
     schedule_limit: 0
     schedule_limit_allowance: 1
     is_enabled: 1
-    auto_complete_success: 1
+    auto_complete_success: 0
     auto_delete_interval_duration: 'PT0S'
   -
     kojo_job_type_id: 2
@@ -25,5 +25,5 @@ kojo_job_type:
     schedule_limit: 3
     schedule_limit_allowance: 1
     is_enabled: 1
-    auto_complete_success: 1
+    auto_complete_success: 0
     auto_delete_interval_duration: 'PT0S'