Skip to content

Commit f23b8d5

Browse files
committed
[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.
1 parent d198792 commit f23b8d5

1 file changed

Lines changed: 81 additions & 15 deletions

File tree

queue_job/jobrunner/channels.py

Lines changed: 81 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -404,7 +404,16 @@ class Channel:
404404
without risking to overflow the system.
405405
"""
406406

407-
def __init__(self, name, parent, capacity=None, sequential=False, throttle=0):
407+
def __init__(
408+
self,
409+
name,
410+
parent,
411+
capacity=None,
412+
sequential=False,
413+
throttle=0,
414+
capacity_default=None,
415+
sequential_default=None,
416+
):
408417
self.name = name
409418
self.parent = parent
410419
if self.parent:
@@ -414,9 +423,13 @@ def __init__(self, name, parent, capacity=None, sequential=False, throttle=0):
414423
self._running = set()
415424
self._failed = set()
416425
self._pause_until = 0 # utc seconds since the epoch
417-
self.capacity = capacity
426+
self.sequential = sequential or (parent and parent.sequential_default)
427+
self.sequential_default = sequential_default
428+
self.capacity = (
429+
(self.sequential and 1) or capacity or (parent and parent.capacity_default)
430+
)
431+
self.capacity_default = capacity_default
418432
self.throttle = throttle # seconds
419-
self.sequential = sequential
420433

421434
@property
422435
def sequential(self):
@@ -432,12 +445,16 @@ def configure(self, config):
432445
Supported keys are:
433446
434447
* capacity
448+
* capacity_default
435449
* sequential
450+
* sequential_default
436451
* throttle
437452
"""
438453
assert self.fullname.endswith(config["name"])
439454
self.capacity = config.get("capacity", None)
455+
self.capacity_default = config.get("capacity_default", None)
440456
self.sequential = bool(config.get("sequential", False))
457+
self.sequential_default = config.get("sequential_default", False)
441458
self.throttle = int(config.get("throttle", 0))
442459
if self.sequential and self.capacity != 1:
443460
raise ValueError("A sequential channel must have a capacity of 1")
@@ -866,22 +883,23 @@ def parse_simple_config(cls, config_string):
866883
continue
867884
config = {}
868885
config_items = split_strip(channel_config_string, ":")
869-
name = config_items[0]
870-
if not name:
886+
if not (name := config_items.pop(0)):
871887
raise ValueError(
872888
f"Invalid channel config {config_string}: missing channel name"
873889
)
874890
config["name"] = name
875-
if len(config_items) > 1:
876-
capacity = config_items[1]
891+
if len(config_items) > 0:
877892
try:
878-
config["capacity"] = int(capacity)
893+
config["capacity"] = int(config_items[0])
894+
config_items.pop(0)
879895
except Exception as ex:
880-
raise ValueError(
881-
f"Invalid channel config {config_string}: "
882-
f"invalid capacity {capacity}"
883-
) from ex
884-
for config_item in config_items[2:]:
896+
if name == "root":
897+
raise ValueError(
898+
f"Invalid channel config {config_string}: "
899+
f"invalid capacity {config_items[0]}"
900+
) from ex
901+
902+
for config_item in config_items:
885903
kv = split_strip(config_item, "=")
886904
if len(kv) == 1:
887905
k, v = kv[0], True
@@ -897,7 +915,16 @@ def parse_simple_config(cls, config_string):
897915
f"Invalid channel config {config_string}: "
898916
f"duplicate key {k}"
899917
)
900-
config[k] = v
918+
if k == "capacity_default":
919+
try:
920+
config[k] = int(v)
921+
except Exception as ex:
922+
raise ValueError(
923+
f"Invalid channel config {config_string}: "
924+
f"invalid capacity_default {v}"
925+
) from ex
926+
else:
927+
config[k] = v
901928
else:
902929
config["capacity"] = 1
903930
res.append(config)
@@ -910,6 +937,17 @@ def simple_configure(self, config_string):
910937
>>> c = cm.get_channel_by_name('root')
911938
>>> c.capacity
912939
1
940+
941+
>>> cm.simple_configure('root:bogus')
942+
Traceback (most recent call last):
943+
...
944+
ValueError: Invalid channel config root:bogus: invalid capacity bogus
945+
946+
>>> cm.simple_configure('root:4,:2')
947+
Traceback (most recent call last):
948+
...
949+
ValueError: Invalid channel config root:4,:2: missing channel name
950+
913951
>>> cm.simple_configure('root:4,autosub.sub:2,seq:1:sequential')
914952
>>> cm.get_channel_by_name('root').capacity
915953
4
@@ -926,7 +964,35 @@ def simple_configure(self, config_string):
926964
1
927965
>>> cm.get_channel_by_name('seq').sequential
928966
True
929-
"""
967+
968+
`capacity_default`
969+
>>> cm.simple_configure('root:4:capacity_default=bogus')
970+
Traceback (most recent call last):
971+
...
972+
ValueError: Invalid channel config root:4:capacity_default=bogus: invalid capacity_default bogus
973+
974+
>>> cm.simple_configure('root:4,sub:3:capacity_default=2')
975+
>>> cm.get_channel_by_name('root.sub').capacity
976+
3
977+
>>> cm.get_channel_by_name('root.sub.auto', autocreate=True).capacity
978+
2
979+
980+
`sequential`
981+
>>> cm.simple_configure('root:4,seq:2:sequential')
982+
Traceback (most recent call last):
983+
...
984+
ValueError: A sequential channel must have a capacity of 1
985+
986+
`sequential_default`
987+
988+
>>> cm.simple_configure('root:4,seq:sequential_default')
989+
>>> cm.get_channel_by_name('root.seq').capacity
990+
>>> cm.get_channel_by_name('root.seq.auto', autocreate=True).capacity
991+
1
992+
>>> cm.get_channel_by_name('root.seq.auto', autocreate=True).sequential
993+
True
994+
995+
""" # noqa: E501
930996
for config in ChannelManager.parse_simple_config(config_string):
931997
self.get_channel_from_config(config)
932998

0 commit comments

Comments
 (0)