#5502 Remake background jobs

inc/classes/BxDolBackgroundJobs.php

@@ -17,7 +17,6 @@ class BxDolBackgroundJobs  extends BxDolFactory implements iBxDolSingleton
 
     protected $_sObjectLog;
 
-    protected $_sParamWorkers;
     protected $_sParamWorkersLimit;
 
     protected $_iAttemptsMax;
@@ -31,7 +30,6 @@ class BxDolBackgroundJobs  extends BxDolFactory implements iBxDolSingleton
 
         $this->_sObjectLog = 'sys_background_jobs';
 
-        $this->_sParamWorkers = 'sys_bg_jobs_workers';
         $this->_sParamWorkersLimit = 'sys_bg_jobs_workers_limit';
 
         $this->_iAttemptsMax = 3;
@@ -114,10 +112,17 @@ class BxDolBackgroundJobs  extends BxDolFactory implements iBxDolSingleton
             ]);
 
         $fStart = microtime(true);
-        $mixedResult = BxDolService::callSerialized($mixedJob['service_call']);
+        $sError = '';
+        $mixedResult = '';
+        try {
+            $mixedResult = BxDolService::callSerialized($mixedJob['service_call']);
+        }
+        catch (Throwable $e) {
+            $sError = $e->getMessage();
+        }
         $fEnd = microtime(true);
 
-        if($mixedResult !== true) {
+        if ($sError) {
             $iAvailableAt = 0;
             $sStatus = BX_DOL_BG_JOBS_STATUS_ERROR;
 
@@ -128,10 +133,11 @@ class BxDolBackgroundJobs  extends BxDolFactory implements iBxDolSingleton
             }
 
             $this->_oQuery->updateJob($mixedJob['name'], [
+                'claim_token' => '',
                 'reserved_at' => 0,
                 'available_at' => $iAvailableAt,
                 'attempts' => $iAttempts,
-                'error' => $mixedResult ?: 'Exited with an error at ' . $fEnd,
+                'error' => $sError . ' / result: [' . $mixedResult . ']',
                 'status' => $sStatus
             ]);
         }
@@ -145,13 +151,16 @@ class BxDolBackgroundJobs  extends BxDolFactory implements iBxDolSingleton
 
     public function processAll($iLimit = 0)
     {
-        $aJobs = $this->_oQuery->getJobs(['sample' => 'process', 'with_priority' => true, 'limit' => $iLimit ?: (getParam('sys_bg_jobs_process_per_run') ?: 5)]);
+        if($this->_isWorkersLimitReached())
+            return false;
+     
+        $sClaimToken = bin2hex(random_bytes(16));
+        $this->_oQuery->claimJobs($sClaimToken, $iLimit ?: (getParam('sys_bg_jobs_process_per_run') ?: 5));
+
+        $aJobs = $this->_oQuery->getClaimedJobs($sClaimToken);
         if(!$aJobs || !is_array($aJobs))
             return true;
 
-        if(!$this->_startWorker())
-            return false;
-
         $this->_oQuery->updateJobByIds(array_keys($aJobs), [
             'reserved_at' => time()
         ]);
@@ -163,7 +172,7 @@ class BxDolBackgroundJobs  extends BxDolFactory implements iBxDolSingleton
 
         bx_log($this->_sObjectLog, "Processed: all (" . $iProcessed . " from " . count($aJobs) . ")", BX_LOG_INFO);
 
-        return $this->_stopWorker();
+        return true;
     }
 
     public function prune($iTimeout, $sStatus = BX_DOL_BG_JOBS_STATUS_ERROR)
@@ -175,18 +184,14 @@ class BxDolBackgroundJobs  extends BxDolFactory implements iBxDolSingleton
         ]);
     }
 
-    protected function _startWorker()
+    protected function _isWorkersLimitReached()
     {
-        $iWorkers = 0;
-        if(($iWorkers = (int)getParam($this->_sParamWorkers)) >= (int)getParam($this->_sParamWorkersLimit))
-            return false;
-
-        return setParam($this->_sParamWorkers, $iWorkers + 1);
-    }
+        $aJobs = $this->_oQuery->getJobs(['sample' => 'running']);
+        $iWorkers = $aJobs ? count($aJobs) : 0;        
+        if ($iWorkers >= (int)getParam($this->_sParamWorkersLimit))
+            return true;
 
-    protected function _stopWorker()
-    {
-        return setParam($this->_sParamWorkers, (int)getParam($this->_sParamWorkers) - 1);
+        return false;
     }
 }
 

inc/classes/BxDolBackgroundJobsQuery.php

@@ -17,6 +17,29 @@ class BxDolBackgroundJobsQuery extends BxDolDb
     	parent::__construct();
     }
 
