L-electron-Rare/moodle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MDL-64610 task: Add support for per-task concurrency limits.

Browse Source
This commit is contained in:
Mikhail Golenkov
2020-01-02 16:05:47 +11:00
parent 7850f76a93
commit 1d0b74b2a3
4 changed files with 131 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/classes/task/adhoc_task.php
+50
View File
@@ -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();
}
}
}
lib/classes/task/manager.php
+46 -5
View File
@@ -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;
}
}
lib/cronlib.php
+1 -1
View File
@@ -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) {
lib/tests/adhoc_task_test.php
+34
View File
@@ -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);
}
}