Flow controller perf and close fix - #3
Conversation
The previous bytearray buffer paid O(len(buffer)) twice on every send:
chunk = bytes(self._local_buffer[:bytes_to_send]) # copy n bytes
del self._local_buffer[:bytes_to_send] # shift the tail
At the high-water mark the buffer is ~512 KB and `send_chunks` fires
~50 times per second under load, so each send pays ~1 MB of memcopy
just to dequeue ~32 KB of audio. The slice+delete pattern dominates the
outbound hot path.
Switch the buffer to `collections.deque[bytes]` of incoming chunks plus
a separate `_local_buffer_size: int` byte counter. Each chunk is
appended in `__call__` and the new `_pop_bytes` helper drains chunks
from the head via `popleft`, slicing only the partial head chunk and
allocating the returned bytes once via `b"".join`.
Per-send cost drops from O(buffer_size) to O(send_size). The working-
range guard in `flow_control` (`_local_buffer_size >= self._min_batch`)
stays O(1) because the byte count is maintained inline rather than
walking the deque.
No public-API changes; `drop_buffer` resets both the deque and the size
counter together.
The old `close(gracefully=True)` called `time.sleep(ptime / 1000)` in a loop waiting for the local buffer to drain. `time.sleep` blocks the asyncio event loop, which prevents the `flow_control` task from running - so the very task responsible for draining the buffer can't make progress while we wait for it. In effect, `gracefully=True` deadlocks shutdown until the cancel finally fires. Split the API in two: - `close()` is now synchronous and cancel-only. It drops any pending audio in the local buffer. This matches how the shipped AsteriskWebsocketTransport already tears down (no caller in this repo passes `gracefully=True`). - `aclose(gracefully: bool = False)` is async. When `gracefully` is True, it `await asyncio.sleep(...)`s one chunk-duration at a time while the buffer drains, so the `flow_control` task keeps running and can actually send the audio it's holding. Callers wanting the original drain-then-cancel behavior switch from `flow_controller.close(gracefully=True)` to `await flow_controller.aclose(gracefully=True)`. Non-graceful callers keep using `close()` unchanged.
The per-frame `logger.trace` in `__call__` and the per-send
`logger.debug` in `send_chunks` used eager f-strings:
logger.debug(f"Sent {len(chunk)} bytes to websocket. ...")
Python evaluates the f-string and builds the formatted string before
the call even reaches loguru, so the work happens on every send even
when DEBUG/TRACE is filtered out by the sink (the production case).
Switch both to loguru's native `{}` placeholder form:
logger.debug("Sent {} bytes to websocket. ...", len(chunk), ...)
Loguru defers the formatting until a sink actually accepts the record,
so a filtered call drops the formatting work entirely. The per-call
saving is microseconds, but on the outbound hot path it's the right
shape - same call sites, no behavior change when logs are enabled.
Adds the first test suite for this package. Coverage focuses on the three behaviors changed in the preceding commits: - `_pop_bytes` drains whole chunks, slices the head when the budget falls below it, and caps at the buffered byte count. - `__call__` tracks `_local_buffer_size` and ignores empty chunks. - `drop_buffer` resets the deque, the size counter, and the remote utilization estimate together. - `close()` is synchronous and cancels the flow_control task immediately, without touching the local buffer. - `aclose(gracefully=False)` matches `close()` behavior. - `aclose(gracefully=True)` yields the event loop (regression guard: the previous `time.sleep` form would block until cancel and hang the very draining task we were waiting on). The test bounds the graceful wait with `asyncio.wait_for(..., timeout=1.0)` so a regression fails fast instead of hanging CI. - Hot-path log calls use loguru's positional-arg lazy form so the format string is only expanded when a sink accepts the record. Adds `pytest` and `pytest-asyncio` to the `dev` dependency group, and a small `[tool.pytest.ini_options]` block that enables auto-mode for async tests and pins `testpaths = ["tests"]`.
|
Hi @Salman778, thanks for this great contribution! All your points look totally reasonable. And a special thanks for the tests. |
|
@NikolayShakin For what it's worth, both of this user's PRs are AI slop PRs. I might recommend not engaging with them. |
|
Hi @abalashov, do you see some specific problems with the code? |
The problem with slop PRs isn't first and foremost their correctness, but rather the burden they impose on others to review it, with their all-over-the-place, scattershot approach. Code generators generate far faster than you (or I) could ever hope to review. Human-submitted patches focus on a specific problem area and state the problem being solved at a high level, and clearly and naturally state what is being solved. It is a signature hallmark of lazy copy-pasta PRs that they articulate exactly what was done, in excruciating detail, but not the why. Github/open source is currently overrun with this type of faux "contribution", and poses management problems that are wholly independent of the merits of any given bit of code, and these have to be weighed as a management and political challenge, first and foremost. Obviously, you're the maintainer, and it's up to you whether to accept obvious slop PRs. However, I strongly recommend that you consider the Asterisk's project stance on AI slop PRs as inspiration for your own: |
|
At the very least, this could have been three separate PRs. This makes them easier to revert separately. These kinds of "fix the world, all over the place" approaches reflect a total lack of human reasoning in the process of the code craftsmanship, and no consideration for the software development lifecycle. This has the spectral signature of pure LLM work. |
Agreed, I’ll collapse it to a single async close(gracefully=False) and push to this branch. |
|
Hi @Salman778, I did some tests with deque vs bytearray in flowcontroller, and f-stigns vs lazy parsing and got interesting results. So I would ask you to keep using f-sting for convenience; at least in this code it doesn't give any performance gain. Unless, f-sting were used without variables (example) with static strings. |
- collapse close()/aclose() into a single async close(gracefully=False) - revert lazy loguru formatting to f-strings (no measurable gain) - lower per-send log in send_chunks from debug to trace
693a664 to
d07984f
Compare
|
Done, pushed: |
Keeps the dict dispatch, isinstance check, and static-string cleanups; only the variable-bearing log calls go back to f-strings, matching the maintainer's feedback on NikolayShakin#3
Keeps the dict dispatch, isinstance check, and static-string cleanups; only the variable-bearing log calls go back to f-strings, matching the maintainer's feedback on NikolayShakin#3
| # the flow_control task to keep draining `_local_buffer`. Using | ||
| # `time.sleep` here would block the event loop and prevent the | ||
| # very draining we're waiting on. | ||
| while self._local_buffer_size > 0: |
There was a problem hiding this comment.
One more thing. Could you please add or self._remote_buffer_utilization > 0 in the while loop to wait until the remote buffer is empty as well, because if we close the websocket connection right after we send everything for the buffer, @abalashov found that Asterisk will remove the channel from the bridge right away, so the user will not hear anything that we uploaded to the buffer at the very end.
So if we want to play everything provided by TTS (which is gracefully), we need to wait until the remote buffer is empty.
There was a problem hiding this comment.
Curious - are you using self._remote_buffer_utilization as a casual proxy for the status of QUEUE_DRAINED? :)
I started with that before moving onto QUEUE_DRAINED, and it worked about as well. But of course, this is a Pipecat-side calculation...
There was a problem hiding this comment.
I prefer the pipecat-side calculation because it only depends on the correctness of the calculation algorithm. If we rely on QUEUE_DRAINED events, we still depend on the algorithm remote_buffer_utilization calculation algorithm (based on it, we subscribe to QUEUE_DRAINED every time the remote buffer is empty and we start sending new media), but besides that, we depend on the correctness of many other logical paths and algorithms, including the logic of Asterisk. So I think local(pipecat-side) calculation is more robust.
When (it's planned) REPORT_QUEUE_DRAINED/QUEUE_DRAINED subscription is implemented, we can use QUEUE_DRAINED messages to sync (reset to 0) our locally-calculated remote buffer utilization and measure the drift between the effective and calculated remote buffer utilization.
There was a problem hiding this comment.
That's interesting. Yet in my use-case, as you know, QUEUE_DRAINED is considered the final word on "So if we want to play everything provided by TTS", but in this case, no?
I ask the question because it would have been far easier to just trust the calculation algorithm and have the transport expose something from the serialiser to while True: await asyncio.sleep(0.05) until the remote buffer drains. :-) There's a reason neither of us thought that was the best idea, so I am trying to tease out the philosophical difference between that case and this one.
There was a problem hiding this comment.
Well, definitely there are more elegant options for waiting for the remote buffer to be empty; for instance, with asyncio.event() and unlocking it from the main loop of the flow_controller. But we will still need to calculate remote buffer utilization to understand the moment when we need to send REPORT_QUEUE_DRAINED. So the accuracy of the logic still boils down to the correctness of the remote buffer utilization algorithm, even if we consider QUEUE_DRAINED the final word.
There was a problem hiding this comment.
I will defer to your wisdom on this. However, it seems to me there is a conflict between two somewhat irreconcilable sources of truth here, and that the system should stick to one or the other.
3 participants
Summary
Three small, independent changes to
FlowControllerplus the first testsuite for this package. Each one is a separate commit so they can be
reviewed (or accepted/dropped) on their own merits:
perf: replacebytearraylocal buffer withdeque[bytes].The current
send_chunksdoeschunk = bytes(buf[:n]); del buf[:n],which is O(buffer_size) twice on every send. With the buffer near
the high-water mark (~512 KB by default) and
send_chunksfiring~50× per second under load, each send pays ~1 MB of memcopy just to
dequeue ~32 KB of audio. Switching to a deque of chunks + a separate
_local_buffer_size: intmakes per-send cost O(send_size) instead ofO(buffer_size). The new
_pop_byteshelper drains whole chunks viapopleftand slices only the partial head chunk, allocating thereturned bytes once via
b"".join.fix: splitclose(gracefully=True)into syncclose()+ asyncaclose(). Todayclose(gracefully=True)callstime.sleep(ptime / 1000)in a loop waiting for the local buffer todrain.
time.sleepblocks the asyncio event loop, which means theflow_controltask can't run — so the very task that's supposed todrain the buffer makes no progress while we wait for it. The new
aclose()usesawait asyncio.sleep(...)so the event loop keepsrunning.
close()becomes synchronous and cancel-only, which matcheshow
AsteriskWebsocketTransportand the bundled example already useit (neither passes
gracefully=True). Callers that want the originaldrain-then-cancel behavior switch from
fc.close(gracefully=True)toawait fc.aclose(gracefully=True).perf: use loguru's lazy{}formatting on the hot-path logs.The per-frame
logger.tracein__call__and the per-sendlogger.debuginsend_chunksuse f-strings, which Python expandseagerly — even when the sink filters that level out. Loguru's native
logger.debug("... {} ...", value)form defers the substitutionuntil a sink accepts the record. Microseconds per call, but it's the
right shape for hot-path log statements.
Tests
Adds a new
tests/test_flow_controller.py(13 cases) pluspytest/pytest-asyncioin thedevdependency group. This is thefirst test file in the repo. Coverage:
_pop_bytesdrains whole chunks, slices the head, and handles mixedpop sizes correctly.
__call__ignores empty chunks and updates_local_buffer_size.drop_bufferresets deque + size + utilization together.close()is sync and cancels the flow-control task immediately.aclose(gracefully=True)yields the event loop so the flow-controltask can drain the buffer. The test bounds the graceful wait with
asyncio.wait_for(..., timeout=1.0)so a regression fails fastinstead of hanging CI.
logger.debug/logger.traceare called with positional args(lazy form), not pre-formatted strings.
Run with
uv run pytest.Compatibility / breaking changes
The only signature change is
close(gracefully: bool = False) → close()aclose(gracefully: bool = False). No caller inside this repopasses
gracefully=True, so the bundled transport and example continueto work unchanged. External callers using
close(gracefully=True)willneed to switch to
await fc.aclose(gracefully=True)— happy to makethis a deprecation shim instead if you'd prefer.
Background
I'm using this transport in a production voice agent built on Pipecat.
While auditing the outbound path I traced these three issues; the
bytearrayslice+delete was the biggest single contributor tooutbound CPU under sustained load. Thanks for the work on this package
and for the recent
_min_batch * 2overflow fix — happy to iterate onthis PR however suits you.