perf(storage): implement fast-path for queue delivery in _StreamMultiplexer

[zhixiangli]

Summary

This PR optimizes the message delivery logic in _StreamMultiplexer to reduce latency and event loop overhead.

Performance Improvements:

1. Fast-path Delivery: Implemented a "fast-path" that attempts queue.put_nowait(item) for all target queues. For queues with available capacity, this is a synchronous operation that avoids:
   • Creating and scheduling a coroutine.
   • Yielding to the event loop.
   • Overhead associated with asyncio.wait_for.
2. Single-Queue Slow-path Optimization: In cases where exactly one queue is full, the multiplexer now directly awaits the _put_with_timeout coroutine. This bypasses the overhead of asyncio.gather, which is now only used when multiple queues are full simultaneously.
3. Reduced Event Loop Pressure: By minimizing the number of tasks created and yields performed during high-throughput streaming, these changes help the multiplexer keep up with fast-arriving gRPC responses.

[gemini-code-assist]

Code Review

This pull request optimizes the delivery of items to multiple queues in the _stream_multiplexer.py module. It introduces a new _put_to_queues method that implements a fast path using put_nowait for available queues and a slow path with a timeout for full queues, reducing unnecessary overhead from asyncio.gather and coroutine yields when queues have capacity. The _recv_loop has been refactored to use this optimized delivery mechanism. I have no feedback to provide as there were no review comments to evaluate.