[18.0][IMP] queue_job: Default subchannel capacity and sequential.

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=False,
+ ):
  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 = (
+ capacity if (capacity is not None) else (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 (default for sub channels)
  * sequential
+ * sequential_default (default for sub channels)
  * 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,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
@@ -926,7 +964,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)
 

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")

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")