From 5350cf8e524dfa141c110829af42cd8f730f6c5f Mon Sep 17 00:00:00 2001 From: yuri Date: Mon, 5 Nov 2018 16:45:42 +0200 Subject: [PATCH] parallel jobs and daemon --- application/Espo/Core/Application.php | 44 +++++ application/Espo/Core/CronManager.php | 117 +++++++++--- application/Espo/Core/Utils/Cron/Job.php | 13 +- application/Espo/Core/Utils/Cron/JobTask.php | 50 ++++++ application/Espo/Core/defaults/config.php | 1 - .../Espo/Core/defaults/systemConfig.php | 8 + .../Resources/metadata/entityDefs/Job.json | 2 +- composer.json | 6 +- composer.lock | 169 +++++++++++++++++- daemon.php | 38 ++++ 10 files changed, 408 insertions(+), 40 deletions(-) create mode 100644 application/Espo/Core/Utils/Cron/JobTask.php create mode 100644 daemon.php diff --git a/application/Espo/Core/Application.php b/application/Espo/Core/Application.php index 03845c73e4..8941568c4e 100644 --- a/application/Espo/Core/Application.php +++ b/application/Espo/Core/Application.php @@ -151,6 +151,50 @@ class Application $cronManager->run(); } + public function runDaemon() + { + $maxProcessNumber = $this->getConfig()->get('daemonMaxProcessNumber'); + $interval = $this->getConfig()->get('daemonInterval'); + $timeout = $this->getConfig()->get('daemonProcessTimeout'); + + if (!$maxProcessNumber || !$interval) { + $GLOBALS['log']->error("Daemon config params are not set."); + return; + } + + $processList = []; + while (true) { + $toSkip = false; + $runningCount = 0; + foreach ($processList as $i => $process) { + if ($process->isRunning()) { + $runningCount++; + } else if ($process->isRunning()) { + unset($processList[$i]); + } + } + $processList = array_values($processList); + if (count($runningCount) >= $maxProcessNumber) { + $toSkip = true; + } + if (!$toSkip) { + $process = new \Symfony\Component\Process\Process(['php', 'cron.php']); + $process->setTimeout($timeout); + $process->run(); + } + sleep($interval); + } + } + + public function runJob($id) + { + $auth = $this->createAuth(); + $auth->useNoAuth(); + + $cronManager = new \Espo\Core\CronManager($this->container); + $cronManager->runJobById($id); + } + public function runRebuild() { $dataManager = $this->getContainer()->get('dataManager'); diff --git a/application/Espo/Core/CronManager.php b/application/Espo/Core/CronManager.php index 983d0c390f..27577a9bd1 100644 --- a/application/Espo/Core/CronManager.php +++ b/application/Espo/Core/CronManager.php @@ -28,9 +28,12 @@ ************************************************************************/ namespace Espo\Core; + use \PDO; use Espo\Core\Utils\Json; + use Espo\Core\Exceptions\NotFound; +use Espo\Core\Exceptions\Error; class CronManager { @@ -48,8 +51,12 @@ class CronManager private $cronScheduledJobUtil; + private $useProcessPool = false; + const PENDING = 'Pending'; + const READY = 'Ready'; + const RUNNING = 'Running'; const SUCCESS = 'Success'; @@ -70,6 +77,14 @@ class CronManager $this->scheduledJobUtil = $this->container->get('scheduledJob'); $this->cronJobUtil = new \Espo\Core\Utils\Cron\Job($this->config, $this->entityManager); $this->cronScheduledJobUtil = new \Espo\Core\Utils\Cron\ScheduledJob($this->config, $this->entityManager); + + if ($this->getConfig()->get('jobRunInParallel')) { + if (\Spatie\Async\Pool::isSupported()) { + $this->useProcessPool = true; + } else { + $GLOBALS['log']->warning("CronManager: useProcessPool requires pcntl and posix extensions."); + } + } } protected function getContainer() @@ -146,6 +161,16 @@ class CronManager return false; } + protected function useProcessPool() + { + return $this->useProcessPool; + } + + public function setUseProcessPool($useProcessPool) + { + $this->useProcessPool = $useProcessPool; + } + /** * Run Cron * @@ -166,6 +191,13 @@ class CronManager $this->getCronJobUtil()->removePendingJobDuplicates(); $pendingJobList = $this->getCronJobUtil()->getPendingJobList(); + if ($this->useProcessPool()) { + $pool = \Spatie\Async\Pool::create() + ->autoload(getcwd() . '/vendor/autoload.php') + ->concurrency($this->getConfig()->get('jobPoolConcurrency')) + ->timeout($this->getConfig()->get('jobPeriodForActiveProcess')); + } + foreach ($pendingJobList as $job) { $skip = false; $this->getEntityManager()->getPdo()->query('LOCK TABLES `job` WRITE'); @@ -183,39 +215,76 @@ class CronManager $this->getEntityManager()->getPdo()->query('UNLOCK TABLES'); continue; } + if ($this->useProcessPool()) { + $job->set('status', self::READY); + } else { + $job->set('status', self::RUNNING); + $job->set('pid', \Espo\Core\Utils\System::getPid()); + } - $job->set('status', self::RUNNING); - $job->set('pid', $this->getCronJobUtil()->getPid()); $this->getEntityManager()->saveEntity($job); $this->getEntityManager()->getPdo()->query('UNLOCK TABLES'); - $isSuccess = true; - $skipLog = false; - - try { - if ($job->get('scheduledJobId')) { - $this->runScheduledJob($job); - } else { - $this->runService($job); - } - } catch (\Exception $e) { - $isSuccess = false; - if ($e->getCode() === -1) { - $job->set('attempts', 0); - $skipLog = true; - } else { - $GLOBALS['log']->error('CronManager: Failed job running, job ['.$job->id.']. Error Details: '.$e->getMessage()); - } + if ($this->useProcessPool()) { + $task = new \Espo\Core\Utils\Cron\JobTask($job->id); + $pool->add($task); + } else { + $this->runJob($job); } + } - $status = $isSuccess ? self::SUCCESS : self::FAILED; + if ($this->useProcessPool()) { + $pool->wait(); + } + } - $job->set('status', $status); - $this->getEntityManager()->saveEntity($job); + public function runJobById($id) + { + if (empty($id)) throw new Error(); - if ($job->get('scheduledJobId') && !$skipLog) { - $this->getCronScheduledJobUtil()->addLogRecord($job->get('scheduledJobId'), $status, null, $job->get('targetId'), $job->get('targetType')); + $job = $this->getEntityManager()->getEntity('Job', $id); + + if (!$job) throw new Error("Job {$id} not found."); + + if ($job->get('status') !== self::READY) { + throw new Error("Can't run job {$id} with no status Ready."); + } + + $job->set('status', self::RUNNING); + $job->set('pid', \Espo\Core\Utils\System::getPid()); + $this->getEntityManager()->saveEntity($job); + + $this->runJob($job); + } + + public function runJob($job) + { + $isSuccess = true; + $skipLog = false; + + try { + if ($job->get('scheduledJobId')) { + $this->runScheduledJob($job); + } else { + $this->runService($job); } + } catch (\Exception $e) { + $isSuccess = false; + if ($e->getCode() === -1) { + $job->set('attempts', 0); + $skipLog = true; + } else { + $GLOBALS['log']->error('CronManager: Failed job running, job ['.$job->id.']. Error Details: '.$e->getMessage()); + } + } + + $status = $isSuccess ? self::SUCCESS : self::FAILED; + + $job->set('status', $status); + $this->getEntityManager()->saveEntity($job); + + if ($job->get('scheduledJobId') && !$skipLog) { + $this->getCronScheduledJobUtil()->addLogRecord($job->get('scheduledJobId'), $status, null, $job->get('targetId'), $job->get('targetType')); } } diff --git a/application/Espo/Core/Utils/Cron/Job.php b/application/Espo/Core/Utils/Cron/Job.php index 955fd95d2c..b0cc255653 100644 --- a/application/Espo/Core/Utils/Cron/Job.php +++ b/application/Espo/Core/Utils/Cron/Job.php @@ -109,7 +109,7 @@ class Job { $where = [ 'scheduledJobId' => $scheduledJobId, - 'status' => CronManager::RUNNING + 'status' => [CronManager::RUNNING, CronManager::READY] ]; if ($targetId && $targetType) { $where['targetId'] = $targetId; @@ -127,7 +127,7 @@ class Job $query = " SELECT scheduled_job_id FROM job WHERE - `status` = 'Running' AND + (`status` = 'Running' OR `status` = 'Ready') AND scheduled_job_id IS NOT NULL AND target_id IS NULL AND deleted = 0 @@ -205,7 +205,7 @@ class Job $select = " SELECT id, scheduled_job_id, execute_time, target_id, target_type, pid FROM `job` WHERE - `status` = '" . CronManager::RUNNING ."' AND execute_time < '".date('Y-m-d H:i:s', $time)."' + (`status` = '" . CronManager::RUNNING ."' OR `status` = '" . CronManager::READY ."') AND execute_time < '".date('Y-m-d H:i:s', $time)."' "; $sth = $pdo->prepare($select); $sth->execute(); @@ -357,9 +357,4 @@ class Job } } } - - public function getPid() - { - return System::getPid(); - } -} \ No newline at end of file +} diff --git a/application/Espo/Core/Utils/Cron/JobTask.php b/application/Espo/Core/Utils/Cron/JobTask.php new file mode 100644 index 0000000000..2a588cd7e4 --- /dev/null +++ b/application/Espo/Core/Utils/Cron/JobTask.php @@ -0,0 +1,50 @@ +jobId = $jobId; + } + + public function configure() + { + } + + public function run() + { + $app = new \Espo\Core\Application(); + $app->runJob($this->jobId); + } +} diff --git a/application/Espo/Core/defaults/config.php b/application/Espo/Core/defaults/config.php index ff67ef3c68..0c89a01e28 100644 --- a/application/Espo/Core/defaults/config.php +++ b/application/Espo/Core/defaults/config.php @@ -179,4 +179,3 @@ return array ( 'noteEditThresholdPeriod' => '7 days', 'isInstalled' => false ); - diff --git a/application/Espo/Core/defaults/systemConfig.php b/application/Espo/Core/defaults/systemConfig.php index 09051bd833..7b5f4286ca 100644 --- a/application/Espo/Core/defaults/systemConfig.php +++ b/application/Espo/Core/defaults/systemConfig.php @@ -58,7 +58,12 @@ return [ 'jobPeriod' => 7800, /** Max execution time (in seconds) allocated for a sinle job. If exceeded then set to Failed.*/ 'jobPeriodForActiveProcess' => 36000, /** Max execution time (in seconds) allocated for a sinle job with active process. If exceeded then set to Failed.*/ 'jobRerunAttemptNumber' => 1, /** Number of attempts to re-run failed jobs. */ + 'jobRunInParallel' => false, /** Jobs will be executed in parallel processes. */ + 'jobPoolConcurrency' => 8, /** Max number of processes run simultaneously. */ 'cronMinInterval' => 4, /** Min interval (in seconds) between two cron runs. */ + 'daemonMaxProcessNumber' => 5, + 'daemonInterval' => 10, /** Interval between cron process runs in seconds. */ + 'daemonProcessTimeout' => 36000, 'crud' => array( 'get' => 'read', 'post' => 'create', @@ -113,7 +118,10 @@ return [ 'jobMaxPortion', 'jobPeriod', 'jobRerunAttemptNumber', + 'jobUseThreads', 'cronMinInterval', + 'daemonInterval', + 'daemonMaxThreadNumber', 'authenticationMethod', 'adminPanelIframeUrl', 'ldapHost', diff --git a/application/Espo/Resources/metadata/entityDefs/Job.json b/application/Espo/Resources/metadata/entityDefs/Job.json index 41a904c975..98d8dd8784 100644 --- a/application/Espo/Resources/metadata/entityDefs/Job.json +++ b/application/Espo/Resources/metadata/entityDefs/Job.json @@ -7,7 +7,7 @@ }, "status": { "type": "enum", - "options": ["Pending", "Running", "Success", "Failed"], + "options": ["Pending", "Ready", "Running", "Success", "Failed"], "default": "Pending" }, "executeTime": { diff --git a/composer.json b/composer.json index 382eb18240..b0a36b8435 100644 --- a/composer.json +++ b/composer.json @@ -4,7 +4,7 @@ "homepage": "https://github.com/espocrm/espocrm.git", "license": "GPL-3.0-only", "require": { - "php": ">=5.6.0", + "php": ">=7.1.0", "ext-pdo_mysql": "*", "ext-openssl": "*", "ext-json": "*", @@ -29,7 +29,9 @@ "php-mime-mail-parser/php-mime-mail-parser": "3.*", "zbateson/mail-mime-parser": "0.4.*", "phpoffice/phpexcel": "^1.8", - "phpoffice/phpspreadsheet": "^1.1" + "phpoffice/phpspreadsheet": "^1.1", + "spatie/async": "0.0.4", + "symfony/process": "4.1.7" }, "autoload": { "psr-0": { diff --git a/composer.lock b/composer.lock index 335f055567..6203cdc956 100644 --- a/composer.lock +++ b/composer.lock @@ -4,8 +4,8 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#composer-lock-the-lock-file", "This file is @generated automatically" ], - "hash": "6a0640e0f5e0603fc55bbed741e069ef", - "content-hash": "06d06c8778f2fe549244a579ee6d048d", + "hash": "ecef27207089d736defa12613a2e7102", + "content-hash": "6a7ab1e25c048f4144772e60140d0982", "packages": [ { "name": "composer/semver", @@ -686,6 +686,67 @@ ], "time": "2013-11-23 19:48:39" }, + { + "name": "opis/closure", + "version": "3.1.1", + "source": { + "type": "git", + "url": "https://github.com/opis/closure.git", + "reference": "d3209e46ad6c69a969b705df0738fd0dbe26ef9e" + }, + "dist": { + "type": "zip", + "url": "https://api.github.com/repos/opis/closure/zipball/d3209e46ad6c69a969b705df0738fd0dbe26ef9e", + "reference": "d3209e46ad6c69a969b705df0738fd0dbe26ef9e", + "shasum": "" + }, + "require": { + "php": "^5.4 || ^7.0" + }, + "require-dev": { + "jeremeamia/superclosure": "^2.0", + "phpunit/phpunit": "^4.0" + }, + "type": "library", + "extra": { + "branch-alias": { + "dev-master": "3.0.x-dev" + } + }, + "autoload": { + "psr-4": { + "Opis\\Closure\\": "src/" + }, + "files": [ + "functions.php" + ] + }, + "notification-url": "https://packagist.org/downloads/", + "license": [ + "MIT" + ], + "authors": [ + { + "name": "Marius Sarca", + "email": "marius.sarca@gmail.com" + }, + { + "name": "Sorin Sarca", + "email": "sarca_sorin@hotmail.com" + } + ], + "description": "A library that can be used to serialize closures (anonymous functions) and arbitrary objects.", + "homepage": "https://opis.io/closure", + "keywords": [ + "anonymous functions", + "closure", + "function", + "serializable", + "serialization", + "serialize" + ], + "time": "2018-10-02 13:36:53" + }, { "name": "php-mime-mail-parser/php-mime-mail-parser", "version": "3.0.2", @@ -1044,6 +1105,108 @@ ], "time": "2015-03-08 18:41:17" }, + { + "name": "spatie/async", + "version": "0.0.4", + "source": { + "type": "git", + "url": "https://github.com/spatie/async.git", + "reference": "8b76df4ab77dcf7680eee4b83353d038e28f92f7" + }, + "dist": { + "type": "zip", + "url": "https://api.github.com/repos/spatie/async/zipball/8b76df4ab77dcf7680eee4b83353d038e28f92f7", + "reference": "8b76df4ab77dcf7680eee4b83353d038e28f92f7", + "shasum": "" + }, + "require": { + "opis/closure": "^3.0", + "php": "^7.1", + "symfony/process": "^3.3 || ^4.0" + }, + "require-dev": { + "larapack/dd": "^1.1", + "phpunit/phpunit": "^6.0", + "symfony/stopwatch": "^4.0" + }, + "type": "library", + "autoload": { + "files": [ + "src/helpers.php" + ], + "psr-4": { + "Spatie\\Async\\": "src" + } + }, + "notification-url": "https://packagist.org/downloads/", + "license": [ + "MIT" + ], + "authors": [ + { + "name": "Brent Roose", + "email": "brent@spatie.be", + "homepage": "https://spatie.be", + "role": "Developer" + } + ], + "description": "Asynchronous and parallel PHP with the PCNTL extension", + "homepage": "https://github.com/spatie/async", + "keywords": [ + "async", + "spatie" + ], + "time": "2018-01-29 15:09:29" + }, + { + "name": "symfony/process", + "version": "v4.1.7", + "source": { + "type": "git", + "url": "https://github.com/symfony/process.git", + "reference": "3e83acef94d979b1de946599ef86b3a352abcdc9" + }, + "dist": { + "type": "zip", + "url": "https://api.github.com/repos/symfony/process/zipball/3e83acef94d979b1de946599ef86b3a352abcdc9", + "reference": "3e83acef94d979b1de946599ef86b3a352abcdc9", + "shasum": "" + }, + "require": { + "php": "^7.1.3" + }, + "type": "library", + "extra": { + "branch-alias": { + "dev-master": "4.1-dev" + } + }, + "autoload": { + "psr-4": { + "Symfony\\Component\\Process\\": "" + }, + "exclude-from-classmap": [ + "/Tests/" + ] + }, + "notification-url": "https://packagist.org/downloads/", + "license": [ + "MIT" + ], + "authors": [ + { + "name": "Fabien Potencier", + "email": "fabien@symfony.com" + }, + { + "name": "Symfony Community", + "homepage": "https://symfony.com/contributors" + } + ], + "description": "Symfony Process Component", + "homepage": "https://symfony.com", + "time": "2018-10-14 20:48:13" + }, { "name": "symfony/yaml", "version": "v2.7.0", @@ -1851,7 +2014,7 @@ "prefer-stable": false, "prefer-lowest": false, "platform": { - "php": ">=5.6.0", + "php": ">=7.1.0", "ext-pdo_mysql": "*", "ext-openssl": "*", "ext-json": "*", diff --git a/daemon.php b/daemon.php new file mode 100644 index 0000000000..e34a6dd3e6 --- /dev/null +++ b/daemon.php @@ -0,0 +1,38 @@ +runDaemon();