From d491489366a57a894ec8170ff821e8ec28107bee Mon Sep 17 00:00:00 2001 From: Adam Heinz Date: Wed, 26 Feb 2025 09:46:48 -0500 Subject: [PATCH] [IMP] queue_job: Default subchannel capacity and sequential. This adds `capacity_default` and `sequential_default` options that set `capacity` and `sequential` for autocreated subchannels. It also allows non-root channel configurations to omit capacity. --- queue_job/jobrunner/channels.py | 88 ++++++++++++++++++++----- queue_job/tests/test_runner_channels.py | 3 +- queue_job/tests/test_runner_runner.py | 3 +- 3 files changed, 77 insertions(+), 17 deletions(-) diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index e3cb480420..7d6642dab8 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -408,7 +408,15 @@ class Channel: """ def __init__( - self, name, parent, capacity=None, sequential=False, throttle=0, paused=False + self, + name, + parent, + capacity=None, + sequential=False, + throttle=0, + paused=False, + capacity_default=None, + sequential_default=False, ): self.name = name self.parent = parent @@ -419,9 +427,13 @@ def __init__( self._running = set() self._failed = set() self._pause_until = 0 # utc seconds since the epoch - self.capacity = capacity + self.capacity = ( + capacity if (capacity is not None) else (parent and parent.capacity_default) + ) + self.capacity_default = capacity_default self.throttle = throttle # seconds - self.sequential = sequential + self.sequential = sequential or (parent and parent.sequential_default) + self.sequential_default = sequential_default self.paused = paused @property @@ -438,13 +450,17 @@ def configure(self, config): Supported keys are: * capacity + * capacity_default (default for sub channels) * sequential + * sequential_default (default for sub channels) * throttle * paused """ assert self.fullname.endswith(config["name"]) self.capacity = config.get("capacity", None) + self.capacity_default = config.get("capacity_default", None) self.sequential = bool(config.get("sequential", False)) + self.sequential_default = config.get("sequential_default", False) self.throttle = int(config.get("throttle", 0)) self.paused = bool(config.get("paused", False)) if self.sequential and self.capacity != 1: @@ -902,22 +918,23 @@ def parse_simple_config(cls, config_string): continue config = {} config_items = split_strip(channel_config_string, ":") - name = config_items[0] - if not name: + if not (name := config_items.pop(0)): raise ValueError( f"Invalid channel config {config_string}: missing channel name" ) config["name"] = name - if len(config_items) > 1: - capacity = config_items[1] + if len(config_items) > 0: try: - config["capacity"] = int(capacity) + config["capacity"] = int(config_items[0]) + config_items.pop(0) except Exception as ex: - raise ValueError( - f"Invalid channel config {config_string}: " - f"invalid capacity {capacity}" - ) from ex - for config_item in config_items[2:]: + if name == "root": + raise ValueError( + f"Invalid channel config {config_string}: " + f"invalid capacity {config_items[0]}" + ) from ex + + for config_item in config_items: kv = split_strip(config_item, "=") if len(kv) == 1: k, v = kv[0], True @@ -932,7 +949,16 @@ def parse_simple_config(cls, config_string): raise ValueError( f"Invalid channel config {config_string}: duplicate key {k}" ) - config[k] = v + if k == "capacity_default": + try: + config[k] = int(v) + except Exception as ex: + raise ValueError( + f"Invalid channel config {config_string}: " + f"invalid capacity_default {v}" + ) from ex + else: + config[k] = v else: config["capacity"] = 1 res.append(config) @@ -945,6 +971,17 @@ def simple_configure(self, config_string): >>> c = cm.get_channel_by_name('root') >>> c.capacity 1 + + >>> cm.simple_configure('root:bogus') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:bogus: invalid capacity bogus + + >>> cm.simple_configure('root:4,:2') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:4,:2: missing channel name + >>> cm.simple_configure('root:4,autosub.sub:2,seq:1:sequential') >>> cm.get_channel_by_name('root').capacity 4 @@ -961,7 +998,28 @@ def simple_configure(self, config_string): 1 >>> cm.get_channel_by_name('seq').sequential True - """ + + >>> cm.simple_configure('root:4:capacity_default=bogus') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:4:capacity_default=bogus: invalid capacity_default bogus + + >>> cm.simple_configure('root:4,sub:3:capacity_default=2') + >>> cm.get_channel_by_name('root.sub').capacity + 3 + >>> cm.get_channel_by_name('root.sub.auto', autocreate=True).capacity + 2 + + >>> cm.simple_configure('root:4,seq:2:sequential') + Traceback (most recent call last): + ... + ValueError: A sequential channel must have a capacity of 1 + + >>> cm.simple_configure('root:4,seq:sequential_default') + >>> cm.get_channel_by_name('root.seq.auto', autocreate=True).sequential + True + + """ # noqa: E501 for config in ChannelManager.parse_simple_config(config_string): self.get_channel_from_config(config) diff --git a/queue_job/tests/test_runner_channels.py b/queue_job/tests/test_runner_channels.py index b00dd26261..313e350bd3 100644 --- a/queue_job/tests/test_runner_channels.py +++ b/queue_job/tests/test_runner_channels.py @@ -12,6 +12,7 @@ @tagged("doctest") class TestDoctest(BaseCase): def test_doctest(self): - doctest.testmod( + results = doctest.testmod( channels, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE ) + self.assertEqual(results.failed, 0, "doctest failed") diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 1adfa790dd..04c3fc1946 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -13,9 +13,10 @@ @tagged("doctest") class TestDoctest(BaseCase): def test_doctest(self): - doctest.testmod( + results = doctest.testmod( runner, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE ) + self.assertEqual(results.failed, 0, "doctest failed") @tagged("-at_install", "post_install")