From 5d6cdc2feef5b7884e61c3109d0598500c3d9fc3 Mon Sep 17 00:00:00 2001 From: yuri Date: Thu, 29 Nov 2018 15:16:42 +0200 Subject: [PATCH] job queue --- application/Espo/Core/CronManager.php | 49 +++++++++++++++---- application/Espo/Core/Mail/Importer.php | 3 +- application/Espo/Core/Utils/Cron/Job.php | 10 ++-- .../Common/AssignmentEmailNotification.php | 9 ++-- application/Espo/Hooks/Common/Stream.php | 18 ++++--- application/Espo/Jobs/ProcessJobQueueQ1.php | 44 +++++++++++++++++ .../Espo/Resources/i18n/en_US/Job.json | 4 +- .../Resources/layouts/Job/detailSmall.json | 12 +++-- .../Espo/Resources/layouts/Job/filters.json | 4 +- .../Espo/Resources/layouts/Job/list.json | 10 ++-- .../Resources/metadata/entityDefs/Job.json | 16 +++++- .../metadata/entityDefs/ScheduledJob.json | 5 ++ application/Espo/Services/Pdf.php | 3 +- application/Espo/Services/User.php | 7 +-- 14 files changed, 152 insertions(+), 42 deletions(-) create mode 100644 application/Espo/Jobs/ProcessJobQueueQ1.php diff --git a/application/Espo/Core/CronManager.php b/application/Espo/Core/CronManager.php index d86de933ff..be7007d63a 100644 --- a/application/Espo/Core/CronManager.php +++ b/application/Espo/Core/CronManager.php @@ -189,9 +189,25 @@ class CronManager $this->getCronJobUtil()->updateFailedJobAttempts(); $this->createJobsFromScheduledJobs(); $this->getCronJobUtil()->removePendingJobDuplicates(); - $pendingJobList = $this->getCronJobUtil()->getPendingJobList(); - if ($this->useProcessPool()) { + $this->processPendingJobs(); + } + + public function processPendingJobs($queue = null, $limit = null, $poolDisabled = false, $noLock = false) + { + if (is_null($limit)) { + $limit = intval($this->getConfig()->get('jobMaxPortion', 0)); + } + + $pendingJobList = $this->getCronJobUtil()->getPendingJobList($queue, $limit); + + $useProcessPool = $this->useProcessPool(); + + if ($poolDisabled) { + $useProcessPool = false; + } + + if ($useProcessPool) { $pool = \Spatie\Async\Pool::create() ->autoload(getcwd() . '/vendor/autoload.php') ->concurrency($this->getConfig()->get('jobPoolConcurrencyNumber')) @@ -200,8 +216,8 @@ class CronManager foreach ($pendingJobList as $job) { $skip = false; - $this->getEntityManager()->getPdo()->query('LOCK TABLES `job` WRITE'); - if ($this->getCronJobUtil()->isJobPending($job->id)) { + if (!$noLock) $this->lockJobTable(); + if ($noLock || $this->getCronJobUtil()->isJobPending($job->id)) { if ($job->get('scheduledJobId')) { if ($this->getCronJobUtil()->isScheduledJobRunning($job->get('scheduledJobId'), $job->get('targetId'), $job->get('targetType'))) { $skip = true; @@ -212,10 +228,10 @@ class CronManager } if ($skip) { - $this->getEntityManager()->getPdo()->query('UNLOCK TABLES'); + if (!$noLock) $this->unlockTables(); continue; } - if ($this->useProcessPool()) { + if ($useProcessPool) { $job->set('status', self::READY); } else { $job->set('status', self::RUNNING); @@ -223,9 +239,9 @@ class CronManager } $this->getEntityManager()->saveEntity($job); - $this->getEntityManager()->getPdo()->query('UNLOCK TABLES'); + if (!$noLock) $this->unlockTables(); - if ($this->useProcessPool()) { + if ($useProcessPool) { $task = new \Espo\Core\Utils\Cron\JobTask($job->id); $pool->add($task); } else { @@ -233,11 +249,21 @@ class CronManager } } - if ($this->useProcessPool()) { + if ($useProcessPool) { $pool->wait(); } } + protected function lockJobTable() + { + $this->getEntityManager()->getPdo()->query('LOCK TABLES `job` WRITE'); + } + + protected function unlockTables() + { + $this->getEntityManager()->getPdo()->query('UNLOCK TABLES'); + } + public function runJobById($id) { if (empty($id)) throw new Error(); @@ -281,6 +307,11 @@ class CronManager $status = $isSuccess ? self::SUCCESS : self::FAILED; $job->set('status', $status); + + if ($isSuccess) { + $job->set('executedAt', date('Y-m-d H:i:s')); + } + $this->getEntityManager()->saveEntity($job); if ($job->get('scheduledJobId') && !$skipLog) { diff --git a/application/Espo/Core/Mail/Importer.php b/application/Espo/Core/Mail/Importer.php index 5a3c7db52f..5fce35f894 100644 --- a/application/Espo/Core/Mail/Importer.php +++ b/application/Espo/Core/Mail/Importer.php @@ -520,7 +520,8 @@ class Importer 'targetType' => 'Email', 'targetId' => $duplicate->id ], - 'executeAt' => $executeAt + 'executeAt' => $executeAt, + 'queue' => 'q1' ]); $this->getEntityManager()->saveEntity($job); } diff --git a/application/Espo/Core/Utils/Cron/Job.php b/application/Espo/Core/Utils/Cron/Job.php index b0cc255653..77284c6c05 100644 --- a/application/Espo/Core/Utils/Cron/Job.php +++ b/application/Espo/Core/Utils/Cron/Job.php @@ -74,10 +74,8 @@ class Job ])->findOne(); } - public function getPendingJobList() + public function getPendingJobList($queue = null, $limit = 0) { - $limit = intval($this->getConfig()->get('jobMaxPortion', 0)); - $selectParams = [ 'select' => [ 'id', @@ -89,11 +87,13 @@ class Job 'methodName', 'method', // TODO remove deprecated 'serviceName', - 'data' + 'data', + 'queue' ], 'whereClause' => [ 'status' => CronManager::PENDING, - 'executeTime<=' => date('Y-m-d H:i:s') + 'executeTime<=' => date('Y-m-d H:i:s'), + 'queue' => $queue ], 'orderBy' => 'executeTime' ]; diff --git a/application/Espo/Hooks/Common/AssignmentEmailNotification.php b/application/Espo/Hooks/Common/AssignmentEmailNotification.php index b8cbf3c256..f79dfd2d46 100644 --- a/application/Espo/Hooks/Common/AssignmentEmailNotification.php +++ b/application/Espo/Hooks/Common/AssignmentEmailNotification.php @@ -87,17 +87,18 @@ class AssignmentEmailNotification extends \Espo\Core\Hooks\Base protected function createJob(Entity $entity, $userId) { $job = $this->getEntityManager()->getEntity('Job'); - $job->set(array( + $job->set([ 'serviceName' => 'EmailNotification', 'methodName' => 'notifyAboutAssignmentJob', - 'data' => json_encode(array( + 'data' => [ 'userId' => $userId, 'assignerUserId' => $this->getUser()->id, 'entityId' => $entity->id, 'entityType' => $entity->getEntityType() - )), + ], 'executeTime' => date('Y-m-d H:i:s'), - )); + 'queue' => 'q1' + ]); $this->getEntityManager()->saveEntity($job); } } diff --git a/application/Espo/Hooks/Common/Stream.php b/application/Espo/Hooks/Common/Stream.php index 21c75d0377..8f862f3e55 100644 --- a/application/Espo/Hooks/Common/Stream.php +++ b/application/Espo/Hooks/Common/Stream.php @@ -258,15 +258,16 @@ class Stream extends \Espo\Core\Hooks\Base if (!empty($autofollowUserIdList)) { $job = $this->getEntityManager()->getEntity('Job'); - $job->set(array( + $job->set([ 'serviceName' => 'Stream', 'methodName' => 'afterRecordCreatedJob', - 'data' => array( + 'data' => [ 'userIdList' => $autofollowUserIdList, 'entityType' => $entity->getEntityType(), 'entityId' => $entity->id - ) - )); + ], + 'queue' => 'q1' + ]); $this->getEntityManager()->saveEntity($job); } } else { @@ -337,14 +338,15 @@ class Stream extends \Espo\Core\Hooks\Base ) ) { $job = $this->getEntityManager()->getEntity('Job'); - $job->set(array( + $job->set([ 'serviceName' => 'Stream', 'methodName' => 'controlFollowersJob', - 'data' => array( + 'data' => [ 'entityType' => $entity->getEntityType(), 'entityId' => $entity->id - ) - )); + ], + 'queue' => 'q1' + ]); $this->getEntityManager()->saveEntity($job); } } diff --git a/application/Espo/Jobs/ProcessJobQueueQ1.php b/application/Espo/Jobs/ProcessJobQueueQ1.php new file mode 100644 index 0000000000..144f058430 --- /dev/null +++ b/application/Espo/Jobs/ProcessJobQueueQ1.php @@ -0,0 +1,44 @@ +getConfig()->get('jobQ1MaxPortion', 500); + + $cronManager = new \Espo\Core\CronManager($this->getContainer()); + + $cronManager->processPendingJobs('q1', $limit, true, true); + } +} diff --git a/application/Espo/Resources/i18n/en_US/Job.json b/application/Espo/Resources/i18n/en_US/Job.json index afdfe4f2cd..3b84d0d2ef 100644 --- a/application/Espo/Resources/i18n/en_US/Job.json +++ b/application/Espo/Resources/i18n/en_US/Job.json @@ -2,6 +2,7 @@ "fields": { "status": "Status", "executeTime": "Execute At", + "executedAt": "Executed At", "attempts": "Attempts Left", "failedAttempts": "Failed Attempts", "serviceName": "Service", @@ -11,7 +12,8 @@ "scheduledJobJob": "Scheduled Job Name", "data": "Data", "targetType": "Target Type", - "targetId": "Target ID" + "targetId": "Target ID", + "queue": "Queue" }, "options": { "status": { diff --git a/application/Espo/Resources/layouts/Job/detailSmall.json b/application/Espo/Resources/layouts/Job/detailSmall.json index 24738a1b34..9a2a5c8edd 100644 --- a/application/Espo/Resources/layouts/Job/detailSmall.json +++ b/application/Espo/Resources/layouts/Job/detailSmall.json @@ -3,9 +3,15 @@ "label":"", "rows":[ [{"name":"name"}, {"name": "status"}], - [{"name":"executeTime"}, {"name": "attempts"}], - [{"name":"createdAt"}, {"name": "failedAttempts"}], - [{"name":"modifiedAt"}, false] + [{"name":"queue"}, false] + ] + }, + { + "label":"", + "rows":[ + [{"name":"executeTime"}, {"name": "createdAt"}], + [{"name":"executedAt"}, {"name": "modifiedAt"}], + [{"name":"attempts"}, {"name": "failedAttempts"}] ] }, { diff --git a/application/Espo/Resources/layouts/Job/filters.json b/application/Espo/Resources/layouts/Job/filters.json index 29c29b1431..4cde905a49 100644 --- a/application/Espo/Resources/layouts/Job/filters.json +++ b/application/Espo/Resources/layouts/Job/filters.json @@ -1,5 +1,7 @@ [ + "status", "createdAt", "executeTime", - "status" + "executedAt", + "queue" ] diff --git a/application/Espo/Resources/layouts/Job/list.json b/application/Espo/Resources/layouts/Job/list.json index b117bcb191..a7136d04ed 100644 --- a/application/Espo/Resources/layouts/Job/list.json +++ b/application/Espo/Resources/layouts/Job/list.json @@ -1,7 +1,7 @@ [ - {"name":"name", "width": 30}, - {"name":"status"}, - {"name":"executeTime"}, - {"name":"attempts", "width": 10}, - {"name":"createdAt", "align": "right", "width": 15} + {"name":"name"}, + {"name":"status", "width": 14}, + {"name":"executeTime", "width": 14}, + {"name":"queue", "width": 10}, + {"name":"createdAt", "align": "right", "width": 14} ] diff --git a/application/Espo/Resources/metadata/entityDefs/Job.json b/application/Espo/Resources/metadata/entityDefs/Job.json index 98d8dd8784..d96990f084 100644 --- a/application/Espo/Resources/metadata/entityDefs/Job.json +++ b/application/Espo/Resources/metadata/entityDefs/Job.json @@ -8,7 +8,13 @@ "status": { "type": "enum", "options": ["Pending", "Ready", "Running", "Success", "Failed"], - "default": "Pending" + "default": "Pending", + "style": { + "Success": "success", + "Failed": "danger", + "Running": "warning", + "Ready": "warning" + } }, "executeTime": { "type": "datetime", @@ -39,6 +45,14 @@ "link": "scheduledJob", "field": "job" }, + "queue": { + "type": "varchar", + "maxLength": 36, + "default": null + }, + "executedAt": { + "type": "datetime" + }, "pid": { "type": "int" }, diff --git a/application/Espo/Resources/metadata/entityDefs/ScheduledJob.json b/application/Espo/Resources/metadata/entityDefs/ScheduledJob.json index c36caafbcf..8c917fe066 100644 --- a/application/Espo/Resources/metadata/entityDefs/ScheduledJob.json +++ b/application/Espo/Resources/metadata/entityDefs/ScheduledJob.json @@ -89,6 +89,11 @@ "name": "Check for New Versions of Installed Extensions", "isSystem": true, "scheduling": "25 5 * * *" + }, + "ProcessJobQueueQ1": { + "name": "Process Job Queue Q1", + "isSystem": true, + "scheduling": "*/1 * * * *" } } } diff --git a/application/Espo/Services/Pdf.php b/application/Espo/Services/Pdf.php index 5ffe5c9d0d..1a9e9ae005 100644 --- a/application/Espo/Services/Pdf.php +++ b/application/Espo/Services/Pdf.php @@ -227,7 +227,8 @@ class Pdf extends \Espo\Core\Services\Base 'data' => [ 'id' => $attachment->id ], - 'executeTime' => (new \DateTime())->modify('+' . $this->removeMassFilePeriod)->format('Y-m-d H:i:s') + 'executeTime' => (new \DateTime())->modify('+' . $this->removeMassFilePeriod)->format('Y-m-d H:i:s'), + 'queue' => 'q1' ]); $this->getEntityManager()->saveEntity($job); diff --git a/application/Espo/Services/User.php b/application/Espo/Services/User.php index 8904863d3a..d132eab42d 100644 --- a/application/Espo/Services/User.php +++ b/application/Espo/Services/User.php @@ -225,14 +225,15 @@ class User extends Record $job = $this->getEntityManager()->getEntity('Job'); - $job->set(array( + $job->set([ 'serviceName' => 'User', 'methodName' => 'removeChangePasswordRequestJob', 'data' => [ 'id' => $passwordChangeRequest->id ], - 'executeTime' => $dt->format('Y-m-d H:i:s') - )); + 'executeTime' => $dt->format('Y-m-d H:i:s'), + 'queue' => 'q1' + ]); $this->getEntityManager()->saveEntity($job);