|
| 1 | +"""Async transport layer for SyncTable delta propagation. |
| 2 | +
|
| 3 | +Provides async-compatible transports for use with asyncio, aiohttp, |
| 4 | +FastAPI, WebSocket servers, and any other async framework. |
| 5 | +
|
| 6 | +Because XOR accumulation is instant (no I/O), only the network send |
| 7 | +needs to be async. Receives and reads remain synchronous. |
| 8 | +
|
| 9 | +Usage: |
| 10 | + from atomik_core.async_transport import AsyncCallbackTransport, AsyncSyncTable |
| 11 | +
|
| 12 | + # Low-level: async callback transport |
| 13 | + async def send_ws(data: bytes): |
| 14 | + await websocket.send(data) |
| 15 | +
|
| 16 | + transport = AsyncCallbackTransport(table, send_fn=send_ws) |
| 17 | + await transport.send_pending() # flush any queued deltas |
| 18 | +
|
| 19 | + # High-level: AsyncSyncTable wraps SyncTable with async put() |
| 20 | + async_table = AsyncSyncTable(256, send_fn=send_ws) |
| 21 | + await async_table.put(0, 0xDEADBEEF) # sends delta immediately |
| 22 | +
|
| 23 | +SPDX-License-Identifier: Apache-2.0 |
| 24 | +""" |
| 25 | + |
| 26 | +from __future__ import annotations |
| 27 | + |
| 28 | +from atomik_core.stream import DeltaMessage |
| 29 | +from atomik_core.sync import SyncTable |
| 30 | + |
| 31 | + |
| 32 | +class AsyncCallbackTransport: |
| 33 | + """Async-compatible transport for SyncTable. |
| 34 | +
|
| 35 | + Accepts an async send function for use with asyncio, aiohttp, FastAPI, etc. |
| 36 | + Since SyncTable's on_delta callback is synchronous, deltas are queued |
| 37 | + internally and flushed via the async send_pending() method. |
| 38 | +
|
| 39 | + Usage: |
| 40 | + async def send_ws(data: bytes): |
| 41 | + await websocket.send(data) |
| 42 | +
|
| 43 | + transport = AsyncCallbackTransport(table, send_fn=send_ws) |
| 44 | + await transport.send_pending() # flush any queued deltas |
| 45 | + """ |
| 46 | + |
| 47 | + __slots__ = ("_table", "_send_fn", "_queue") |
| 48 | + |
| 49 | + def __init__(self, table: SyncTable, send_fn=None) -> None: |
| 50 | + """Create an async callback transport. |
| 51 | +
|
| 52 | + Args: |
| 53 | + table: The SyncTable to wrap. |
| 54 | + send_fn: An async callable that accepts bytes. Called with |
| 55 | + serialized DeltaMessages when send_pending() is awaited. |
| 56 | + """ |
| 57 | + self._table = table |
| 58 | + self._send_fn = send_fn |
| 59 | + self._queue: list[bytes] = [] |
| 60 | + table.on_delta(self._on_local_delta) |
| 61 | + |
| 62 | + def _on_local_delta(self, msg: DeltaMessage) -> None: |
| 63 | + """Called synchronously when local table generates a delta. |
| 64 | +
|
| 65 | + Queues the serialized message for later async sending. |
| 66 | + """ |
| 67 | + self._queue.append(msg.to_bytes()) |
| 68 | + |
| 69 | + async def send_pending(self) -> int: |
| 70 | + """Send all queued deltas via the async send function. |
| 71 | +
|
| 72 | + Returns: |
| 73 | + Number of deltas sent. |
| 74 | + """ |
| 75 | + if self._send_fn is None or not self._queue: |
| 76 | + count = len(self._queue) |
| 77 | + self._queue.clear() |
| 78 | + return count |
| 79 | + pending = self._queue |
| 80 | + self._queue = [] |
| 81 | + for data in pending: |
| 82 | + await self._send_fn(data) |
| 83 | + return len(pending) |
| 84 | + |
| 85 | + def receive_bytes(self, data: bytes) -> None: |
| 86 | + """Process incoming delta bytes from the network. |
| 87 | +
|
| 88 | + Synchronous — XOR accumulation is instant, no I/O needed. |
| 89 | +
|
| 90 | + Args: |
| 91 | + data: Exactly 16 bytes in network byte order (DeltaMessage wire format). |
| 92 | + """ |
| 93 | + msg = DeltaMessage.from_bytes(data) |
| 94 | + self._table.receive(msg) |
| 95 | + |
| 96 | + @property |
| 97 | + def pending_count(self) -> int: |
| 98 | + """Number of deltas queued for sending.""" |
| 99 | + return len(self._queue) |
| 100 | + |
| 101 | + |
| 102 | +class AsyncSyncTable: |
| 103 | + """Async wrapper around SyncTable with automatic delta sending. |
| 104 | +
|
| 105 | + Provides an async put() that writes locally and immediately sends |
| 106 | + the delta via the configured async send function. Reads and receives |
| 107 | + remain synchronous since they involve no I/O. |
| 108 | +
|
| 109 | + Usage: |
| 110 | + async def send_ws(data: bytes): |
| 111 | + await websocket.send(data) |
| 112 | +
|
| 113 | + table = AsyncSyncTable(256, send_fn=send_ws) |
| 114 | + await table.put(0, 0xDEADBEEF) # write + send in one await |
| 115 | + value = table.get(0) # sync read (instant) |
| 116 | + """ |
| 117 | + |
| 118 | + __slots__ = ("_table", "_transport") |
| 119 | + |
| 120 | + def __init__(self, num_contexts: int = 256, width: int = 64, send_fn=None) -> None: |
| 121 | + """Create an async-capable synchronized state table. |
| 122 | +
|
| 123 | + Args: |
| 124 | + num_contexts: Number of addressable context slots (default 256). |
| 125 | + width: Bit width per context (default 64). |
| 126 | + send_fn: An async callable that accepts bytes. Called with |
| 127 | + serialized DeltaMessages on each put(). |
| 128 | + """ |
| 129 | + self._table = SyncTable(num_contexts=num_contexts, width=width) |
| 130 | + self._transport = AsyncCallbackTransport(self._table, send_fn=send_fn) |
| 131 | + |
| 132 | + async def put(self, addr: int, value: int) -> DeltaMessage: |
| 133 | + """Write a value and send the delta to peers. |
| 134 | +
|
| 135 | + Args: |
| 136 | + addr: Context address to write. |
| 137 | + value: Desired new state value. |
| 138 | +
|
| 139 | + Returns: |
| 140 | + The DeltaMessage that was sent (or would be sent if no send_fn). |
| 141 | + """ |
| 142 | + msg = self._table.put(addr, value) |
| 143 | + await self._transport.send_pending() |
| 144 | + return msg |
| 145 | + |
| 146 | + def get(self, addr: int) -> int: |
| 147 | + """Read the current state at an address. |
| 148 | +
|
| 149 | + Synchronous — reads are instant (reference XOR accumulator). |
| 150 | +
|
| 151 | + Args: |
| 152 | + addr: Context address to read. |
| 153 | +
|
| 154 | + Returns: |
| 155 | + Current state value. |
| 156 | + """ |
| 157 | + return self._table.get(addr) |
| 158 | + |
| 159 | + def receive(self, msg: DeltaMessage) -> None: |
| 160 | + """Apply a delta received from a remote peer. |
| 161 | +
|
| 162 | + Synchronous — XOR accumulation is instant. Does NOT trigger |
| 163 | + sending (prevents echo loops). |
| 164 | +
|
| 165 | + Args: |
| 166 | + msg: DeltaMessage from a remote peer. |
| 167 | + """ |
| 168 | + self._table.receive(msg) |
| 169 | + |
| 170 | + def receive_bytes(self, data: bytes) -> None: |
| 171 | + """Apply delta bytes received from the network. |
| 172 | +
|
| 173 | + Synchronous — XOR accumulation is instant. |
| 174 | +
|
| 175 | + Args: |
| 176 | + data: Exactly 16 bytes in network byte order. |
| 177 | + """ |
| 178 | + self._transport.receive_bytes(data) |
| 179 | + |
| 180 | + @property |
| 181 | + def num_contexts(self) -> int: |
| 182 | + """Number of context slots in this table.""" |
| 183 | + return self._table.num_contexts |
| 184 | + |
| 185 | + @property |
| 186 | + def width(self) -> int: |
| 187 | + """Bit width per context.""" |
| 188 | + return self._table.width |
| 189 | + |
| 190 | + def snapshot(self) -> dict[int, int]: |
| 191 | + """Serialize the full state of all contexts.""" |
| 192 | + return self._table.snapshot() |
| 193 | + |
| 194 | + def diff(self, other: AsyncSyncTable) -> list[int]: |
| 195 | + """Return addresses where this table and another disagree.""" |
| 196 | + return self._table.diff(other._table) |
| 197 | + |
| 198 | + def __repr__(self) -> str: |
| 199 | + return f"AsyncSyncTable({self._table!r})" |
0 commit comments