perf: non-blocking tracing dispatch and flow polling optimization

[vitalii-dynamiq]

Summary

Surgical performance optimizations targeting two high-impact bottlenecks identified under high parallel load (20/50/100 concurrent requests).

Changes

1. Non-blocking trace dispatch (dynamiq/clients/dynamiq.py)

Problem: DynamiqTracingClient.trace() called _send_traces_sync() on the caller thread, issuing a blocking requests.post() with a 60s timeout. Under high concurrency this blocked execution threads and added significant latency to every node/flow completion callback.

Fix:

• Added a background daemon thread with a SimpleQueue for trace dispatch
• trace() now enqueues runs instead of blocking on HTTP POST
• Added close() method + atexit hook for graceful shutdown
• The async path is unchanged (already non-blocking via loop.create_task)

2. Reuse httpx.AsyncClient (dynamiq/clients/dynamiq.py)

Problem: request() created a new httpx.AsyncClient context manager per call, paying TCP/TLS handshake overhead every time.

Fix:

• Lazily create a shared httpx.AsyncClient with double-checked locking
• Reused across all async trace calls for the lifetime of the client

3. Conditional flow polling sleep (dynamiq/flows/flow.py)

Problem: time.sleep(0.003) fired unconditionally on every iteration of the sync flow polling loop, even after futures.wait(FIRST_COMPLETED) had already blocked waiting for real work. This added unnecessary 3ms per-iteration latency.

Fix:

• Guard time.sleep(0.003) so it only fires when no futures completed (empty results dict)
• The sleep still prevents busy-waiting when no nodes are ready
• The async flow's asyncio.sleep(0.003) is left as-is since it properly yields to the event loop

Impact

Bottleneck	Before	After
Sync trace flush	Blocks caller thread up to 60s	Fire-and-forget via background thread
Async HTTP client	New TCP connection per trace call	Persistent connection pool
Flow polling sleep	3ms added after every iteration	3ms only when idle (no completed work)

Testing

• All 718 unit tests pass (1 pre-existing failure in test_example_yaml_files_load due to missing OPENAI_API_KEY - unrelated)
• flake8: clean
• bandit: clean
• darker (black + isort): clean

---

Note

Medium Risk
Introduces background-thread/queue-based trace dispatch and shared httpx.AsyncClient lifecycle management, which can affect shutdown behavior, resource cleanup, and trace delivery under concurrency. The flow-loop sleep gating is low risk but changes timing in synchronous execution.

Overview
Improves performance in tracing and synchronous flow execution by making sync DynamiqTracingClient.trace() fire-and-forget via a background daemon thread/SimpleQueue, with new close()/aclose() cleanup and an atexit hook to flush on shutdown.

Reworks async HTTP usage to reuse a lazily created shared httpx.AsyncClient, including loop-change detection to avoid reusing loop-bound locks across successive event loops.

Reduces unnecessary latency in Flow.run_sync() by only calling time.sleep(0.003) when no node results completed in the polling loop.

^{Written by Cursor Bugbot for commit 47ea564. This will update automatically on new commits. Configure here.}

[github-actions]

Coverage Report •

File	Stmts	Miss	Cover	Missing
dynamiq/clients
dynamiq.py	131	92	29%	22, 48–52, 55–58, 64–67, 75–79, 83–84, 86–91, 93–97, 101–104, 112–113, 118, 127–129, 137–140, 149, 152, 157, 160, 162–171, 174–177, 180–186, 188–192, 194, 196, 200–201, 206, 215–216, 224–225, 227–230, 232–234
dynamiq/flows
flow.py	232	58	75%	104, 111, 118, 152–153, 211–213, 235, 238, 244, 246–250, 257, 260, 266, 268, 270–271, 273, 275–277, 322, 430–434, 442–443, 457–458, 460, 479–480, 482, 485, 504–505, 508–509, 512–513, 515–516, 519–521, 523–527, 529
TOTAL	26747	8978	66%

Tests	Skipped	Failures	Errors	Time
1536	1 💤	0 ❌	0 🔥	1m 47s ⏱️

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

[cursor]

Await inside threading.Lock causes event loop deadlock

High Severity

_get_async_client calls await old_client.aclose() while holding self._async_init_mutex (a threading.Lock). When the await suspends the coroutine, any other coroutine scheduled on the same event loop that reaches with self._async_init_mutex: will call the blocking threading.Lock.acquire(), freezing the event loop thread entirely. Since the event loop is frozen, the original coroutine's aclose() can never complete — resulting in a deadlock. The await and associated cleanup need to happen outside the with self._async_init_mutex: block.

 

^{Triggered by project rule: Bugbot Rules for Dynamiq}