PR#2370: distgit: tasks in different sandboxes to have the same priority - copr/copr

copr / copr

#2370 distgit: tasks in different sandboxes to have the same priority

Merged a year ago by praiskup. Opened a year ago by praiskup.

Unknown source distgit-prio into main

distgit: fair processing of task from multiple sandboxes

Pavel Raiskup • a year ago

7431d5e

dist-git/copr-dist-git.spec

file modified

		`@@ -23,6 +23,7 @@`
		`BuildRequires: python3-pytest`
		`BuildRequires: python3-copr-common >= %copr_common_version`
		`BuildRequires: python3-oslo-concurrency`
		`+ BuildRequires: python3-redis`
		`BuildRequires: python3-setproctitle`

		`Recommends: logrotate`

dist-git/copr_dist_git/import_dispatcher.py

file modified

+17

		`@@ -17,6 +17,20 @@`
		`}`


		`+ class _PriorityCounter:`
		`+ def __init__(self):`
		`+ self._counter = {}`
		`+`
		`+ def get_priority(self, task):`
		`+ """`
		`+ Calculate the "dynamic" import task priority.`
		`+ _counter["sandbox"] = value`
		`+ """`
		`+ self._counter.setdefault(task.sandbox, 0)`
		`+ self._counter[task.sandbox] += 1`
		`+ return self._counter[task.sandbox]`
		`+`
		`+`
		`class ImportDispatcher(Dispatcher):`
		`"""`
		`Kick-off a dispatcher daemon for importing tasks into DistGit.`
		`@@ -45,6 +59,9 @@`
		`def get_frontend_tasks(self):`
		`importer = Importer(self.opts)`
		`tasks = importer.try_to_obtain_new_tasks(limit=999999)`
		`+ counter = _PriorityCounter()`
		`+ for task in tasks:`
		`+ task.dispatcher_priority += counter.get_priority(task)`
		`return tasks`

		`def _create_per_task_logs_directory(self, path):`

dist-git/copr_dist_git/import_task.py

file modified

+3 -1

		`@@ -14,6 +14,7 @@`
		`self.srpm_url = None`
		`self.sandbox = None`
		`self.background = None`
		`+ self.dispatcher_priority = 0`

		`@staticmethod`
		`def from_dict(task_dict):`
		`@@ -39,7 +40,8 @@`

		`@property`
		`def priority(self):`
		`- return 100 if self.background else 0`
		`+ value = 100 if self.background else 0`
		`+ return value + self.dispatcher_priority`

		`@property`
		`def repo_namespace(self):`

dist-git/tests/base.py

file modified

+7 -2

		`@@ -1,16 +1,18 @@`
		`import os`
		`import shutil`
		`import tempfile`
		`- import munch`
		`import time`
		`- import json`
		`+`
		`+ import munch`

		`from copr_dist_git import importer`
		`from copr_dist_git import import_task`
		`+ from copr_dist_git import import_dispatcher`

		`class Base(object):`

		`def setup_method(self, method):`
		`+ # pylint: disable=attribute-defined-outside-init`
		`self.tmp_dir_name = self.make_temp_dir()`
		`self.lookaside_location = os.path.join(self.tmp_dir_name, "lookaside")`
		`self.per_task_location = os.path.join(self.tmp_dir_name, "per-task-logs")`
		`@@ -30,6 +32,7 @@`
		`"multiple_threads": True,`
		`"git_user_name": "Test user",`
		`"git_user_email": "test@test.org",`
		`+ "max_workers": 10,`
		`})`

		`self.importer = importer.Importer(self.opts)`
		`@@ -67,6 +70,8 @@`

		`self.url_task = import_task.ImportTask.from_dict(self.url_task_data)`
		`self.upload_task = import_task.ImportTask.from_dict(self.upload_task_data)`
		`+ self.dispatcher = import_dispatcher.ImportDispatcher(self.opts)`
		`+ self.dispatcher.importer = self.importer`

		`def teardown_method(self, method):`
		`self.rm_tmp_dir()`

dist-git/tests/test_importer.py

file modified

+25

		`@@ -1,5 +1,6 @@`
		`# coding: utf-8`

		`+ from collections import defaultdict`
		`import json`

		`import os`
		`@@ -12,6 +13,7 @@`
		`from unittest import mock`
		`from unittest.mock import MagicMock`

		`+ import copr_dist_git.import_task`

		`MODULE_REF = 'copr_dist_git.importer'`

		`@@ -124,6 +126,29 @@`
		`'pkg_version': '1.2', 'git_hash': '124', 'repo_name': 'foo'})`
		`]))`

		`+`
		`+ @mock.patch("copr_dist_git.import_dispatcher.Importer", return_value=MagicMock())`
		`+ def test_priorities(self, importer):`
		`+ importer.return_value = self.importer`
		`+ def _shortener(the_dict):`
		`+ return copr_dist_git.import_task.ImportTask.from_dict(`
		`+ defaultdict(lambda: "notset", the_dict))`
		`+ self.importer.try_to_obtain_new_tasks = MagicMock()`
		`+ self.importer.try_to_obtain_new_tasks.return_value = [`
		`+ _shortener({"build_id": 1, "sandbox": "a", "background": False}),`
		`+ _shortener({"build_id": 2, "sandbox": "a", "background": False}),`
		`+ _shortener({"build_id": 3, "sandbox": "b", "background": False}),`
		`+ _shortener({"build_id": 3, "sandbox": "c", "background": False}),`
		`+ _shortener({"build_id": 1, "sandbox": "a", "background": True}),`
		`+ ]`
		`+ tasks = self.dispatcher.get_frontend_tasks()`
		`+ assert tasks[0].priority == 1`
		`+ assert tasks[1].priority == 2`
		`+ assert tasks[2].priority == 1`
		`+ assert tasks[3].priority == 1`
		`+ assert tasks[4].priority == 103`
		`+`
		`+`
		`def test_run(self, mc_time, mc_worker):`
		`self.importer.try_to_obtain_new_tasks = MagicMock()`
		`self.importer.do_import = MagicMock()`

praiskup commented a year ago

distgit: fair processing of task from multiple sandboxes

Even users coming "late" should be able to process at least some
importing tasks quickly, and outrun the large queue generated by other
users.  This is guaranteed by giving the most priority (priority=1) to
the first task from each sandbox.  Next priority (2) is given to the
second task from each sandbox, and so on.

This means that no matter how big the queue is, "my first import task"
(excluding background jobs!) gets always the highest priority — and will
be processed relatively quickly.

Merges: #2370