[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,
which will default to 1 for sequential channels, and pass through
to the parent channel otherwise.

Author: amh-mw

queue_job/jobrunner/channels.py

@@ -404,7 +404,16 @@ class Channel:
     without risking to overflow the system.
     """
 
-    def __init__(self, name, parent, capacity=None, sequential=False, throttle=0):
+    def __init__(
+        self,
+        name,
+        parent,
+        capacity=None,
+        sequential=False,
+        throttle=0,
+        capacity_default=None,
+        sequential_default=None,
+    ):
         self.name = name
         self.parent = parent
         if self.parent:
@@ -414,9 +423,13 @@ def __init__(self, name, parent, capacity=None, sequential=False, throttle=0):
         self._running = set()
         self._failed = set()
         self._pause_until = 0  # utc seconds since the epoch
-        self.capacity = capacity
+        self.sequential = sequential or (parent and parent.sequential_default)
+        self.sequential_default = sequential_default
+        self.capacity = (
+            (self.sequential and 1) or capacity or (parent and parent.capacity_default)
+        )
+        self.capacity_default = capacity_default
         self.throttle = throttle  # seconds
-        self.sequential = sequential
 
     @property
     def sequential(self):
@@ -432,12 +445,16 @@ def configure(self, config):
         Supported keys are:
 
         * capacity
+        * capacity_default
         * sequential
+        * sequential_default
         * throttle
         """
         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))
         if self.sequential and self.capacity != 1:
             raise ValueError("A sequential channel must have a capacity of 1")
@@ -866,22 +883,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
@@ -897,7 +915,16 @@ def parse_simple_config(cls, config_string):
                             f"Invalid channel config {config_string}: "
                             f"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)
@@ -910,6 +937,12 @@ def simple_configure(self, config_string):
         >>> c = cm.get_channel_by_name('root')
         >>> c.capacity
         1
+
+        >>> 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
@@ -926,6 +959,30 @@ def simple_configure(self, config_string):
         1
         >>> cm.get_channel_by_name('seq').sequential
         True
+
+        `capacity_default`
+
+        >>> 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
+
+        `sequential`
+        >>> cm.simple_configure('root:4,seq:2:sequential')
+        Traceback (most recent call last):
+            ...
+        ValueError: A sequential channel must have a capacity of 1
+
+        `sequential_default`
+
+        >>> cm.simple_configure('root:4,seq:sequential_default')
+        >>> cm.get_channel_by_name('root.seq').capacity
+        >>> cm.get_channel_by_name('root.seq.auto', autocreate=True).capacity
+        1
+        >>> cm.get_channel_by_name('root.seq.auto', autocreate=True).sequential
+        True
+
         """
         for config in ChannelManager.parse_simple_config(config_string):
             self.get_channel_from_config(config)

queue_job/tests/test_runner_channels.py

@@ -12,4 +12,6 @@
 @tagged("doctest")
 class TestDoctest(BaseCase):
     def test_doctest(self):
-        doctest.testmod(channels, exclude_empty=True, raise_on_error=True)
+        doctest.testmod(
+            channels, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE
+        )

queue_job/tests/test_runner_runner.py

@@ -13,7 +13,9 @@
 @tagged("doctest")
 class TestDoctest(BaseCase):
     def test_doctest(self):
-        doctest.testmod(runner, exclude_empty=True, raise_on_error=True)
+        doctest.testmod(
+            runner, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE
+        )
 
 
 @tagged("-at_install", "post_install")