[IMP] queue_job_cron_jobrunner: channel

Author: Koudja

queue_job_cron_jobrunner/models/queue_job.py

@@ -19,6 +19,7 @@
     RetryableJobError,
 )
 from odoo.addons.queue_job.job import Job
+from odoo.addons.queue_job.jobrunner import QueueJobRunner
 
 _logger = logging.getLogger(__name__)
 
@@ -32,42 +33,68 @@ def _acquire_one_job(self):
 
         :returns: queue.job record (locked for update)
         """
-        # TODO: This method should respect channel priority and capacity,
-        #       rather than just fetching them by creation date.
-        self.flush()
+        runner = QueueJobRunner.from_environ_or_config()
+        job_id = False
+        # Lock all pending jobs
         self.env.cr.execute(
             """
             SELECT id
             FROM queue_job
             WHERE state = 'pending'
             AND (eta IS NULL OR eta <= (now() AT TIME ZONE 'UTC'))
             ORDER BY priority, date_created
-            LIMIT 1 FOR NO KEY UPDATE SKIP LOCKED
+            FOR NO KEY UPDATE
             """
         )
-        row = self.env.cr.fetchone()
-        return self.browse(row and row[0])
+        rows = self.env.cr.fetchall()
+
+        channels = {}
+        for queue_job in self.search([("state", "=", "started")]):
+            if not queue_job.channel:
+                continue
+            channels[queue_job.channel] = channels.get(queue_job.channel, 0) + 1
+        channels_without_capacity = set()
+        for channel_str, running in channels.items():
+            channel = runner.channel_manager.get_channel_by_name(channel_str)
+            if channel.capacity <= running:
+                channels_without_capacity.add(channel_str)
+        _logger.info(
+            "_acquire_one_job channels_without_capacity %s",
+            channels_without_capacity,
+        )
+
+        for row in rows:
+            if not (row and row[0]):
+                continue
+            queue_job = self.browse(row[0])
+            if queue_job.channel and queue_job.channel in channels_without_capacity:
+                continue
+            job = Job._load_from_db_record(queue_job)
+            job.set_started()
+            job.store()
+            _logger.info(
+                "_acquire_one_job queue.job %s[channel=%s,uuid=%s] started",
+                row[0],
+                job.channel,
+                job.uuid,
+            )
+            job_id = row[0]
+            break
+        self.flush()
+        self.env.cr.commit()  # pylint: disable=E8102
+        return self.browse(job_id)
 
     def _process(self, commit=False):
         """Process the job"""
         self.ensure_one()
         job = Job._load_from_db_record(self)
-        # Set it as started
-        job.set_started()
-        job.store()
-        _logger.debug("%s started", job.uuid)
-        # TODO: Commit the state change so that the state can be read from the UI
-        #       while the job is processing. However, doing this will release the
-        #       lock on the db, so we need to find another way.
-        # if commit:
-        #     self.flush()
-        #     self.env.cr.commit()
-
         # Actual processing
         try:
             try:
                 with self.env.cr.savepoint():
+                    _logger.debug("perform %s", job.uuid)
                     job.perform()
+                    _logger.debug("performed %s", job.uuid)
                     job.set_done()
                     job.store()
             except OperationalError as err: