[PERF] bus: do not use registry to dispatch messages

When there are more databases than what the LRU cache can hold,
the gevent worker struggles to keep notifications flowing. Each
outgoing message requires a registry, which is highly inefficient.

This commit replaces the call to `bus.bus@_poll` by a direct SQL query
thus removing the need to build a registry.

closes odoo#235746

Signed-off-by: Julien Castiaux (juc) <juc@odoo.com>

Author: tsm-odoo

addons/bus/__init__.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
 from . import models
+from . import session_helpers
 from . import controllers
 from . import websocket

addons/bus/models/bus.py

@@ -14,7 +14,6 @@
 from odoo import api, fields, models
 from odoo.service.server import CommonServer
 from odoo.tools import json_default, SQL
-from odoo.tools.constants import GC_UNLINK_LIMIT
 from odoo.tools.misc import OrderedSet
 
 _logger = logging.getLogger(__name__)
@@ -41,12 +40,37 @@ def get_notify_payload_max_length(default=8000):
 NOTIFY_PAYLOAD_MAX_LENGTH = get_notify_payload_max_length()
 
 
+def fetch_bus_notifications(cr, channels, last=0, ignore_ids=None):
+    """Fetch notifications from the bus table.
+
+    :param cr: Database cursor.
+    :param channels: List of channels for which notifications should be fetched.
+        May contain channel names, model instances, or (model, string) tuples.
+    :param last: The ID of the last fetched notification. Defaults to 0.
+    :param ignore_ids: IDs to exclude.
+    :return: List of notifications.
+
+    """
+    conditions = [
+        SQL("channel IN %s", tuple(json_dump(channel_with_db(cr.dbname, c)) for c in channels)),
+        SQL("create_date > %s", fields.Datetime.now() - datetime.timedelta(seconds=TIMEOUT))
+        if last == 0
+        else SQL("id > %s", last),
+    ]
+    if ignore_ids:
+        conditions.append(SQL("id NOT IN %s", tuple(ignore_ids)))
+    where = SQL(" AND ").join(conditions)
+    cr.execute(SQL("SELECT id, message FROM bus_bus WHERE %s ORDER BY id", where))
+    return [{"id": r[0], "message": json.loads(r[1])} for r in cr.fetchall()]
+
+
 # ---------------------------------------------------------
 # Bus
 # ---------------------------------------------------------
 def json_dump(v):
     return json.dumps(v, separators=(',', ':'), default=json_default)
 
+
 def hashable(key):
     if isinstance(key, list):
         key = tuple(key)
@@ -164,25 +188,7 @@ def notify():
 
     @api.model
     def _poll(self, channels, last=0, ignore_ids=None):
-        # first poll return the notification in the 'buffer'
-        if last == 0:
-            timeout_ago = fields.Datetime.now() - datetime.timedelta(seconds=TIMEOUT)
-            domain = [('create_date', '>', timeout_ago)]
-        else:  # else returns the unread notifications
-            domain = [('id', '>', last)]
-        if ignore_ids:
-            domain.append(("id", "not in", ignore_ids))
-        channels = [json_dump(channel_with_db(self.env.cr.dbname, c)) for c in channels]
-        domain.append(('channel', 'in', channels))
-        notifications = self.sudo().search_read(domain, ["message"])
-        # list of notification to return
-        result = []
-        for notif in notifications:
-            result.append({
-                'id': notif['id'],
-                'message': json.loads(notif['message']),
-            })
-        return result
+        return fetch_bus_notifications(self.env.cr, channels, last, ignore_ids)
 
     def _bus_last_id(self):
         last = self.env['bus.bus'].search([], order='id desc', limit=1)
@@ -193,12 +199,6 @@ def _bus_last_id(self):
 # Dispatcher
 # ---------------------------------------------------------
 
-class BusSubscription:
-    def __init__(self, channels, last):
-        self.last_notification_id = last
-        self.channels = channels
-
-
 class ImDispatch(threading.Thread):
     def __init__(self):
         super().__init__(daemon=True, name=f'{__name__}.Bus')

addons/bus/websocket.py

@@ -33,7 +33,7 @@
 from odoo.service.server import CommonServer
 from odoo.tools import config
 
-from .models.bus import dispatch
+from .models.bus import dispatch, fetch_bus_notifications
 from .session_helpers import check_session, new_env
 
 _logger = logging.getLogger(__name__)
@@ -754,9 +754,8 @@ def _process_control_command(self, command, data):
     def _dispatch_bus_notifications(self):
         self._waiting_for_dispatch = False
         with acquire_cursor(self._session.db) as cr:
-            env = new_env(cr, self._session)
-            notifications = env['bus.bus']._poll(
-                self._channels, self._last_notif_sent_id, [n[0] for n in self._notif_history]
+            notifications = fetch_bus_notifications(
+                cr, self._channels, self._last_notif_sent_id, [n[0] for n in self._notif_history]
             )
             if not notifications:
                 return