feat: async encoder cache impl

[furionw]

This PR

We implemented AsyncEncoderCache that

• wraps EncoderCacheManager
• optionally, triggers computation on the get_or_compute function

This will be used in

• disagg EPD setting where P wants to either retrieve tensor locally, or sends RPC to E
• Encode to offload tensor to G2
• agg worker to retrieve tensor from G2

Test Plan

pytest components/tests/dynamo/common/multimodal/ -v

Summary by CodeRabbit

Release Notes

• New Features

  • Enhanced multimodal encoder caching with concurrent request deduplication. When multiple simultaneous requests target the same data, computation occurs once with results shared across all requesters, improving efficiency.

• Tests

  • Comprehensive test coverage added for multimodal encoder caching, including concurrent request handling, error scenarios, and performance metrics.

_{✏️ Tip: You can customize this high-level summary in your review settings.}

[coderabbitai]

Walkthrough

This PR introduces AsyncEncoderCache, an async wrapper around EncoderCacheManager with request coalescing for concurrent access patterns. Multiple concurrent requests for the same key execute only one compute operation while others await the shared result. Changes include the core implementation class, a module initializer for public API export, and comprehensive test coverage.

Changes

Cohort / File(s)	Summary
Core Implementation components/src/dynamo/common/multimodal/async_encoder_cache.py	New AsyncEncoderCache class with request coalescing. Methods: __init__(cache), get(key), async get_or_compute(key, compute_fn), and stats property. Uses asyncio.Future to deduplicate concurrent requests for the same key. Includes internal exception suppression helper and in-flight request tracking.
API Export components/src/dynamo/common/multimodal/__init__.py	New module initializer that re-exports AsyncEncoderCache via __all__ to establish public API. Includes SPDX header and module docstring.
Test Suite components/src/dynamo/common/tests/multimodal/test_async_encoder_cache.py	Comprehensive test module covering basic operations, request coalescing behavior, exception propagation and recovery, and statistics reporting. Organized into test classes for distinct scenarios with appropriate EncoderCacheManager fixtures.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Poem

🐰 When many ask for just the same, one hops to compute with care,
While others rest on futures bright, and wait without a spare!
AsyncEncoderCache coalesces all their pleas so fine,
Concurrent streams now flow as one—no duplicate design! 🚀✨

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Description check	⚠️ Warning	The PR description covers the overview and test plan but lacks specific details about where to start reviewing and does not explicitly reference related issues as required by the template structure.	Add a 'Where should the reviewer start?' section highlighting key files (e.g., async_encoder_cache.py, test_async_encoder_cache.py) and add a 'Related Issues' section with appropriate GitHub issue references.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title 'feat: async encoder cache impl' directly describes the main change—implementing AsyncEncoderCache with async capabilities. It is concise, specific, and accurately reflects the primary objective of the changeset.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}