[WIP][data] feat: TransferQueue - An asynchronous streaming data management system

[0oshowero0]

What does this PR do?

This PR introduces the TransferQueue data management module to verl, aiming to accelerate experience data transfer and address performance bottlenecks in post-training systems. Detailed design rationale is available in our RFC (#2662).

This PR adds TransferQueue as a git submodule into verl/experimental/transfer_queue. Besides, we provide end-to-end scripts that integrate verl with TransferQueue.

TransferQueue is a high-performance data storage and transfer module with panoramic data visibility and streaming scheduling capabilities, optimized for efficient dataflow in post-training workflows (in progress).

The system will introduce the following core components:

• TransferQueueClient: Deployed on each Worker, manages the communication with TransferQueue system via simple put/get semantics.

• TransferQueueController: Centralized dataflow scheduler tracking the production and consumption status of training samples.

• TransferQueueStorage: Distributed storage units that holds the actual experience data.

The primary motivation for integrating TransferQueue to verl now is to alleviate the data transfer bottleneck of the single controller RayPPOTrainer. Currently, all DataProto objects must be routed through RayPPOTrainer, resulting in a single point bottleneck of the whole post-training system.

Leveraging TransferQueue, we separate experience data transfer from metadata dispatch by

• Replacing DataProto with BatchMeta (metadata) and TensorDict (actual data) structures
• Preserving verl's original Dispatch/Collect logic via BatchMeta (maintaining single-controller debuggability)
• Accelerating data transfer by TransferQueue's distributed storage units

For WorkerGroup class, we hide the above translation process by decorator. For AgentLoop related class, we explicitely do the adaption in AgentLoopBase.

Checklist Before Starting

• Search for similar PRs. Paste at least one query link here: ...
• Format the PR title as [{modules}] {type}: {description} (This will be checked by the CI)
  • {modules} include fsdp, megatron, sglang, vllm, rollout, trainer, ci, training_utils, recipe, hardware, deployment, ray, worker, single_controller, misc, perf, model, algo, env, tool, ckpt, doc, data
  • If this PR involves multiple modules, separate them with , like [megatron, fsdp, doc]
  • {type} is in feat, fix, refactor, chore, test
  • If this PR breaks any API (CLI arguments, config, function signature, etc.), add [BREAKING] to the beginning of the title.
  • Example: [BREAKING][fsdp, megatron] feat: dynamic batching

Test

We've validated TransferQueue functionality through

• Unit test of (Async)TransferQueueClient, TransferQueueController, and TransferQueueSimpleUnit
• End-to-end demo that mimics the usage in verl

API and Usage Example

The primary interaction points are AsyncTransferQueueClient and TransferQueueClient, serving as the communication interface with the TransferQueue system.

Core client interfaces:

• (async_)get_meta(data_fields: list[str], batch_size:int, global_step:int, get_n_samples:bool, task_name:str) -> BatchMeta
• (async_)get_data(metadata:BatchMeta) -> TensorDict
• (async_)put(data:TensorDict, metadata:BatchMeta, global_step)
• (async_)clear(global_step: int)

You may refer to the example here, where we mimics the verl usage in both async & sync scenarios:
https://github.com/TransferQueue/TransferQueue/tree/dev/recipe/simple_use_case.

---

For verl integration, we put TransferQueue as a submodule in verl/experimental/transfer_queue. So please run the following git command first:

git submodule init
git submodule update

Then you can try our recipe (still in developing).

Design & Code Changes

Refer to our Paper, RFC, and Zhihu post :)

Checklist Before Submitting

Important

Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review.

• Read the Contribute Guide.
• Apply pre-commit checks: pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always
• Add / Update the documentation.
• Add unit or end-to-end test(s) to the CI workflow to cover all the code. If not feasible, explain why: ...
• Once your PR is ready for CI, send a message in the ci-request channel in the verl Slack workspace. (If not accessible, please try the Feishu group (飞书群).)

[0oshowero0]

Current Progress:

• BatchMeta can be successfully dispatched to each Worker
• BatchMeta <-> DataProto bidirectional conversion is ready, tested by generate_sequence when async_rollout_mode=False
• Support all WorkerGroup classes
• Support AgentLoopManager class (async_rollout_mode=True)
• RayPPOTrainer single controller adaption
• E2E long run