From 1d0b74b2a36f563bf60eb2b15660e9aaa94682e4 Mon Sep 17 00:00:00 2001 From: Mikhail Golenkov Date: Thu, 2 Jan 2020 16:05:47 +1100 Subject: [PATCH] MDL-64610 task: Add support for per-task concurrency limits. --- lib/classes/task/adhoc_task.php | 50 ++++++++++++++++++++++++++++++++ lib/classes/task/manager.php | 51 +++++++++++++++++++++++++++++---- lib/cronlib.php | 2 +- lib/tests/adhoc_task_test.php | 34 ++++++++++++++++++++++ 4 files changed, 131 insertions(+), 6 deletions(-) diff --git a/lib/classes/task/adhoc_task.php b/lib/classes/task/adhoc_task.php index 2e1e042d41b..bd3e8afa4a6 100644 --- a/lib/classes/task/adhoc_task.php +++ b/lib/classes/task/adhoc_task.php @@ -43,6 +43,9 @@ abstract class adhoc_task extends task_base { /** @var integer|null $userid - Adhoc tasks may choose to run as a specific user. */ private $userid = null; + /** @var \core\lock\lock The concurrency task lock for this task. */ + private $concurrencylock = null; + /** * Setter for $id. * @param int|null $id @@ -107,4 +110,51 @@ abstract class adhoc_task extends task_base { $this->userid = $userid; } + /** + * Returns default concurrency limit for this task. + * + * @return int default concurrency limit + */ + protected function get_default_concurrency_limit(): int { + global $CFG; + + if (isset($CFG->task_concurrency_limit_default)) { + return (int) $CFG->task_concurrency_limit_default; + } + return 0; + } + + /** + * Returns effective concurrency limit for this task. + * + * @return int effective concurrency limit for this task + */ + final public function get_concurrency_limit(): int { + global $CFG; + + $classname = get_class($this); + + if (isset($CFG->task_concurrency_limit[$classname])) { + return (int) $CFG->task_concurrency_limit[$classname]; + } + return $this->get_default_concurrency_limit(); + } + + /** + * Sets concurrency task lock. + * + * @param \core\lock\lock $lock concurrency lock to be set + */ + final public function set_concurrency_lock(\core\lock\lock $lock): void { + $this->concurrencylock = $lock; + } + + /** + * Release the concurrency lock for this task type. + */ + final public function release_concurrency_lock(): void { + if ($this->concurrencylock) { + $this->concurrencylock->release(); + } + } } diff --git a/lib/classes/task/manager.php b/lib/classes/task/manager.php index 091974bfb07..4c573e89f45 100644 --- a/lib/classes/task/manager.php +++ b/lib/classes/task/manager.php @@ -552,9 +552,10 @@ class manager { * {@link adhoc_task_failed} or {@link adhoc_task_complete} to release the lock and reschedule the task. * * @param int $timestart + * @param bool $checklimits Should we check limits? * @return \core\task\adhoc_task or null if not found */ - public static function get_next_adhoc_task($timestart) { + public static function get_next_adhoc_task($timestart, $checklimits = true) { global $DB; $where = '(nextruntime IS NULL OR nextruntime < :timestart1)'; @@ -568,10 +569,16 @@ class manager { throw new \moodle_exception('locktimeout'); } + $skipclasses = array(); + foreach ($records as $record) { + if (in_array($record->classname, $skipclasses)) { + // Skip the task if it can't be started due to per-task concurrency limit. + continue; + } + if ($lock = $cronlockfactory->get_lock('adhoc_' . $record->id, 0)) { - $classname = '\\' . $record->classname; // Safety check, see if the task has been already processed by another cron run. $record = $DB->get_record('task_adhoc', array('id' => $record->id)); @@ -587,6 +594,19 @@ class manager { continue; } + $tasklimit = $task->get_concurrency_limit(); + if ($checklimits && $tasklimit > 0) { + if ($concurrencylock = self::get_concurrent_task_lock($task)) { + $task->set_concurrency_lock($concurrencylock); + } else { + // Unable to obtain a concurrency lock. + mtrace("Skipping $record->classname adhoc task class as the per-task limit of $tasklimit is reached."); + $skipclasses[] = $record->classname; + $lock->release(); + continue; + } + } + $task->set_lock($lock); if (!$task->is_blocking()) { $cronlock->release(); @@ -691,13 +711,13 @@ class manager { $delay = 86400; } - $classname = self::get_canonical_class_name($task); - + // Reschedule and then release the locks. $task->set_next_run_time(time() + $delay); $task->set_fail_delay($delay); $record = self::record_from_adhoc_task($task); $DB->update_record('task_adhoc', $record); + $task->release_concurrency_lock(); if ($task->is_blocking()) { $task->get_cron_lock()->release(); } @@ -721,7 +741,8 @@ class manager { // Delete the adhoc task record - it is finished. $DB->delete_records('task_adhoc', array('id' => $task->get_id())); - // Reschedule and then release the locks. + // Release the locks. + $task->release_concurrency_lock(); if ($task->is_blocking()) { $task->get_cron_lock()->release(); } @@ -858,4 +879,24 @@ class manager { } return $classname; } + + /** + * Gets the concurrent lock required to run an adhoc task. + * + * @param adhoc_task $task The task to obtain the lock for + * @return \core\lock\lock The lock if one was obtained successfully + * @throws \coding_exception + */ + protected static function get_concurrent_task_lock(adhoc_task $task): ?\core\lock\lock { + $adhoclock = null; + $cronlockfactory = \core\lock\lock_config::get_lock_factory(get_class($task)); + + for ($run = 0; $run < $task->get_concurrency_limit(); $run++) { + if ($adhoclock = $cronlockfactory->get_lock("concurrent_run_{$run}", 0)) { + return $adhoclock; + } + } + + return null; + } } diff --git a/lib/cronlib.php b/lib/cronlib.php index db0193fe857..4ad08cbbaa6 100644 --- a/lib/cronlib.php +++ b/lib/cronlib.php @@ -168,7 +168,7 @@ function cron_run_adhoc_tasks(int $timenow, $keepalive = 0, $checklimits = true) break; } - $task = \core\task\manager::get_next_adhoc_task(time()); + $task = \core\task\manager::get_next_adhoc_task(time(), $checklimits); if ($task) { if ($waiting) { diff --git a/lib/tests/adhoc_task_test.php b/lib/tests/adhoc_task_test.php index c5d796102f8..f90f238bd55 100644 --- a/lib/tests/adhoc_task_test.php +++ b/lib/tests/adhoc_task_test.php @@ -362,4 +362,38 @@ class core_adhoc_task_testcase extends advanced_testcase { $this->assertEquals($user->id, $task->get_userid()); } + + /** + * Test get_concurrency_limit() method to return 0 by default. + */ + public function test_get_concurrency_limit() { + $this->resetAfterTest(true); + $task = new \core\task\adhoc_test_task(); + $concurrencylimit = $task->get_concurrency_limit(); + $this->assertEquals(0, $concurrencylimit); + } + + /** + * Test get_concurrency_limit() method to return a default value set in config. + */ + public function test_get_concurrency_limit_default() { + $this->resetAfterTest(true); + set_config('task_concurrency_limit_default', 10); + $task = new \core\task\adhoc_test_task(); + $concurrencylimit = $task->get_concurrency_limit(); + $this->assertEquals(10, $concurrencylimit); + } + + /** + * Test get_concurrency_limit() method to return a value for specific task class. + */ + public function test_get_concurrency_limit_for_task() { + global $CFG; + $this->resetAfterTest(true); + set_config('task_concurrency_limit_default', 10); + $CFG->task_concurrency_limit = array('core\task\adhoc_test_task' => 5); + $task = new \core\task\adhoc_test_task(); + $concurrencylimit = $task->get_concurrency_limit(); + $this->assertEquals(5, $concurrencylimit); + } }