[IMP] queue_job: Configure default subchannel capacity.

This adds a new `subcapacity` option to channels that allows the
configuration of the default capacity for autocreated child channels.

For example, environment `ODOO_QUEUE_JOB_CHANNELS=root:8:subcapacity=1`
would set the capacity of an autocreated `root.sub` channel to 1.

Author: amh-mw

queue_job/jobrunner/channels.py

@@ -404,7 +404,15 @@ 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,
+        def_sub_capacity=None,
+    ):
         self.name = name
         self.parent = parent
         if self.parent:
@@ -414,9 +422,10 @@ 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.capacity = capacity or (parent and parent.def_sub_capacity)
         self.throttle = throttle  # seconds
         self.sequential = sequential
+        self.def_sub_capacity = def_sub_capacity
 
     @property
     def sequential(self):
@@ -433,11 +442,13 @@ def configure(self, config):
 
         * capacity
         * sequential
+        * def_sub_capacity
         * throttle
         """
         assert self.fullname.endswith(config["name"])
         self.capacity = config.get("capacity", None)
         self.sequential = bool(config.get("sequential", False))
+        self.def_sub_capacity = config.get("def_sub_capacity", None)
         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")
@@ -897,7 +908,16 @@ def parse_simple_config(cls, config_string):
                             f"Invalid channel config {config_string}: "
                             f"duplicate key {k}"
                         )
-                    config[k] = v
+                    if k == "def_sub_capacity":
+                        try:
+                            config[k] = int(v)
+                        except Exception as ex:
+                            raise ValueError(
+                                f"Invalid channel config {config_string}: "
+                                f"invalid def_sub_capacity {v}"
+                            ) from ex
+                    else:
+                        config[k] = v
             else:
                 config["capacity"] = 1
             res.append(config)
@@ -910,16 +930,20 @@ def simple_configure(self, config_string):
         >>> c = cm.get_channel_by_name('root')
         >>> c.capacity
         1
-        >>> cm.simple_configure('root:4,autosub.sub:2,seq:1:sequential')
+        >>> cm.simple_configure('root:4,defsub:def_sub_capacity:3,autosub.sub:2,seq:1:sequential')
         >>> cm.get_channel_by_name('root').capacity
         4
         >>> cm.get_channel_by_name('root').sequential
         False
+        >>> cm.get_channel_by_name('root').def_sub_capacity
         >>> cm.get_channel_by_name('root.autosub').capacity
+        >>> cm.get_channel_by_name('root.autosub').def_sub_capacity
+        3
         >>> cm.get_channel_by_name('root.autosub.sub').capacity
         2
         >>> cm.get_channel_by_name('root.autosub.sub').sequential
         False
+        >>> cm.get_channel_by_name('root.autosub.sub').def_sub_capacity
         >>> cm.get_channel_by_name('autosub.sub').capacity
         2
         >>> cm.get_channel_by_name('seq').capacity