core (xmpp): Message processing queue improvements:
Messages are now queued per conversation (i.e. one-to-one chats, or MUCs). This allows messages to be treated in parallel instead of waiting sequentially for each one to be processed.
This commit is contained in:
Goffi
parent
e41d2cf26f
commit
992d14cb86
1 changed files with 176 additions and 28 deletions
|
|
@ -17,14 +17,14 @@
|
|||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import calendar
|
||||
from collections.abc import Iterable
|
||||
from collections.abc import Awaitable, Callable, Iterable
|
||||
import copy
|
||||
from functools import partial
|
||||
import mimetypes
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import time
|
||||
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple, cast
|
||||
from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, cast
|
||||
from urllib.parse import unquote, urlparse
|
||||
import uuid
|
||||
|
||||
|
|
@ -33,6 +33,7 @@ from treq.api import default_pool
|
|||
from treq.client import HTTPClient
|
||||
from twisted.internet import defer, error as internet_error, reactor
|
||||
from twisted.internet import ssl
|
||||
from twisted.internet.base import DelayedCall
|
||||
from twisted.internet.endpoints import TCP4ClientEndpoint, UNIXClientEndpoint
|
||||
from twisted.python import failure
|
||||
from twisted.web.client import Agent
|
||||
|
|
@ -56,7 +57,7 @@ from libervia.backend.memory import cache
|
|||
from libervia.backend.memory import encryption
|
||||
from libervia.backend.memory import persistent
|
||||
from libervia.backend.models.core import MessageData
|
||||
from libervia.backend.models.tor import TorData, TorMode
|
||||
from libervia.backend.models.tor import TorMode
|
||||
from libervia.backend.tools import xml_tools
|
||||
from libervia.backend.tools import utils
|
||||
from libervia.backend.tools.common import data_format
|
||||
|
|
@ -83,6 +84,9 @@ NS_ROSTER_VER = "urn:xmpp:features:rosterver"
|
|||
# with roster jids
|
||||
ROSTER_VER_KEY = "@version@"
|
||||
|
||||
# Cleaning timeout (in seconds)
|
||||
MESSAGE_QUEUE_TIMEOUT = 300
|
||||
|
||||
|
||||
class ClientPluginWrapper:
|
||||
"""Use a plugin with default value if plugin is missing"""
|
||||
|
|
@ -1397,18 +1401,175 @@ class SatXMPPComponent(SatXMPPEntity, core_types.SatXMPPComponent, component.Com
|
|||
return client
|
||||
|
||||
|
||||
class TimestampedElement(domish.Element):
|
||||
"""domish.Element with a timestamp"""
|
||||
|
||||
def __init__(self, element: domish.Element) -> None:
|
||||
super().__init__((element.uri, element.name))
|
||||
self.defaultUri = element.defaultUri
|
||||
self.attributes = element.attributes
|
||||
self.children = element.children
|
||||
self.localPrefixes = element.localPrefixes
|
||||
self.parent = element.parent
|
||||
|
||||
self._received_timestamp: float = time.time()
|
||||
|
||||
@property
|
||||
def received_timestamp(self) -> float:
|
||||
"""Element creation timestamp.
|
||||
|
||||
It is used when message is received in <message> processing workflow, hence its
|
||||
name.
|
||||
"""
|
||||
return self._received_timestamp
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return (
|
||||
f"<{self.__class__.__name__} {self.toXml()} "
|
||||
f"timestamp={self._received_timestamp}>"
|
||||
)
|
||||
|
||||
|
||||
class MessagesQueue:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
client: SatXMPPEntity,
|
||||
key: str,
|
||||
manager: "MessagesQueueManager",
|
||||
) -> None:
|
||||
self.client: SatXMPPEntity = client
|
||||
self.key: str = key
|
||||
self.queue: defer.DeferredQueue[TimestampedElement] = defer.DeferredQueue()
|
||||
self._manager: MessagesQueueManager = manager
|
||||
self._timer: DelayedCall | None = None
|
||||
defer.ensureDeferred(self._process_loop())
|
||||
|
||||
@property
|
||||
def size(self) -> int:
|
||||
"""Number of messages waiting in the queue"""
|
||||
return len(self.queue.pending)
|
||||
|
||||
def put(self, message_elt: TimestampedElement) -> None:
|
||||
"""Add a message to the queue."""
|
||||
self.queue.put(message_elt)
|
||||
self._reset_timer()
|
||||
|
||||
def _reset_timer(self) -> None:
|
||||
if self._timer is not None and self._timer.active():
|
||||
self._timer.cancel()
|
||||
self._timer = reactor.callLater(MESSAGE_QUEUE_TIMEOUT, self._cleanup)
|
||||
|
||||
def _cleanup(self) -> None:
|
||||
"""Called when timeout expires"""
|
||||
if not self.queue.pending:
|
||||
# The queue is empty, we can destroy it.
|
||||
self._manager.remove_queue(self.key)
|
||||
else:
|
||||
# There are still message pending, we wait again.
|
||||
self._reset_timer()
|
||||
|
||||
async def _process_loop(self) -> None:
|
||||
"""Process messages - runs until queue is destroyed"""
|
||||
while True:
|
||||
message = await self.queue.get()
|
||||
try:
|
||||
await self._manager._process_callback(self.client, message)
|
||||
except Exception:
|
||||
log.exception(f"Error processing message in {self.key}")
|
||||
|
||||
|
||||
class MessagesQueueManager:
|
||||
"""Manages queues of messages."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
process_callback: Callable[[SatXMPPEntity, domish.Element], Awaitable[None]],
|
||||
) -> None:
|
||||
self.queues: dict[str, MessagesQueue] = {}
|
||||
self._process_callback = process_callback
|
||||
|
||||
@staticmethod
|
||||
def get_queue_key(client: SatXMPPEntity, message_elt: domish.Element) -> str:
|
||||
"""Get the key associate with the MessagesQueue.
|
||||
|
||||
Queue are grouped per one2one conversation and per MUC. The key is generated from
|
||||
"from" and "to" attribute to have message routed with the right queue.
|
||||
"""
|
||||
try:
|
||||
from_jid = jid.JID(message_elt["from"])
|
||||
except KeyError:
|
||||
from_jid = client.server_jid
|
||||
|
||||
try:
|
||||
to_jid = jid.JID(message_elt["to"])
|
||||
except KeyError:
|
||||
to_jid = client.jid
|
||||
|
||||
msg_type = message_elt.getAttribute("type", "normal")
|
||||
|
||||
if msg_type == "groupchat":
|
||||
room_jid_s = from_jid.userhost()
|
||||
return f"muc:{room_jid_s}"
|
||||
else:
|
||||
# We sort the pair to have the same key for both directions.
|
||||
key = tuple(sorted([from_jid.userhost(), to_jid.userhost()]))
|
||||
return f"one2one:{key[0]}:{key[1]}"
|
||||
|
||||
def get_or_create_queue(
|
||||
self, client: SatXMPPEntity, message_elt: TimestampedElement
|
||||
) -> MessagesQueue:
|
||||
"""Get existing queue or create new one"""
|
||||
key = self.get_queue_key(client, message_elt)
|
||||
if key not in self.queues:
|
||||
log.debug(f"Creating queue for {key}")
|
||||
self.queues[key] = MessagesQueue(
|
||||
client=client,
|
||||
key=key,
|
||||
manager=self,
|
||||
)
|
||||
return self.queues[key]
|
||||
|
||||
def put(self, client: SatXMPPEntity, message_elt: TimestampedElement) -> None:
|
||||
"""Put message in queue.
|
||||
|
||||
The queue will be created if needed.
|
||||
@param client: XMPP client
|
||||
@param message_elt: Message to put in queue.
|
||||
"""
|
||||
queue = self.get_or_create_queue(client, message_elt)
|
||||
queue.put(message_elt)
|
||||
|
||||
def remove_queue(self, key: str) -> None:
|
||||
"""Remove an unused queue."""
|
||||
try:
|
||||
queue = self.queues.pop(key)
|
||||
except KeyError:
|
||||
pass
|
||||
else:
|
||||
if queue._timer is not None and queue._timer.active():
|
||||
queue._timer.cancel()
|
||||
|
||||
def get_stats(self) -> dict[str, Any]:
|
||||
"""Debug/monitoring stats"""
|
||||
return {
|
||||
"active_queues": len(self.queues),
|
||||
"queue_sizes": {key: queue.size for key, queue in self.queues.items()},
|
||||
}
|
||||
|
||||
|
||||
class SatMessageProtocol(xmppim.MessageProtocol):
|
||||
|
||||
def __init__(self):
|
||||
xmppim.MessageProtocol.__init__(self)
|
||||
self.messages_queue = defer.DeferredQueue()
|
||||
self.queue_manager = MessagesQueueManager(self.process_message)
|
||||
|
||||
def setHandlerParent(self, parent):
|
||||
super().setHandlerParent(parent)
|
||||
defer.ensureDeferred(self.process_messages())
|
||||
|
||||
@property
|
||||
def client(self):
|
||||
def client(self) -> SatXMPPEntity:
|
||||
assert self.parent is not None
|
||||
return self.parent
|
||||
|
||||
def normalize_ns(self, elt: domish.Element, namespace: Optional[str]) -> None:
|
||||
|
|
@ -1518,10 +1679,10 @@ class SatMessageProtocol(xmppim.MessageProtocol):
|
|||
|
||||
# delay and timestamp
|
||||
try:
|
||||
received_timestamp = message_elt._received_timestamp
|
||||
received_timestamp = message_elt.received_timestamp
|
||||
except AttributeError:
|
||||
# message_elt._received_timestamp should have been set in onMessage
|
||||
# but if parse_message is called directly, it can be missing
|
||||
# message_elt.received_timestamp is avaiable with TimestampedElement, but if
|
||||
# parse_message is called directly, it can be missing
|
||||
log.debug(
|
||||
"missing received timestamp for {message_elt}".format(
|
||||
message_elt=message_elt
|
||||
|
|
@ -1543,32 +1704,19 @@ class SatMessageProtocol(xmppim.MessageProtocol):
|
|||
G.host.trigger.point("message_parse", client, message_elt, data)
|
||||
return data
|
||||
|
||||
def onMessage(self, message_elt: domish.Element) -> None:
|
||||
message_elt._received_timestamp = time.time()
|
||||
self.messages_queue.put(message_elt)
|
||||
def onMessage(self, message: domish.Element) -> None:
|
||||
"""Called when a <message> is received.
|
||||
|
||||
async def process_messages(self) -> None:
|
||||
"""Process message in order
|
||||
|
||||
Messages are processed in a queue to avoid race conditions and ensure orderly
|
||||
processing.
|
||||
This put the message in queue, so they are processed in order.
|
||||
"""
|
||||
client = self.parent
|
||||
if client is None:
|
||||
log.error("client should not be None!")
|
||||
raise exceptions.InternalError()
|
||||
while True:
|
||||
message_elt = await self.messages_queue.get()
|
||||
try:
|
||||
await self.process_message(client, message_elt)
|
||||
except Exception:
|
||||
log.exception(f"Can't process message {message_elt.toXml()}")
|
||||
self.queue_manager.put(self.client, TimestampedElement(message))
|
||||
|
||||
def _on_processing_timeout(
|
||||
self, message_elt: domish.Element, async_point_d: defer.Deferred
|
||||
) -> None:
|
||||
key = MessagesQueueManager.get_queue_key(self.client, message_elt)
|
||||
log.error(
|
||||
"Processing of following message took too long, cancelling:"
|
||||
f"Processing of following message took too long in {key}, cancelling:"
|
||||
f"{message_elt.toXml()}"
|
||||
)
|
||||
async_point_d.cancel()
|
||||
|
|
|
|||
Loading…
Reference in a new issue