[18.0][ADD] queue_job: per DB channel managers with UI configuration and hot reload

queue_job/jobrunner/__init__.py

@@ -20,7 +20,7 @@
     queue_job_config = config.misc.get("queue_job", {})
 
 
-from .runner import QueueJobRunner, _channels
+from .runner import QueueJobRunner, _channels, _max_capacity
 
 _logger = logging.getLogger(__name__)
 
@@ -87,7 +87,9 @@ def signal_time_expired_handler(self, n, stack):
 
 
 def _is_runner_enabled():
-    return not _channels().strip().startswith("root:0")
+    if _channels().strip().startswith("root:0"):
+        return False
+    return _max_capacity() != 0
 
 
 def _start_runner_thread(server_type):
@@ -100,7 +102,8 @@ def _start_runner_thread(server_type):
         else:
             _logger.info(
                 "jobrunner thread (in %s) NOT started, "
-                "because the root channel's capacity is set to 0",
+                "because the root channel's capacity or the max capacity "
+                "is set to 0",
                 server_type,
             )
 

queue_job/jobrunner/channels.py

@@ -3,19 +3,32 @@
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl.html)
 import logging
 from collections import namedtuple
+from dataclasses import asdict, dataclass
 from functools import total_ordering
 from heapq import heappop, heappush
 from weakref import WeakValueDictionary
 
 from ..exception import ChannelNotFound
 from ..job import CANCELLED, DONE, ENQUEUED, FAILED, PENDING, STARTED, WAIT_DEPENDENCIES
 
+RELOAD_PAYLOAD = "reload"
 NOT_DONE = (WAIT_DEPENDENCIES, PENDING, ENQUEUED, STARTED, FAILED)
 JobSortingKey = namedtuple("SortingKey", "eta priority date_created seq")
 
 _logger = logging.getLogger(__name__)
 
 
+@dataclass
+class ChannelConfig:
+    """Configuration of a channel"""
+
+    name: str
+    capacity: int = 0
+    sequential: bool = False
+    throttle: int = 0
+    paused: bool = False
+
+
 class PriorityQueue:
     """A priority queue that supports removing arbitrary objects.
 
@@ -965,6 +978,11 @@ def simple_configure(self, config_string):
         for config in ChannelManager.parse_simple_config(config_string):
             self.get_channel_from_config(config)
 
+    def configure(self, configs):
+        """Configure the channel manager from list of :class:`ChannelConfig`"""
+        for config in configs:
+            self.get_channel_from_config(asdict(config))
+
     def get_channel_from_config(self, config):
         """Return a Channel object from a parsed configuration.
 
@@ -1115,3 +1133,8 @@ def get_jobs_to_run(self, now):
 
     def get_wakeup_time(self):
         return self._root_channel.get_wakeup_time()
+
+    @property
+    def running_count(self) -> int:
+        """Number of jobs currently running"""
+        return len(self._root_channel._running)