MDL-67363 task: Add adhoc task quality of service balancing
This commit is contained in:
@@ -471,6 +471,75 @@ class manager {
|
||||
return $tasks;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensure quality of service for the ad hoc task queue.
|
||||
*
|
||||
* This reshuffles the adhoc tasks queue to balance by type to ensure a
|
||||
* level of quality of service per type, while still maintaining the
|
||||
* relative order of tasks queued by timestamp.
|
||||
*
|
||||
* @param array $records array of task records
|
||||
* @return void
|
||||
*/
|
||||
public static function ensure_adhoc_task_qos(array $records): array {
|
||||
|
||||
$count = count($records);
|
||||
if ($count == 0) {
|
||||
return $records;
|
||||
}
|
||||
|
||||
$queues = []; // This holds a queue for each type of adhoc task.
|
||||
$limits = []; // The relative limits of each type of task.
|
||||
$limittotal = 0;
|
||||
|
||||
// Split the single queue up into queues per type.
|
||||
foreach ($records as $record) {
|
||||
$type = $record->classname;
|
||||
if (!array_key_exists($type, $queues)) {
|
||||
$queues[$type] = [];
|
||||
}
|
||||
if (!array_key_exists($type, $limits)) {
|
||||
$limits[$type] = 1;
|
||||
$limittotal += 1;
|
||||
}
|
||||
$queues[$type][] = $record;
|
||||
}
|
||||
|
||||
$qos = []; // Our new queue with ensured quality of service.
|
||||
$seed = $count % $limittotal; // Which task queue to shuffle from first?
|
||||
|
||||
do {
|
||||
$shuffled = 0;
|
||||
|
||||
// Now cycle through task type queues and interleaving the tasks
|
||||
// back into a single queue.
|
||||
foreach ($limits as $type => $limit) {
|
||||
|
||||
// Just interleaving the queue is not enough, because after
|
||||
// any task is processed the whole queue is rebuilt again. So
|
||||
// we need to deterministically start on different types of
|
||||
// tasks so that *on average* we rotate through each type of task.
|
||||
//
|
||||
// We achieve this by using a $seed to start moving tasks off a
|
||||
// different queue each time. The seed is based on the task count
|
||||
// modulo the number of types of tasks on the queue. As we count
|
||||
// down this naturally cycles through each type of record.
|
||||
if ($seed < 1) {
|
||||
$shuffled = 1;
|
||||
$seed += 1;
|
||||
continue;
|
||||
}
|
||||
$task = array_splice($queues[$type], 0, 1);
|
||||
$qos = array_merge($qos, $task);
|
||||
|
||||
// Stop if we didn't move any tasks onto the main queue.
|
||||
$shuffled += count($task);
|
||||
}
|
||||
} while ($shuffled > 0);
|
||||
|
||||
return $qos;
|
||||
}
|
||||
|
||||
/**
|
||||
* This function will dispatch the next adhoc task in the queue. The task will be handed out
|
||||
* with an open lock - possibly on the entire cron process. Make sure you call either
|
||||
@@ -491,6 +560,8 @@ class manager {
|
||||
$params = array('timestart1' => $timestart);
|
||||
$records = $DB->get_records_select('task_adhoc', $where, $params);
|
||||
|
||||
$records = self::ensure_adhoc_task_qos($records);
|
||||
|
||||
foreach ($records as $record) {
|
||||
|
||||
if ($lock = $cronlockfactory->get_lock('adhoc_' . $record->id, 0)) {
|
||||
|
||||
@@ -0,0 +1,166 @@
|
||||
<?php
|
||||
// This file is part of Moodle - http://moodle.org/
|
||||
//
|
||||
// Moodle 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.
|
||||
//
|
||||
// Moodle 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 Moodle. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
/**
|
||||
* This file contains the unit tests for the task manager.
|
||||
*
|
||||
* @package core
|
||||
* @copyright 2019 Brendan Heywood <[email protected]>
|
||||
* @license http://www.gnu.org/copyleft/gpl.html GNU GPL v3 or later
|
||||
*/
|
||||
|
||||
defined('MOODLE_INTERNAL') || die();
|
||||
|
||||
/**
|
||||
* This file contains the unit tests for the task manager.
|
||||
*
|
||||
* @copyright 2019 Brendan Heywood <[email protected]>
|
||||
* @license http://www.gnu.org/copyleft/gpl.html GNU GPL v3 or later
|
||||
*/
|
||||
class core_task_manager_testcase extends advanced_testcase {
|
||||
|
||||
public function test_ensure_adhoc_task_qos_provider() {
|
||||
return [
|
||||
[
|
||||
[],
|
||||
[],
|
||||
],
|
||||
// A queue with a lopside initial load that needs to be staggered.
|
||||
[
|
||||
[
|
||||
(object)['id' => 1, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 2, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 3, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 4, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
],
|
||||
[
|
||||
(object)['id' => 1, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 2, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 3, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 4, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
],
|
||||
],
|
||||
// The same lopsided queue but now the first item is gone.
|
||||
[
|
||||
[
|
||||
(object)['id' => 2, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 3, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 4, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
],
|
||||
[
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 2, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 3, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 4, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
],
|
||||
],
|
||||
// The same lopsided queue but now the first two items is gone.
|
||||
[
|
||||
[
|
||||
(object)['id' => 3, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 4, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
],
|
||||
[
|
||||
(object)['id' => 3, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 4, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
],
|
||||
],
|
||||
// The same lopsided queue but now the first three items are gone.
|
||||
[
|
||||
[
|
||||
(object)['id' => 4, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
],
|
||||
[
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 4, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
],
|
||||
],
|
||||
[
|
||||
[
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
],
|
||||
[
|
||||
(object)['id' => 5, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 7, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
|
||||
(object)['id' => 6, 'classname' => '\core\task\asynchronous_backup_task'],
|
||||
(object)['id' => 8, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
|
||||
(object)['id' => 9, 'classname' => '\tool_dataprivacy\task\process_data_request_task'],
|
||||
],
|
||||
],
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* Test that the Quality of Service reordering works.
|
||||
*
|
||||
* @dataProvider test_ensure_adhoc_task_qos_provider
|
||||
*
|
||||
* @param array $input array of tasks
|
||||
* @param array $expected array of reordered tasks
|
||||
* @return void
|
||||
*/
|
||||
public function test_ensure_adhoc_task_qos(array $input, array $expected) {
|
||||
$this->resetAfterTest();
|
||||
$result = \core\task\manager::ensure_adhoc_task_qos($input);
|
||||
$this->assertEquals($expected, $result);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user