feat: improve writer performance with a deque

src/dbus_fast/aio/message_bus.py

@@ -2,7 +2,7 @@
 import asyncio
 import logging
 import socket
-from asyncio import Queue
+from collections import deque
 from copy import copy
 from typing import Any, Optional
 
@@ -36,7 +36,7 @@ def _future_set_result(fut: asyncio.Future, result: Any) -> None:
 
 class _MessageWriter:
     def __init__(self, bus: "MessageBus") -> None:
-        self.messages = Queue()
+        self.messages = deque()
         self.negotiate_unix_fd = bus._negotiate_unix_fd
         self.bus = bus
         self.sock = bus._sock
@@ -51,12 +51,12 @@ def write_callback(self, remove_writer: bool = True) -> None:
         try:
             while True:
                 if self.buf is None:
-                    if self.messages.qsize() == 0:
+                    if not self.messages:
                         # nothing more to write
                         if remove_writer:
                             self.loop.remove_writer(self.fd)
                         return
-                    buf, unix_fds, fut = self.messages.get_nowait()
+                    buf, unix_fds, fut = self.messages.pop()
                     self.unix_fds = unix_fds
                     self.buf = memoryview(buf)
                     self.offset = 0
@@ -90,7 +90,7 @@ def write_callback(self, remove_writer: bool = True) -> None:
             self.bus._finalize(e)
 
     def buffer_message(self, msg: Message, future=None):
-        self.messages.put_nowait(
+        self.messages.append(
             (
                 msg._marshall(negotiate_unix_fd=self.negotiate_unix_fd),
                 copy(msg.unix_fds),
@@ -103,7 +103,7 @@ def _write_without_remove_writer(self):
         self.write_callback(remove_writer=False)
 
     def schedule_write(self, msg: Message = None, future=None):
-        queue_is_empty = self.messages.qsize() == 0
+        queue_is_empty = not self.messages
         if msg is not None:
             self.buffer_message(msg, future)
         if self.bus.unique_name:
@@ -115,7 +115,7 @@ def schedule_write(self, msg: Message = None, future=None):
                 self._write_without_remove_writer()
             if (
                 self.buf is not None
-                or self.messages.qsize() != 0
+                or self.messages
                 or not self.fut
                 or not self.fut.done()
             ):

tests/test_request_name.py

@@ -1,3 +1,5 @@
+import sys
+
 import pytest
 
 from dbus_fast import (
@@ -68,6 +70,7 @@ async def get_name_owner(name):
     bus2.disconnect()
 
 
+@pytest.mark.skipif(sys.version_info[:3][1] == 10, reason="segfaults on py3.10")
 @pytest.mark.skipif(not has_gi, reason=skip_reason_no_gi)
 def test_request_name_glib():
     test_name = "glib.test.request.name"