small performance improvements

Author: SamDanielThangarajan

README.md

@@ -15,3 +15,24 @@ by extending the base message class defined in the package.
 
 More documentation can be found in the [github-pages](https://nasdaq.github.io/nasdaq-protocols/)
 
+
+Performance
+-----------
+
+For improved throughput, install the optional performance dependencies:
+
+```bash
+pip install nasdaq-protocols[performance]
+```
+
+This installs [uvloop](https://github.com/MagicStack/uvloop) (Linux/macOS only), which provides ~4-5x faster event loop scheduling.
+
+To activate it, call early in your application before any asyncio work:
+
+```python
+from nasdaq_protocols.common import use_fast_loop
+use_fast_loop()
+```
+
+If uvloop is not installed, `use_fast_loop()` is a no-op.
+

pyproject.toml

@@ -23,6 +23,9 @@ dependencies = [
 ]
 dynamic = ["version"]
 
+[project.optional-dependencies]
+performance = ["uvloop>=0.19; sys_platform != 'win32'"]
+
 [project.scripts]
 nasdaq-ouch-codegen = "nasdaq_protocols.ouch.codegen:generate"
 nasdaq-itch-codegen = "nasdaq_protocols.itch.codegen:generate"

src/nasdaq_protocols/common/__init__.py

@@ -4,3 +4,4 @@
 from .session import *
 from .message import *
 from .sync_executor import *
+from ._loop import use_fast_loop

src/nasdaq_protocols/common/_loop.py

@@ -0,0 +1,16 @@
+"""Optional fast event loop integration."""
+
+
+def use_fast_loop():
+    """Install uvloop as the default event loop policy if available.
+
+    Call this early in your application entry point before creating any
+    asyncio event loops. If uvloop is not installed, this is a no-op.
+
+    Install with: pip install nasdaq-protocols[performance]
+    """
+    try:
+        import uvloop
+        uvloop.install()
+    except ImportError:
+        pass

src/nasdaq_protocols/common/session.py

@@ -109,6 +109,7 @@ class Reader(Stoppable):
     on_msg_coro: OnMsgCoro = attrs.field(validator=Validators.not_none())
     on_close_coro: OnCloseCoro = attrs.field(validator=Validators.not_none())
     _buffer: bytearray = attrs.field(init=False, factory=bytearray)
+    _read_pos: int = attrs.field(init=False, default=0)
     _drain_buffer: bytearray | None = attrs.field(init=False, default=None)
     _task: asyncio.Task = attrs.field(init=False, default=None)
     _stopped: bool = attrs.field(init=False, default=False)
@@ -130,6 +131,10 @@ async def buffer_until_drained(self, discard_buffer: bool = False):
         finally:
             self._drain_mode = None
             if not discard_buffer:
+                # Compact before extending with drain buffer
+                if self._read_pos > 0:
+                    del self._buffer[:self._read_pos]
+                    self._read_pos = 0
                 self._buffer.extend(self._drain_buffer)
             self._drain_buffer = None
 
@@ -153,11 +158,11 @@ def is_stopped(self):
 
     async def _process(self):
         while not self._stopped:
-            len_before = len_after = len(self._buffer)
-            if len_before > 0:
+            available_before = available_after = len(self._buffer) - self._read_pos
+            if available_before > 0:
                 await self._process_1()
-                len_after = len(self._buffer)
-            if self._drain_mode and not self._drain_mode.is_set() and len_after == len_before:
+                available_after = len(self._buffer) - self._read_pos
+            if self._drain_mode and not self._drain_mode.is_set() and available_after == available_before:
                 self._drain_mode.set()
 
             await asyncio.sleep(0.0001)

src/nasdaq_protocols/fix/_reader.py

@@ -16,20 +16,28 @@
 class FixMessageReader(common.Reader):
     def deserialize(self):
         empty_response = (None, False, False)
-        if self._buffer.find(MSG_TYPE_TAG) != -1:
-            start = self._buffer.find(b'=', SKIP_FIRST_EQ_POS)
+        view = bytes(self._buffer[self._read_pos:])
+
+        if view.find(MSG_TYPE_TAG) != -1:
+            start = view.find(b'=', SKIP_FIRST_EQ_POS)
             if start == -1:
                 return empty_response
-            end = self._buffer.find(SOH, start)
+            end = view.find(SOH, start)
             if end == -1:
                 return empty_response
-            body_length = int(self._buffer[start+1:end])
+            body_length = int(view[start+1:end])
             msg_len = calc_msg_len(end+1, body_length)
-            if len(self._buffer) < msg_len:
+            if len(view) < msg_len:
                 return empty_response
 
-            _len, msg = Message.from_bytes(self._buffer[:msg_len])
-            self._buffer = self._buffer[msg_len:]
+            _len, msg = Message.from_bytes(view[:msg_len])
+            self._read_pos += msg_len
+
+            # Compact when more than half the buffer is consumed
+            if self._read_pos > len(self._buffer) // 2:
+                del self._buffer[:self._read_pos]
+                self._read_pos = 0
+
             return msg, msg.is_logout(), msg.is_heartbeat()
         return empty_response
 

src/nasdaq_protocols/soup/_reader.py

@@ -10,16 +10,22 @@
 class SoupMessageReader(common.Reader):
     def deserialize(self) -> Any:
         empty_response = (None, False, False)
-        buff_len = len(self._buffer)
+        available = len(self._buffer) - self._read_pos
 
-        if buff_len < 2:
+        if available < 2:
             return empty_response
 
-        siz = int.from_bytes(self._buffer[:2], 'big')
-        if (siz+2) > buff_len:
+        siz = int.from_bytes(self._buffer[self._read_pos:self._read_pos + 2], 'big')
+        if (siz + 2) > available:
             return empty_response
 
-        _, msg = SoupMessage.from_bytes(self._buffer[:siz + 2])
-        self._buffer = self._buffer[siz + 2:]
+        frame = bytes(memoryview(self._buffer)[self._read_pos:self._read_pos + siz + 2])
+        _, msg = SoupMessage.from_bytes(frame)
+        self._read_pos += siz + 2
+
+        # Compact when more than half the buffer is consumed
+        if self._read_pos > len(self._buffer) // 2:
+            del self._buffer[:self._read_pos]
+            self._read_pos = 0
 
         return msg, msg.is_logout(), msg.is_heartbeat()