From 4971da0bfa94b4c8ce01cd1ff7924c8e38766802 Mon Sep 17 00:00:00 2001 From: Brendan Heywood Date: Thu, 17 Oct 2019 11:20:55 +1100 Subject: [PATCH] MDL-67363 task: Add adhoc task quality of service balancing --- lib/classes/task/manager.php | 71 ++++++++++++++ lib/tests/task_manager_test.php | 166 ++++++++++++++++++++++++++++++++ 2 files changed, 237 insertions(+) create mode 100644 lib/tests/task_manager_test.php diff --git a/lib/classes/task/manager.php b/lib/classes/task/manager.php index 87fe34287ac..f81ca5ea086 100644 --- a/lib/classes/task/manager.php +++ b/lib/classes/task/manager.php @@ -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)) { diff --git a/lib/tests/task_manager_test.php b/lib/tests/task_manager_test.php new file mode 100644 index 00000000000..cb99cf68211 --- /dev/null +++ b/lib/tests/task_manager_test.php @@ -0,0 +1,166 @@ +. + +/** + * This file contains the unit tests for the task manager. + * + * @package core + * @copyright 2019 Brendan Heywood + * @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 + * @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); + } + +} +