parallel jobs and daemon

2018-11-05 16:45:42 +02:00
parent 0c2959da80
commit 5350cf8e52
10 changed files with 408 additions and 40 deletions
@@ -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');
@@ -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'));
        }
    }

@@ -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();
-    }
-}
+}
@@ -0,0 +1,50 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM - Open Source CRM application.
+ * Copyright (C) 2014-2018 Yuri Kuznetsov, Taras Machyshyn, Oleksiy Avramenko
+ * Website: http://www.espocrm.com
+ *
+ * EspoCRM is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * EspoCRM is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with EspoCRM. If not, see http://www.gnu.org/licenses/.
+ *
+ * The interactive user interfaces in modified source and object code versions
+ * of this program must display Appropriate Legal Notices, as required under
+ * Section 5 of the GNU General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+namespace Espo\Core\Utils\Cron;
+
+class JobTask extends \Spatie\Async\Task
+{
+    private $jobId;
+
+    public function __construct($jobId)
+    {
+        $this->jobId = $jobId;
+    }
+
+    public function configure()
+    {
+    }
+
+    public function run()
+    {
+        $app = new \Espo\Core\Application();
+        $app->runJob($this->jobId);
+    }
+}
@@ -179,4 +179,3 @@ return array (
    'noteEditThresholdPeriod' => '7 days',
    'isInstalled' => false
 );
-
@@ -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',
@@ -7,7 +7,7 @@
        },
        "status": {
            "type": "enum",
-            "options": ["Pending", "Running", "Success", "Failed"],
+            "options": ["Pending", "Ready", "Running", "Success", "Failed"],
            "default": "Pending"
        },
        "executeTime": {
@@ -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": {
@@ -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": "*",
@@ -0,0 +1,38 @@
+<?php
+/************************************************************************
+ * This file is part of EspoCRM.
+ *
+ * EspoCRM - Open Source CRM application.
+ * Copyright (C) 2014-2018 Yuri Kuznetsov, Taras Machyshyn, Oleksiy Avramenko
+ * Website: http://www.espocrm.com
+ *
+ * EspoCRM is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * EspoCRM is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with EspoCRM. If not, see http://www.gnu.org/licenses/.
+ *
+ * The interactive user interfaces in modified source and object code versions
+ * of this program must display Appropriate Legal Notices, as required under
+ * Section 5 of the GNU General Public License version 3.
+ *
+ * In accordance with Section 7(b) of the GNU General Public License version 3,
+ * these Appropriate Legal Notices must retain the display of the "EspoCRM" word.
+ ************************************************************************/
+
+$sapiName = php_sapi_name();
+if (substr($sapiName, 0, 3) != 'cli') {
+    die("Cron can be run only via CLI");
+}
+
+include "bootstrap.php";
+
+$app = new \Espo\Core\Application();
+$app->runDaemon();