+    public function getClaimedJobs($sClasimToken)
+    {
+        return $this->getAll("
+            SELECT * FROM `sys_background_jobs` 
+            WHERE `claim_token` = :claim_token AND `status` = '" . BX_DOL_BG_JOBS_STATUS_PENDING . "' 
+            ORDER BY `priority` DESC, `added` ASC", [
+                'claim_token' => $sClasimToken
+            ]
+        );
+    }
+
+    public function claimJobs($sClasimToken, $iLimit)
+    {
+        return $this->query("UPDATE `sys_background_jobs` 
+            SET `reserved_at`=UNIX_TIMESTAMP(), `claim_token`=:claim_token 
+            WHERE claim_token = '' AND `reserved_at`='0' AND (`available_at`='0' OR `available_at`<=UNIX_TIMESTAMP()) AND `status`='" . BX_DOL_BG_JOBS_STATUS_PENDING . "' 
+            ORDER BY `priority` DESC, `added` ASC 
+            LIMIT :limit", [
+                'claim_token' => $sClasimToken,
+                'limit' => (int)$iLimit
+        ]) !== false;
+    }
+
     public function getJobs($aParams = [])
     {
         $aMethod = ['name' => 'getAll', 'params' => [0 => 'query']];
@@ -33,6 +56,13 @@ class BxDolBackgroundJobsQuery extends BxDolDb
                 $sWhereClause = " AND `name`=:name";
                 break;
 
+            case 'running':
+                $aMethod['name'] = 'getAllWithKey';
+                $aMethod['params'][1] = 'id';
+
+                $sWhereClause = " AND `status`='" . BX_DOL_BG_JOBS_STATUS_PROCESSING . "'";
+                break;
+
             case 'process':
                 $aMethod['name'] = 'getAllWithKey';
                 $aMethod['params'][1] = 'id';

install/sql/system.sql

@@ -465,7 +465,6 @@ INSERT INTO `sys_options`(`category_id`, `name`, `caption`, `value`, `type`, `ex
 (@iCategoryId, 'sys_std_pgt_etemplates_editor', '_adm_stg_cpt_option_sys_std_pgt_etemplates_editor', 'on', 'checkbox', '', '', '', '', 290),
 
 (@iCategoryId, 'sys_bg_jobs_process_per_run', '_adm_stg_cpt_option_sys_bg_jobs_process_per_run', '5', 'digit', '', '', '', '', 300),
-(@iCategoryId, 'sys_bg_jobs_workers', '_adm_stg_cpt_option_sys_bg_jobs_workers', '0', 'digit', '', '', '', '', 302),
 (@iCategoryId, 'sys_bg_jobs_workers_limit', '_adm_stg_cpt_option_sys_bg_jobs_workers_limit', '3', 'digit', '', '', '', '', 304),
 (@iCategoryId, 'sys_bg_jobs_cleanup_timeout', '_adm_stg_cpt_option_sys_bg_jobs_cleanup_timeout', '30', 'digit', '', '', '', '', 306),
 
@@ -2654,6 +2653,7 @@ CREATE TABLE `sys_recommendation_data` (
 CREATE TABLE `sys_background_jobs` (
   `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
   `name` varchar(128) NOT NULL default '',
+  `claim_token` varchar(64) NOT NULL default '',
   `added` int(11) unsigned NOT NULL default '0',
   `reserved_at` int(11) unsigned NOT NULL default '0',
   `available_at` int(11) unsigned NOT NULL default '0',
@@ -2663,7 +2663,8 @@ CREATE TABLE `sys_background_jobs` (
   `error` varchar(255) NOT NULL default '',
   `status` varchar(16) NOT NULL default 'pending',
   PRIMARY KEY (`id`),
-  UNIQUE KEY `name` (`name`)
+  UNIQUE KEY `name` (`name`),
+  KEY `claim_token` (`claim_token`)
 );
 
 -- --------------------------------------------------------

modules/boonex/english/data/langs/system/en.xml

@@ -1905,7 +1905,6 @@
     <string name="_adm_stg_cpt_option_sys_std_pgt_etemplates_editor"><![CDATA[Studio Polyglot: Enable visual editor]]></string>
 
     <string name="_adm_stg_cpt_option_sys_bg_jobs_process_per_run"><![CDATA[Background jobs: Number of jobs to process per run]]></string>
-    <string name="_adm_stg_cpt_option_sys_bg_jobs_workers"><![CDATA[Background jobs: Number of active parallel workers]]></string>
     <string name="_adm_stg_cpt_option_sys_bg_jobs_workers_limit"><![CDATA[Background jobs: Max number of parallel workers]]></string>
     <string name="_adm_stg_cpt_option_sys_bg_jobs_cleanup_timeout"><![CDATA[Background jobs: Cleanup jobs with errors after (in days)]]></string>
 

modules/boonex/russian/data/langs/system/ru.xml

@@ -1880,7 +1880,6 @@
     <string name="_adm_stg_cpt_option_sys_std_pgt_etemplates_editor"><![CDATA[Полиглот в Студии: Включить визуальный редактор]]></string>
 
     <string name="_adm_stg_cpt_option_sys_bg_jobs_process_per_run"><![CDATA[Фоновые задачи: Количество заданий для обработки за один запуск]]></string>
-    <string name="_adm_stg_cpt_option_sys_bg_jobs_workers"><![CDATA[Фоновые задачи: Количество активных параллельных работников]]></string>
     <string name="_adm_stg_cpt_option_sys_bg_jobs_workers_limit"><![CDATA[Фоновые задачи: Максимальное количество параллельных работников]]></string>
     <string name="_adm_stg_cpt_option_sys_bg_jobs_cleanup_timeout"><![CDATA[Фоновые задачи: Удаление заданий с ошибками через (в днях)]]></string>