feat: add benchmark

Author: brian14708

.gitignore

@@ -2,5 +2,6 @@
 /site
 /.venv
 /.nox
+/.codspeed
 /logs
 __pycache__

pyproject.toml

@@ -18,6 +18,7 @@ dev = [
     "pydantic>=2.11.9",
     "pytest>=7.0",
     "pytest-asyncio>=0.21",
+    "pytest-codspeed>=4.1.1",
 ]
 type-checking = [
     "basedmypy>=2.10.0",

src/duron/_core/context.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import asyncio
-import base64
+import binascii
 from contextvars import ContextVar
 from random import Random
 from typing import (
@@ -192,7 +192,10 @@ async def create_promise(
             self._loop,
             ExternalPromiseCreate(metadata=metadata, return_type=dtype),
         )
-        return (base64.b64encode(fut.id).decode(), cast("asyncio.Future[_T]", fut))
+        return (
+            binascii.b2a_base64(fut.id, newline=False).decode(),
+            cast("asyncio.Future[_T]", fut),
+        )
 
     async def barrier(self) -> int:
         if asyncio.get_running_loop() is not self._loop:

src/duron/_core/invoke.py

@@ -289,7 +289,7 @@ def __init__(
         ]
         | None = None,
     ) -> None:
-        self._loop = create_loop(asyncio.get_event_loop(), b"")
+        self._loop = create_loop(asyncio.get_running_loop(), b"")
         self._task = self._loop.create_task(task)
         self._log = log
         self._codec = codec
@@ -370,14 +370,18 @@ async def _step(self) -> WaitSet | None:
         while True:
             self._loop.tick(self.now())
             result = self._loop.poll_completion(self._task)
-            if result is None or self._pending_ops.issuperset(o.id for o in result.ops):
+            if result is None:
                 return result
 
+            new_ops = False
             for s in result.ops:
                 sid = s.id
                 if sid not in self._pending_ops:
                     self._pending_ops.add(sid)
                     await self.enqueue_op(sid, s)
+                    new_ops = True
+            if not new_ops:
+                return result
 
     async def handle_message(
         self,
@@ -489,7 +493,7 @@ async def enqueue_op(self, id_: bytes, fut: OpFuture[object]) -> None:
                 async def cb() -> None:
                     entry: PromiseCompleteEntry = {
                         "ts": self.now(),
-                        "id": _encode_id(id_, 1),
+                        "id": _encode_id(id_, ack=True),
                         "type": "promise/complete",
                         "promise_id": _encode_id(id_),
                     }
@@ -604,7 +608,7 @@ async def complete_external_promise(
             raise ValueError(msg)
         entry: PromiseCompleteEntry = {
             "ts": self.now(),
-            "id": _encode_id(_decode_id(id_), 1),
+            "id": _encode_id(_decode_id(id_), ack=True),
             "type": "promise/complete",
             "promise_id": id_,
         }
@@ -672,10 +676,10 @@ async def close_stream(
         return cnt
 
 
-def _encode_id(id_: bytes, flag: int = 0) -> str:
-    if flag != 0:
+def _encode_id(id_: bytes, *, ack: bool = False) -> str:
+    if ack:
         id_ = blake2b(
-            flag.to_bytes(4, "little", signed=True) + id_,
+            b"\x01\x00\x00\x00" + id_,
             digest_size=12,
         ).digest()
     return base64.b64encode(id_).decode()