[Feat] Support Mooncake Store backend by zhaohaidao · Pull Request #162 · TransferQueue/TransferQueue

zhaohaidao · 2025-12-29T04:30:29Z

Summary by CodeRabbit

Release Notes

New Features
- Added Mooncake distributed storage backend support
Tests
- Performance result(ref: https://www.yuque.com/raw?filekey=yuque%2F0%2F2025%2Fpy%2F23208217%2F1763363631876-d92d4e08-fc99-422b-b4a4-8ebd20be681c.py&from=https%3A%2F%2Fwww.yuque.com%2Fhaomingzi-lfse7%2Fhlx5g0%2Ftml8ke0zkgn6roey).

2025-12-29 12:28:21,431 - INFO - TQ MOONCAKE BANDWIDTH TEST SUMMARY
2025-12-29 12:28:21,431 - INFO - ============================================================
2025-12-29 12:28:21,431 - INFO - Total Data Size: 15.673828 GB
2025-12-29 12:28:21,431 - INFO - PUT Time: 7.90345335s
2025-12-29 12:28:21,431 - INFO - GET Time: 13.62884474s
2025-12-29 12:28:21,431 - INFO - PUT Throughput: 15.86529577 Gb/s
2025-12-29 12:28:21,431 - INFO - GET Throughput: 9.20038546 Gb/s

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

Copilot

Pull request overview

This PR adds support for MooncakeStore as a distributed storage backend for TransferQueue. MooncakeStore provides high-performance storage with demonstrated throughput of 15.87 Gb/s for PUT operations and 9.20 Gb/s for GET operations.

Key changes include:

Implementation of MooncakeStorageClient with batched put/get operations and special handling for bfloat16 tensors
Implementation of MooncakeStorageManager with configuration validation
Improvements to the base KVStorageManager including async put operations and better error handling

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
transfer_queue/storage/clients/mooncake_client.py	New storage client implementation for Mooncake with batched operations, tensor serialization, and pickle-based serialization for non-tensor values
transfer_queue/storage/managers/mooncake_manager.py	New storage manager that validates Mooncake-specific configuration (metadata_server, master_server_address, client_name)
transfer_queue/storage/managers/base.py	Enhanced KVStorageManager with async executor for put operations, improved validation and error handling, and bug fix for clear_data method signature
transfer_queue/storage/clients/init.py	Added MooncakeStorageClient to module exports
transfer_queue/storage/managers/init.py	Added MooncakeStorageManager to module exports
transfer_queue/storage/init.py	Added MooncakeStorageManager and YuanrongStorageManager to public API exports

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-12-29T06:59:26Z

+        if num_samples == 0:
+            return


This validation is redundant. If num_samples is 0, the function would have already returned at line 432-433 when checking if metadata.field_names is empty, since there would be no fields with zero samples. Consider removing this check or moving it before the data.batch_size check if there's a valid scenario where metadata has field names but zero samples.

Suggested change

if num_samples == 0:

return

Can put this check at the beginning of the function

Copilot · 2025-12-29T06:59:27Z

+class MooncakeStorageClient(TransferQueueStorageKVClient):
+    def __init__(self, config: dict[str, Any]):
+        if not MOONCAKE_STORE_IMPORTED:
+            raise ImportError("Mooncake Store not installed. Please install via: pip install mooncake-transfer-engine")


The error message could be more helpful by including the installation command. Consider updating to: "Mooncake Store not installed. Please install it using: pip install mooncake-transfer-engine" (note the period at the end and rephrasing for clarity).

Suggested change

raise ImportError("Mooncake Store not installed. Please install via: pip install mooncake-transfer-engine")

raise ImportError(

"Mooncake Store not installed. Please install it using: pip install mooncake-transfer-engine."

)

Copilot · 2025-12-29T06:59:27Z

+    def clear(self, keys: list[str]):
+        for key in keys:
+            ret = self._store.remove(key)
+            if ret != 0:
+                logger.warning(f"remove failed for key '{key}' with error code: {ret}")


The clear operation removes keys one at a time in a loop, which could be inefficient for large numbers of keys. If the MooncakeDistributedStore supports batch removal, consider implementing a batch operation similar to how put and get are batched. If not, at least document why individual removal is necessary or consider using asyncio for concurrent removals when this is called from an async context.

Copilot · 2025-12-29T06:59:27Z

+@StorageClientFactory.register("MooncakeStorageClient")
+class MooncakeStorageClient(TransferQueueStorageKVClient):
+    def __init__(self, config: dict[str, Any]):
+        if not MOONCAKE_STORE_IMPORTED:
+            raise ImportError("Mooncake Store not installed. Please install via: pip install mooncake-transfer-engine")
+
+        self.local_hostname = config.get("local_hostname", "localhost")
+        self.metadata_server = config.get("metadata_server")
+        self.global_segment_size = config.get("global_segment_size", 512 * 1024 * 1024)
+        self.local_buffer_size = config.get("local_buffer_size", 128 * 1024 * 1024)
+        self.protocol = config.get("protocol", "tcp")
+        self.device_name = config.get("device_name", "")
+        self.master_server_address = config.get("master_server_address")
+
+        if self.metadata_server is None:
+            raise ValueError("Missing 'metadata_server' in config")
+        if self.master_server_address is None:
+            raise ValueError("Missing 'master_server_address' in config")
+
+        self._store = MooncakeDistributedStore()
+        ret = self._store.setup(
+            self.local_hostname,
+            self.metadata_server,
+            self.global_segment_size,
+            self.local_buffer_size,
+            self.protocol,
+            self.device_name,
+            self.master_server_address,
+        )
+        if ret != 0:
+            raise RuntimeError(f"Mooncake store setup failed with error code: {ret}")
+
+    def put(self, keys: list[str], values: list[Any]):
+        if not isinstance(keys, list) or not isinstance(values, list):
+            raise ValueError("keys and values must be lists")
+        if len(keys) != len(values):
+            raise ValueError("Number of keys must match number of values")
+
+        tensor_keys = []
+        tensor_values = []
+        non_tensor_keys = []
+        non_tensor_values = []
+
+        for key, value in zip(keys, values, strict=True):
+            if isinstance(value, torch.Tensor):
+                tensor = value.contiguous()
+                if tensor.device.type == "cuda":
+                    tensor = tensor.cpu()
+                tensor_keys.append(key)
+                tensor_values.append(tensor)
+            else:
+                non_tensor_keys.append(key)
+                non_tensor_values.append(pickle.dumps(value))
+
+        if tensor_keys:
+            self._batch_put_tensors(tensor_keys, tensor_values)
+
+        if non_tensor_keys:
+            self._batch_put_bytes(non_tensor_keys, non_tensor_values)
+
+    def _batch_put_tensors(self, keys: list[str], tensors: list[Tensor]):
+        for i in range(0, len(keys), BATCH_SIZE_LIMIT):
+            batch_keys = keys[i : i + BATCH_SIZE_LIMIT]
+            batch_tensors = tensors[i : i + BATCH_SIZE_LIMIT]
+
+            batch_values = []
+            for tensor in batch_tensors:
+                if tensor.dtype == torch.bfloat16:
+                    bytes_data = tensor.detach().view(torch.int16).numpy().tobytes()
+                else:
+                    bytes_data = tensor.detach().numpy().tobytes()
+                batch_values.append(bytes_data)
+
+            ret = self._store.put_batch(batch_keys, batch_values)
+            if ret != 0:
+                raise RuntimeError(f"put_batch failed with error code: {ret}")
+
+    def _batch_put_bytes(self, keys: list[str], values: list[bytes]):
+        for i in range(0, len(keys), BATCH_SIZE_LIMIT):
+            batch_keys = keys[i : i + BATCH_SIZE_LIMIT]
+            batch_values = values[i : i + BATCH_SIZE_LIMIT]
+
+            ret = self._store.put_batch(batch_keys, batch_values)
+            if ret != 0:
+                raise RuntimeError(f"put_batch failed with error code: {ret}")
+
+    def get(self, keys: list[str], shapes=None, dtypes=None) -> list[Any]:
+        if shapes is None or dtypes is None:
+            raise ValueError("MooncakeStorageClient needs shapes and dtypes")
+        if not (len(keys) == len(shapes) == len(dtypes)):
+            raise ValueError("Lengths of keys, shapes, dtypes must match")
+
+        tensor_indices = []
+        non_tensor_indices = []
+
+        for i, dtype in enumerate(dtypes):
+            if dtype is not None:
+                tensor_indices.append(i)
+            else:
+                non_tensor_indices.append(i)
+
+        results = [None] * len(keys)
+
+        if tensor_indices:
+            tensor_keys = [keys[i] for i in tensor_indices]
+            tensor_shapes = [shapes[i] for i in tensor_indices]
+            tensor_dtypes = [dtypes[i] for i in tensor_indices]
+            tensor_results = self._batch_get_tensors(tensor_keys, tensor_shapes, tensor_dtypes)
+            for idx, tensor in zip(tensor_indices, tensor_results, strict=True):
+                results[idx] = tensor
+
+        if non_tensor_indices:
+            non_tensor_keys = [keys[i] for i in non_tensor_indices]
+            non_tensor_results = self._batch_get_bytes(non_tensor_keys)
+            for idx, data in zip(non_tensor_indices, non_tensor_results, strict=True):
+                results[idx] = pickle.loads(data)
+
+        return results
+
+    def _batch_get_tensors(self, keys: list[str], shapes: list, dtypes: list) -> list[Tensor]:
+        tensors = [None] * len(keys)
+
+        for i in range(0, len(keys), BATCH_SIZE_LIMIT):
+            batch_keys = keys[i : i + BATCH_SIZE_LIMIT]
+            batch_shapes = shapes[i : i + BATCH_SIZE_LIMIT]
+            batch_dtypes = dtypes[i : i + BATCH_SIZE_LIMIT]
+
+            batch_results = self._store.get_batch(batch_keys)
+
+            if len(batch_results) != len(batch_keys):
+                raise RuntimeError(f"get_batch returned {len(batch_results)} items, expected {len(batch_keys)}")
+
+            for j, (raw_bytes, shape, dtype) in enumerate(zip(batch_results, batch_shapes, batch_dtypes, strict=True)):
+                if dtype == torch.bfloat16:
+                    tensors[i + j] = torch.frombuffer(raw_bytes, dtype=torch.int16).view(shape).view(torch.bfloat16)
+                else:
+                    tensors[i + j] = torch.frombuffer(raw_bytes, dtype=dtype).view(shape)
+
+        return tensors
+
+    def _batch_get_bytes(self, keys: list[str]) -> list[bytes]:
+        results = []
+        for i in range(0, len(keys), BATCH_SIZE_LIMIT):
+            batch_keys = keys[i : i + BATCH_SIZE_LIMIT]
+            batch_results = self._store.get_batch(batch_keys)
+            if len(batch_results) != len(batch_keys):
+                raise RuntimeError(f"get_batch returned {len(batch_results)} items, expected {len(batch_keys)}")
+            results.extend(batch_results)
+        return results
+
+    def clear(self, keys: list[str]):
+        for key in keys:
+            ret = self._store.remove(key)
+            if ret != 0:
+                logger.warning(f"remove failed for key '{key}' with error code: {ret}")
+
+    def close(self):
+        if self._store:
+            self._store.close()
+            self._store = None


No tests are provided for the new MooncakeStorageClient and MooncakeStorageManager. Following the pattern established in test_storage_client_factory.py for YuanrongStorageClient, please add tests that verify the client registration, initialization, and basic operations. At minimum, test that the client is properly registered with the factory and can be instantiated with valid configuration.

Copilot · 2025-12-29T06:59:27Z

        keys = self._generate_keys(data.keys(), metadata.global_indexes)
        values = self._generate_values(data)
-        self.storage_client.put(keys=keys, values=values)
+        loop = asyncio.get_event_loop()


Use asyncio.get_running_loop() instead of asyncio.get_event_loop(). The latter is deprecated in Python 3.10+ and can raise errors if there's no running event loop. Since this code is inside an async function, a loop is guaranteed to be running, so get_running_loop() is more appropriate and efficient.

Suggested change

loop = asyncio.get_event_loop()

loop = asyncio.get_running_loop()

0oshowero0 · 2025-12-29T07:07:47Z

+        for key, value in zip(keys, values, strict=True):
+            if isinstance(value, torch.Tensor):
+                tensor = value.contiguous()
+                if tensor.device.type == "cuda":


Mooncake store supports GPUDirect transfer (tensor in gpu -> host mem). Is it possble to support this feature in TQ?

Theoretically, it's supported, but I haven't put in the effort to investigate yet. Maybe I can make it a to-do?

0oshowero0 · 2025-12-29T07:17:32Z

+
+        for key, value in zip(keys, values, strict=True):
+            if isinstance(value, torch.Tensor):
+                tensor = value.contiguous()


Maybe put all of these tensor related operations into _batch_put_tensors?

I might not have get your point. Could you elaborate? This is just a categorization.

0oshowero0 · 2025-12-29T07:31:11Z

+                    bytes_data = tensor.detach().numpy().tobytes()
+                batch_values.append(bytes_data)
+
+            ret = self._store.put_batch(batch_keys, batch_values)


Can this API achieves better performance than the current implementation?

https://kvcache-ai.github.io/Mooncake/python-api-reference/mooncake-store.html#batch-put-tensor

You' re right. batch-put-tensor achieves better performance than batch_put, Unfortunately, get_batch_tensor did not show any improvement, even when using asyncio.

2026-01-02 17:40:57,316 - INFO - ============================================================
2026-01-02 17:40:57,316 - INFO - TQ MOONCAKE BANDWIDTH TEST SUMMARY
2026-01-02 17:40:57,316 - INFO - ============================================================
2026-01-02 17:40:57,317 - INFO - Total Data Size: 15.673828 GB
2026-01-02 17:40:57,317 - INFO - PUT Time: 5.46488714s
2026-01-02 17:40:57,317 - INFO - GET Time: 13.52535391s
2026-01-02 17:40:57,317 - INFO - PUT Throughput: 22.94477850 Gb/s
2026-01-02 17:40:57,317 - INFO - GET Throughput: 9.27078329 Gb/s
2026-01-02 17:40:57,317 - INFO - Network Round-trip Throughput: 13.20579604 Gb/s
2026-01-02 17:40:57,375 - INFO - TQ Mooncake bandwidth test completed successfully!

0oshowero0 · 2025-12-29T08:03:32Z

            return TensorDict({}, batch_size=len(metadata))
        keys = self._generate_keys(metadata.field_names, metadata.global_indexes)
        shapes, dtypes = self._get_shape_type_list(metadata)
        values = self.storage_client.get(keys=keys, shapes=shapes, dtypes=dtypes)


Do we need to let get become async just as put?

Letting get asynchronous didn't yield any performance benefits, so I reverted the change. The bottleneck is likely in mooncake-store, not the usage pattern.

zhaohaidao · 2026-01-02T09:54:46Z

@0oshowero0 @dpj135 Comments are addressed. PTAL if u have time.

dpj135 · 2026-01-04T03:08:52Z

Nice Work

@NINGBENZHE

…4829) ### What does this PR do? - Update TQ to formal release version. - Fix the shallow copy bug for chunking `BatchMeta` https://gitcode.com/Ascend/TransferQueue/pull/2 - Fix race condition for modifying torch num_threads https://gitcode.com/Ascend/TransferQueue/pull/5 - More robust port binding https://gitcode.com/Ascend/TransferQueue/pull/3 - Optimize memory usage for zero-copy transfer TransferQueue/TransferQueue#163 - add check_data_production_status and check_consumption_status and support polling get metadata TransferQueue/TransferQueue#157 @NINGBENZHE - (alpha) Support Mooncake Store backend TransferQueue/TransferQueue#162 @zhaohaidao - (alpha) Support Ray RDT backend TransferQueue/TransferQueue#167 - Update docs. ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: ... - [x] 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### Checklist Before Submitting > [!IMPORTANT] > Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review. - [x] Read the [Contribute Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [x] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [x] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [x] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [ ] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. --------- Signed-off-by: 0oshowero0 <o0shower0o@outlook.com>

@NINGBENZHE

…erl-project#4829) ### What does this PR do? - Update TQ to formal release version. - Fix the shallow copy bug for chunking `BatchMeta` https://gitcode.com/Ascend/TransferQueue/pull/2 - Fix race condition for modifying torch num_threads https://gitcode.com/Ascend/TransferQueue/pull/5 - More robust port binding https://gitcode.com/Ascend/TransferQueue/pull/3 - Optimize memory usage for zero-copy transfer TransferQueue/TransferQueue#163 - add check_data_production_status and check_consumption_status and support polling get metadata TransferQueue/TransferQueue#157 @NINGBENZHE - (alpha) Support Mooncake Store backend TransferQueue/TransferQueue#162 @zhaohaidao - (alpha) Support Ray RDT backend TransferQueue/TransferQueue#167 - Update docs. ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: ... - [x] 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### Checklist Before Submitting > [!IMPORTANT] > Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review. - [x] Read the [Contribute Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [x] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [x] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [x] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [ ] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. --------- Signed-off-by: 0oshowero0 <o0shower0o@outlook.com>

* [recipe] feat: migrate `recipe` to the dedicated repo `verl-recipe` as a submodule (#4795) ### What does this PR do? This PR 1. migrates most recipes from the `recipe` directory to the dedicated repo [`verl-recipe`](https://github.com/verl-project/verl-recipe), 2. adds `verl-recipe` as a submodule, 3. adds instruction to update the submodule reference in the PR template, 4. migrates [`transfer_queue`](verl/experimental/transfer_queue), [`fully_async_policy`](verl/experimental/fully_async_policy), [`one_step_off_policy`](verl/experimental/one_step_off_policy) and [`vla`](verl/experimental/vla) to [`verl/experimental`](verl/experimental) since they are planned to be merged into the main library, 5. updates related CI and paths accordingly, 6. updates the README news and awesome projects about this migration, 7. forces into a single commit and tries its best to recognize `rename` to keep the commit history trackable. See the "conjugate" PR at https://github.com/verl-project/verl-recipe/pull/7. ### Test See the CI. ### Todo - [ ] Ignore the final PR commit in git blame if it shows up too frequently. * [model] fix: fix temp dtype (#4813) ### What does this PR do? - As title. Prevent temperature to be int. ### 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### 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](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [ ] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [ ] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. --------- Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * [vllm, sglang, rollout] fix: Fix a mistake when running run_qwen3_vl-30b-megatron.sh with latest verl and vllm0.12 (#4810) * [ckpt] feat: add checkpoint-engine abstraction (#4775) ### What does this PR do? Add Checkpoint Engine abstraction. #### Overview Checkpoint Engine is an unified abstract layer to synchronize weights between various training backends and inference backends. It provides three unified APIs: - send_weights: get named tensors from generator and send them in streaming manner. - receive_weights: return a tensor generator that yield named tensors in streaming manner. - get_weights: return a tensor generator that yield named tensors in streaming manner, used for each inference instance update weight independently from local cache (e.g share memory, disk). For more detail, see `verl/checkpoint_engine/README.md`. #### verl core <img width="640" height="167" alt="image" src="https://github.com/user-attachments/assets/fbd125d7-b461-4c89-9678-b95a2ef89c33" /> #### checkpoint engine <img width="1004" height="409" alt="checkpoint-engine" src="https://github.com/user-attachments/assets/fc263c1f-17b2-4579-9842-87b24e12abc7" /> * [doc, ci] fix: Update Ascend doc and fix e2e_ascend CI (#4816) ### What does this PR do? > Add **concise** overview of what this PR aims to achieve or accomplish. Reference related GitHub issues and PRs that help with the review. As title. ### 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. Not related. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. Not related. ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. Not related. ### 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](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [ ] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [ ] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. * [trainer] feat: VeOmniEngine supports qwen3_vl ulysses (#4806) ### What does this PR do? as title. ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: ... - [x] 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### Checklist Before Submitting > [!IMPORTANT] > Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review. - [x] Read the [Contribute Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [x] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) * [doc] chore: fix checkpoint engine image link (#4821) ### What does this PR do? As title * [hardware] fix: automatically set device for SFT case (#4828) ### What does this PR do? auto_set_device does not cover SFT case. > Add **concise** overview of what this PR aims to achieve or accomplish. Reference related GitHub issues and PRs that help with the review. ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: ... - [x] 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### Checklist Before Submitting > [!IMPORTANT] > Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review. - [x] Read the [Contribute Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [x] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [ ] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. * [data] feat: TransferQueue - Update TransferQueue version and docs (#4829) ### What does this PR do? - Update TQ to formal release version. - Fix the shallow copy bug for chunking `BatchMeta` https://gitcode.com/Ascend/TransferQueue/pull/2 - Fix race condition for modifying torch num_threads https://gitcode.com/Ascend/TransferQueue/pull/5 - More robust port binding https://gitcode.com/Ascend/TransferQueue/pull/3 - Optimize memory usage for zero-copy transfer https://github.com/TransferQueue/TransferQueue/pull/163 - add check_data_production_status and check_consumption_status and support polling get metadata https://github.com/TransferQueue/TransferQueue/pull/157 @NINGBENZHE - (alpha) Support Mooncake Store backend https://github.com/TransferQueue/TransferQueue/pull/162 @zhaohaidao - (alpha) Support Ray RDT backend https://github.com/TransferQueue/TransferQueue/pull/167 - Update docs. ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: ... - [x] 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### Checklist Before Submitting > [!IMPORTANT] > Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review. - [x] Read the [Contribute Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [x] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [x] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [x] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [ ] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. --------- Signed-off-by: 0oshowero0 <o0shower0o@outlook.com> * [doc] Update docs about fully_async_policy (#4826) ### What does this PR do? Update documentation about fully_async_policy and adjust the formatting of the table. --------- Co-authored-by: jsfanfanfan <2981866535@qq.com> * [ckpt] fix: FSDP save ckpt after validation (#4799) ### What does this PR do? This PR fixes a bug in the `save_checkpoint` function for FSDPEngine. In the original logic, if the model engine is used (`use_legacy_worker_impl=disable`), the `wake_up` function in `verl/workers/engine_workers.py` will be invoked during the rollout phase of each step, which will offload the model to CPU. Under normal circumstances, the `compute_log_prob` function called during the training phase can load the model back to GPU. However, the training process is not executed during the validation phase, leaving the model on the CPU. If a checkpoint is saved immediately after validation, it will trigger the following error: `AssertionError: Expects tensor to be on the compute device cuda:0, was on cpu.` <details> <summary>Details</summary> Script: ``` set -x python examples/data_preprocess/geo3k.py --local_dir ~/data/geo3k python -m verl.trainer.main_ppo \ algorithm.adv_estimator=grpo \ data.train_files=$HOME/data/geo3k/train.parquet \ data.val_files=$HOME/data/geo3k/test.parquet \ data.train_batch_size=512 \ data.max_prompt_length=1024 \ data.max_response_length=2048 \ data.filter_overlong_prompts=True \ data.truncation='error' \ data.image_key=images \ actor_rollout_ref.model.path=Qwen/Qwen2.5-VL-3B-Instruct \ actor_rollout_ref.actor.optim.lr=1e-6 \ actor_rollout_ref.model.use_remove_padding=True \ actor_rollout_ref.actor.ppo_mini_batch_size=128 \ actor_rollout_ref.actor.ppo_micro_batch_size_per_gpu=4 \ actor_rollout_ref.actor.use_kl_loss=True \ actor_rollout_ref.actor.kl_loss_coef=0.01 \ actor_rollout_ref.actor.kl_loss_type=low_var_kl \ actor_rollout_ref.actor.entropy_coeff=0 \ actor_rollout_ref.model.enable_gradient_checkpointing=True \ actor_rollout_ref.actor.fsdp_config.param_offload=False \ actor_rollout_ref.actor.fsdp_config.optimizer_offload=False \ actor_rollout_ref.rollout.log_prob_micro_batch_size_per_gpu=16 \ actor_rollout_ref.rollout.tensor_model_parallel_size=2 \ actor_rollout_ref.rollout.name=vllm \ actor_rollout_ref.rollout.gpu_memory_utilization=0.6 \ actor_rollout_ref.rollout.enable_chunked_prefill=False \ actor_rollout_ref.rollout.enforce_eager=False \ actor_rollout_ref.rollout.free_cache_engine=False \ actor_rollout_ref.rollout.n=5 \ actor_rollout_ref.ref.log_prob_micro_batch_size_per_gpu=16 \ actor_rollout_ref.ref.fsdp_config.param_offload=False \ algorithm.use_kl_in_reward=False \ trainer.use_legacy_worker_impl=disable \ trainer.critic_warmup=0 \ trainer.logger=['console','wandb'] \ trainer.project_name='verl_ci_grpo_example_geo3k' \ trainer.experiment_name='qwen2_5_vl_3b_function_rm' \ trainer.n_gpus_per_node=8 \ trainer.nnodes=1 \ trainer.log_val_generations=20 \ trainer.save_freq=5 \ trainer.test_freq=5 \ trainer.total_epochs=15 ``` Error: ``` (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ERROR:2026-01-05 07:35:49,128:Got error when executing task. (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) Traceback (most recent call last): (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "python/ray/_raylet.pyx", line 1890, in ray._raylet.execute_task (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "python/ray/_raylet.pyx", line 1998, in ray._raylet.execute_task (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "python/ray/_raylet.pyx", line 1897, in ray._raylet.execute_task (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "python/ray/_raylet.pyx", line 1825, in ray._raylet.execute_task.function_executor (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "python/ray/_raylet.pyx", line 4651, in ray._raylet.CoreWorker.run_async_func_or_coro_in_event_loop (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/lib/python3.12/concurrent/futures/_base.py", line 449, in result (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return self.__get_result() (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/lib/python3.12/concurrent/futures/_base.py", line 401, in __get_result (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) raise self._exception (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "python/ray/_raylet.pyx", line 4638, in async_func (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/ray/_private/async_compat.py", line 50, in wrapper (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return func(*args, **kwargs) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/ray/_private/function_manager.py", line 691, in actor_method_executor (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return method(__ray_actor, *args, **kwargs) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/ray/util/tracing/tracing_helper.py", line 463, in _resume_span (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return method(self, *_args, **_kwargs) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/opt/tiger/open_verl/verl/single_controller/ray/base.py", line 841, in func (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return getattr(self.worker_dict[key], name)(*args, **kwargs) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/opt/tiger/open_verl/verl/single_controller/base/decorator.py", line 456, in inner (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return func(*args, **kwargs) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/opt/tiger/open_verl/verl/utils/transferqueue_utils.py", line 314, in dummy_inner (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) output = func(*args, **kwargs) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/opt/tiger/open_verl/verl/workers/engine_workers.py", line 541, in save_checkpoint (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) self.actor.save_checkpoint(local_path, hdfs_path, global_step, max_ckpt_to_keep) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/opt/tiger/open_verl/verl/single_controller/base/decorator.py", line 456, in inner (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return func(*args, **kwargs) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/opt/tiger/open_verl/verl/utils/transferqueue_utils.py", line 314, in dummy_inner (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) output = func(*args, **kwargs) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/opt/tiger/open_verl/verl/workers/engine_workers.py", line 343, in save_checkpoint (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return self.engine.save_checkpoint(local_path, hdfs_path, global_step, max_ckpt_to_keep) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/opt/tiger/open_verl/verl/workers/engine/fsdp/transformer_impl.py", line 607, in save_checkpoint (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) self.checkpoint_manager.save_checkpoint( (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/opt/tiger/open_verl/verl/utils/checkpoint/fsdp_checkpoint_manager.py", line 238, in save_checkpoint (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) model_state_dict = self.model.state_dict() (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/nn/modules/module.py", line 2256, in state_dict (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) hook(self, prefix, keep_vars) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/utils/_contextlib.py", line 120, in decorate_context (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return func(*args, **kwargs) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/distributed/fsdp/_state_dict_utils.py", line 777, in _pre_state_dict_hook (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) _pre_state_dict_hook_fn[fsdp_state._state_dict_type]( (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/distributed/fsdp/_state_dict_utils.py", line 517, in _sharded_pre_state_dict_hook (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) _common_unshard_pre_state_dict_hook( (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/distributed/fsdp/_state_dict_utils.py", line 161, in _common_unshard_pre_state_dict_hook (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) _enter_unshard_params_ctx( (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/distributed/fsdp/_state_dict_utils.py", line 125, in _enter_unshard_params_ctx (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) fsdp_state._unshard_params_ctx[module].__enter__() (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/lib/python3.12/contextlib.py", line 137, in __enter__ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) return next(self.gen) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/distributed/fsdp/_unshard_param_utils.py", line 199, in _unshard_fsdp_state_params (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) _unshard(state, handle, computation_stream, computation_stream) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/distributed/fsdp/_runtime_utils.py", line 290, in _unshard (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ran_pre_unshard = handle.pre_unshard() (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) ^^^^^^^^^^^^^^^^^^^^ (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/distributed/fsdp/_flat_param.py", line 1303, in pre_unshard (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) self._check_on_compute_device(self.flat_param) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/distributed/fsdp/_flat_param.py", line 2582, in _check_on_compute_device (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) _p_assert( (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) File "/usr/local/lib/python3.12/dist-packages/torch/distributed/utils.py", line 159, in _p_assert (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) raise AssertionError(s) (WorkerDict pid=42417, ip=[fdbd:dccd:cdd2:2207::30f]) AssertionError: Expects tensor to be on the compute device cuda:0, was on cpu ``` </details> To fix this bug, this PR checks whether the model is located on the CPU before saving the checkpoint and loads it onto the GPU if that is the case. The same bug also exists in Megatron, which requires further fixes. --------- Co-authored-by: weidongliang.339 <weidongliang.339@bytedance.com> Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * [perf] feat: Add MFU for Qwen3-VL dense (#4753) ### What does this PR do? Add the _estimate_qwen3_vit_flop and _estimate_qwen3_vl_flops function to calculate the FLOPs of Qwen3-VL dense models. Update the test cases to verify the calculation accuracy of Qwen3-VL models. ### 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`, `cfg`, `reward` - 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 The following is the output result of running the test file. <img width="1271" height="152" alt="image" src="https://github.com/user-attachments/assets/2a3d426c-bd32-4369-9c07-c8a17c60e98b" /> > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### 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](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [ ] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) * [tool] fix: avoid nested ToolResponse in SandboxFusionTool (#4833) ### What does this PR do? Fix an incorrect double-wrapping of `ToolResponse` in `SandboxFusionTool.execute()`. - `execute_code()` already returns a `ToolResponse`, but `execute()` previously wrapped it again as `ToolResponse(text=result)`. - Since `ToolResponse.text` expects `str | None`, the old behavior could produce an invalid/nested response (or confusing stringified output). - This PR makes `execute()` return the `ToolResponse` directly when appropriate, and only wraps once when the worker returns a non-`ToolResponse` result. ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: - [x] 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`, `cfg`, `reward` - 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 - pre-commit: - `pre-commit install` - `pre-commit run --all-files --show-diff-on-failure --color=always` - Result: **Passed** (ruff/format/mypy/autogen-trainer-cfg/docstring/license/compileall) ### API and Usage Example No API changes. `SandboxFusionTool.execute()` still returns `tuple[ToolResponse, float, dict]`. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes - `verl/tools/sandbox_fusion_tools.py` - If the execution worker returns a `ToolResponse`, return it directly. - Otherwise, convert the result to `str` (or `None`) and wrap once as `ToolResponse(text=...)`. ### Checklist Before Submitting > [!IMPORTANT] > Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review. - [x] Read the Contribute Guide: https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md - [x] Apply pre-commit checks: `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update the documentation: https://github.com/volcengine/verl/tree/main/docs - Not needed for this small bug fix. - [ ] Add unit or end-to-end test(s) to the CI workflow: https://github.com/volcengine/verl/tree/main/.github/workflows - Not added. This change is a small correctness fix and is covered by existing type/validation expectations; pre-commit checks passed. - [ ] Once your PR is ready for CI, send a message in the `ci-request` channel: - https://verl-project.slack.com/archives/C091TCESWB1 - If not accessible: https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a - [ ] Recipe submodule update (if applicable). - Not applicable for this PR. Co-authored-by: winston <email@example.com> * [vllm] fix: fix error in vllm patch for diff vllm version and add ci for moe with fp8 rollout (#4824) ### What does this PR do? fix error in vllm patch for diff vllm version and add ci for moe with fp8 rollout ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: ... - [x] 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### 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](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [ ] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [ ] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. --------- Co-authored-by: Xue Huang <xueh@nvidia.com> * [algo] feat: add optimal token baseline and variance proxy (#4678) # Optimal Token Baseline ## Main feature - Register `AdvantageEstimator.OPTIMAL_TOKEN_BASELINE`. - Extend the DP actor to emit `sum_pi_squared`, expose `calculate_sum_pi_squared` and checkpointing toggles across configs, and add a reusable `calculate_sum_pi_squared_from_logits` function. - Introduce `compute_variance_proxy_metrics` to surface signal/total power/noise diagnostics during training. - Document the method in `docs/algo/otb.md` and ship an executable example at `examples/otb_trainer/run_qwen2_5-7b.sh`. ## Usage - Enable OTB by overriding config keys (OmegaConf overlay): ```yaml algorithm.adv_estimator: optimal_token_baseline actor_rollout_ref: actor: calculate_sum_pi_squared: true sum_pi_squared_checkpointing: false # optional for long contexts rollout: n: 8 ``` - Run the example script (adjust dataset paths & WandB project as needed): ```bash bash examples/otb_trainer/run_qwen2_5-7b.sh ``` - Monitor the new variance proxies in trainer logs: `variance_proxy/proxy1_signal_strength`, `proxy2_total_power`, `proxy3_pure_noise`. ## keyNote - `actor.calculate_sum_pi_squared` requires `actor_rollout_ref.model.use_fused_kernels=False`; fused kernels must surface logits before OTB can run there. - Group sampling is mandatory (`rollout.n > 1`); with single-rollout batches OTB collapses to vanilla returns. --- UPDATE(@tongyx361 ): `compute_sum_pi_squared` is changed to `calculate_sum_pi_squared` for consistency with `calculate_entropy`. --------- Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> Co-authored-by: Shawn/Yuxuan Tong <tongyuxuan361@gmail.com> * [megatron] fix: Fix error in megatron workers (#4832) ### What does this PR do? There is a bug in megatron_workers.py, 745 line is redundant and introduces a bug. It overwrites the estimated_flops and promised_flops calculated on lines 742-744. Also, the condition "vl" in func.__name__ is brittle as it relies on a naming convention. This could lead to silent miscalculations of MFU if a new vision-language model's estimation function is named differently. A more robust approach is to attempt calling the function with the extra arguments and handle the TypeError if it doesn't support them. ### 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### 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](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [ ] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [ ] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. * [misc] feat: delete unnecessary base class in agent loop worker and vLLMHttpServer (#4838) * [misc] feat: consolidate tensordict before dispatch (#4830) * [training_utils] fix: json encode error in filelogger (#4811) ### What does this PR do? - fix: json encode error in filelogger error message: "TypeError: Object of type int32 is not JSON serializable" - ensure it's not Tensor object when logging to metrics ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: ... - [x] 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### Checklist Before Submitting > [!IMPORTANT] > Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review. - [x] Read the [Contribute Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [x] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [x] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [x] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) to cover all the code. If not feasible, explain why: ... - [x] Once your PR is ready for CI, send a message in [the `ci-request` channel](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [x] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. Signed-off-by: zhuangqh <zhuangqhc@gmail.com> * [ckpt] chore: skip saving hf_checkpoint during megatron+lora training & add a separate lora merge script (#4839) ### What does this PR do? When using LoRA, MegatronCheckpointManager.save_checkpoint not only saves the adapter but also saves the huggingface checkpoint, which is unnecessary. This PR skips saving the huggingface checkpoint, and provides a separate script for merging the adapter. Relating to #4063 ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: https://github.com/volcengine/verl/pulls?q=is%3Apr+megatron+lora+save - [x] 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```bash python ./scripts/megatron_merge_lora.py \ --config-name='ppo_megatron_trainer' \ actor_rollout_ref.model.lora.adapter_path=$APAPTER_PATH \ ... # same config as your training script ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### Checklist Before Submitting > [!IMPORTANT] > Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review. - [x] Read the [Contribute Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [x] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) to cover all the code. If not feasible, explain why: ... - [x] Once your PR is ready for CI, send a message in [the `ci-request` channel](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) - [ ] If your PR is related to the `recipe` submodule, please also update the reference to the submodule commit via `git submodule update --remote` or `cd recipe && git pull origin main`. * [rollout, vllm] fix: accuracy issue in verl serve mode + vllm-ascend + dp + ep + tp scenarios (#4783) ### What does this PR do? Fix the accuracy issue in verl + vllm-ascend dp+ep+tp+server scenarios, issue:https://github.com/vllm-project/vllm-ascend/issues/5544 ### Checklist Before Starting - [x] Search for similar PRs. Paste at least one query link here: ... - [x] 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`, `cfg`, `reward` - 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 Tested GRPO on local NPU host: <img width="1047" height="117" alt="58274edd-d0d3-454c-8e39-3188f6f19e71" src="https://github.com/user-attachments/assets/dee7bf2f-6faf-4f44-a8b3-64670d5b1e10" /> ### Design & Code Changes Root cause analysis: currently, the version of Verl + Ascend NPU + vllm-ascend is [v0.11.0](https://verl.readthedocs.io/en/latest/ascend_tutorial/ascend_quick_start.html). In the vllm-ascend v0.11.0 code, the all2all backend (flashinfer_all2allv) is selected and updated to the vllm worker environment. However, verl's ExternalZeroMQDistributedExecutor does not pass this environment to the vllm worker processes like vllm's [RayDistributedExecutor](https://github.com/vllm-project/vllm/blob/0d4044edd85de30d7d4558aeea4d1e95c7c556d6/vllm/v1/executor/ray_executor.py#L337) backend does. Therefore, due to the all2all backend for vllm-ascend is wrong, and then there is a precision issue on vllm-ascend. Implementation: 1. In vLLMAsyncRollout, when initiating vllm work, if it's an NPU scenario, add the environment variables required by vllm-ascend. 2. Add vllm engine environment variables setting in rollout.yaml, supports setting by user. ### Checklist Before Submitting > [!IMPORTANT] > Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review. - [x] Read the [Contribute Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [x] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) Co-authored-by: FightingZhen --------- Signed-off-by: leo-pony <nengjunma@outlook.com> * [fsdp] feat: add validate process on trainer node when use_trainer_do_validate=True (#4683) ### What does this PR do? > Add **concise** overview of what this PR aims to achieve or accomplish. Reference related GitHub issues and PRs that help with the review. User Trainer node to do validate process when run mode on fully-async, It can save time for validate computing and reduce perf/time_of_step peak - add new use_trainer_do_validate on fully_async async_training config to decide whether using trainer node to do validate process - use_trainer_do_validate: default is false - It can improve performance of validate, such as in `dapo_7b_math_fsdp2_8_8.sh`, it can improve about 1X speed <img width="1440" height="608" alt="image" src="https://github.com/user-attachments/assets/436e481e-4f51-4e8e-ad08-b038b3f0e89d" /> <img width="1030" height="762" alt="image" src="https://github.com/user-attachments/assets/ed8e3237-d37d-4eff-b944-fb81ea63f87c" /> - optimized the `process_validation_metrics()` on `_validate()` process, when input datasets len=1444, it latency reduce from 150+s to 40+s <img width="2630" height="448" alt="image" src="https://github.com/user-attachments/assets/b6fb50bc-5856-49c1-91dc-f845e9c410b4" /> <img width="2504" height="518" alt="image" src="https://github.com/user-attachments/assets/b3b5f238-0c5e-4c63-9683-83f34d5a46fd" /> ### 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`, `cfg`, `reward` - 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 > For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc. - on test scripts such as `dapo_7b_math_fsdp2_8_8.sh` add `async_training.use_trainer_do_validate=True` command to do compute - the result of this function on Qwen2.5-Math-7B model - the baseline scripts is `dapo_7b_math_fsdp2_8_8.sh` - the optimized scripts is `dapo_7b_math_fsdp2_8_8.sh` +`async_training.use_trainer_do_validate=True` - the acc and perfomance is below: <img width="1650" height="702" alt="image" src="https://github.com/user-attachments/assets/3419d7bb-a64c-4fe9-b776-3312925f51ab" /> <img width="1580" height="522" alt="image" src="https://github.com/user-attachments/assets/2c3a7e24-7421-4f12-8527-7b997f9c3b89" /> - green: optimized case (`async_training.use_trainer_do_validate=True` ) - gray: baseline case (`async_training.use_trainer_do_validate=False` ) ### API and Usage Example > Demonstrate how the API changes if any, and provide usage example(s) if possible. ```python # Add code snippet or script demonstrating how to use this async_training.use_trainer_do_validate=True \ ``` ### Design & Code Changes > Demonstrate the high-level design if this PR is complex, and list the specific changes. ### 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](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md). - [ ] Apply [pre-commit checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting): `pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always` - [ ] Add / Update [the documentation](https://github.com/volcengine/verl/tree/main/docs). - [ ] Add unit or end-to-end test(s) to [the CI workflow](https://github.com/volcengine/verl/tree/main/.github/workflows) 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](https://verl-project.slack.com/archives/C091TCESWB1) in [the `verl` Slack workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ). (If not accessible, please try [the Feishu group (飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).) --------- Co-authored-by: Shangwei-Li <lishangwei@mail.ustc.edu.cn> * [misc] fix: recipe submodule accidentally been removed (#4843) ### What does this PR do? As title. * [worker, training_utils] fix: Engine Metric Aggregation (#4778) ### What does this PR do? Because some metrics are scaled by global_bsz/global_tokens in `workers.utils.losses.ppo_loss`, the mean in `reduce_metrics` adds an extra scaling of the metric by the number of gradient accumulation steps (see examples in Test sec): https://github.com/volcengine/verl/blob/c191c5eb5c9499dca6666a52bc5f360bfd4bbf4f/verl/utils/metric/utils.py#L53 Aggregation of the `loss` metric handles this by taking sum: https://github.com/volcengine/verl/blob/c191c5eb5c9499dca6666a52bc5f360bfd4bbf4f/verl/workers/engine_workers.py#L143-L144 Depending on how metrics are handled in `workers.utils.losses.ppo_loss`, it may not be correct to aggregate all of them using sum (as in #4785). For example, `actor/pg_loss` and `actor/kl_loss` are scaled by global batch sizes/ token counts, and should be aggregated using sum, while the `pg_metrics` `pg_clipfrac`, `ppo_kl`, and `pg_clipfrac_lower` are scaled by local token counts and should be aggregated using mean. This PR introduces a metric management class to allow flexibility in deciding the aggregation type on a per-metric basis. ### Test This test demonstrates the scaling of metrics with the number of gradient accumulation steps, as well as how this is resolved on this branch. The command for running is below. <img width="980" height="638" alt="image" src="https://github.com/user-attachments/assets/e65ab291-3125-4df4-a0e0-3473bf64cb2a" /> ```bash gsm8k_train_path=$DATA_DIR/gsm8k/train.parquet gsm8k_test_path=$DATA_DIR/gsm8k/test.parquet train_files="['$gsm8k_train_path']" test_files="['$gsm8k_test_path']" ppo_micro_batch_size_per_gpu=2 ppo_micro_batch_size_per_gpu=8 branch=main branch=fixEngineMetrics python3 -m verl.trainer.main_ppo \ algorithm.adv_estimator=grpo \ data.dataloader_num_workers=0 \ data.return_full_prompt=True \ data.train_files="$train_files" \ data.val_files="$test_files" \ data.train_batch_size=8 \ data.max_prompt_length=512 \ data.max_response_length=1024 \ data.filter_overlong_prompts=True \ data.truncation='error' \ actor_rollout_ref.model.path=Qwen/Qwen3-0.6B \ actor_rollout_ref.actor.optim.lr=1e-6 \ actor_rollout_ref.model.use_remove_padding=True \ actor_rollout_ref.actor.ppo_mini_batch_size=8 \ actor_rollout_ref.actor.ppo_micro_batch_size_per_gpu=$ppo_micro_batch_size_per_gpu \ actor_rollout_ref.actor.use_kl_loss=True \ actor_rollout_ref.actor.fsdp_config.param_offload=True \ actor_rollout_ref.actor.fsdp_config.optimizer_offload=True \ actor_rollout_ref.rollout.tensor_model_parallel_size=1 \ actor_rollout_ref.rollout.log_prob_micro_batch_size_per_gpu=8 \ actor_rollout_ref.ref.log_prob_micro_batch_size_per_gpu=8 \ actor_rollout_ref.rollout.name=vllm \ actor_rollout_ref.rollout.gpu_memory_utilization=0.6 \ actor_rollout_ref.rollout.n=5 \ trainer.logger='["console","wandb"]' \ trainer.project_name='fixEngineMetrics' \ trainer.experiment_name="$branch/ppo_micro_batch_size_per_gpu$ppo_micro_batch_size_per_gpu" \ trainer.n_gpus_per_node=2 \ trainer.nnodes=1 \ trainer.save_freq=400 \ trainer.test_freq=40 \ trainer.use_legacy_worker_impl=disable \ trainer.total_epochs=2 \ trainer.total_training_steps=10 \ trainer.resume_mode=disable \ actor_rollout_ref.actor.use_torch_compile=False \ actor_rollout_ref.actor.fsdp_config.use_torch_compile=False \ trainer.val_before_train=False \ actor_rollout_ref.rollout.enforce_eager=True \ actor_rollout_ref.ref.fsdp_config.use_torch_compile=False ``` ### Design & Code Changes Adds a `Metric` class which tracks metric values and aggregation type. * [rollout] fix: configurable agent loop + multimodal data for fully-async (#4842) ## Description * **`verl/experimental/fully_async_policy/agent_loop/agent_loop.py`** * Use `config.agent.default_agent_loop` as the default `agent_name` when `agent_name` is not present in `batch.non_tensor_batch`. * Pass `dataset_cls=self.dataset_cls` and `dataset_config=self.config.data` into `hydra.utils.instantiate(...)` when creating an agent loop instance. * **`verl/experimental/fully_async_policy/agent_loop/partial_tool_agent_loop.py`** * Extract `video_data` from `multi_modal_data` and include `video_data` in the created `AgentData` instance (in addition to existing `image_data`). * **`verl/experimental/fully_async_policy/detach_utils.py`** * Stop popping original batch fields in `prepare_single_generation_data`. * Set `agent_name` to `async_partial_tool_agent` or `partial_single_turn_agent` depending on `config.actor_rollout_ref.rollout.multi_turn.enable`. ## Testing * Verified the fully async training entry can run successfully on 4 GPU server setup (multi-turn enabled, partial rollout enabled, vLLM async mode). ## Related * Fixes and extends the scope of: [4834](https://github.com/volcengine/verl/issues/4834) * [ci] test: switch the vlm rl test case in the npu environment to use the model engine (#4844) * [ckpt] fix: Megatron save ckpt after validation (#4841) ### What does this PR do? This PR fixes a bug in the `save_checkpoint` function for MegatronEngine. https://github.com/volcengine/verl/pull/4799 is a similar PR, which modifies FSDPEngine. In the original logic, if the model engine is used (`use_legacy_worker_impl=disable`), the `wake_up` function in `verl/workers/engine_workers.py` will be invoked during the rollout phase of each step, which will offload the model to CPU. Under normal circumstances, the `compute_log_prob` function called during the training phase can load the model back to GPU. However, the training process is not executed during the validation phase, leaving the model on the CPU. If a checkpoint is saved immediately after validation, it will trigger the following error: `AssertionError: Expects tensor to be on the compute device cuda:0, was on cpu.` To fix this bug, this PR checks whether the model is located on the CPU before saving the checkpoint and loads it onto the GPU if that is the case. --------- Co-authored-by: weidongliang.339 <weidongliang.339@bytedance.com> * [megatron] feat: Share actor and ref in LoRA (#4673) For `compute_ref_log_prob`, we can do that by disabling lora layers temporarily for the forward pass, as base weight are frozen and only lora layers are trained. This has already been supported in FSDP LoRA. ### What does this PR do? > Add **concise** overview of what this PR aims to achieve or accomplish. Reference …

赵海源 added 30 commits December 21, 2025 16:50

support mooncake-store as tq backend

cad2af5

fix

cb90030

add performance_test

dd35034

fix

7dbc81d

fix

51df082

fix

f465d9b

mc batch put/get

347364c

use batch api

46d6889

add mooncake_config.yaml

e6a3939

profiling put

4c2df53

fix

bc2b590

fix

14d5916

fix

1ca1d0f

fix

4da986d

profile other overhead

b03e457

profile get overhead

26d244b

optmize get_batch

7b710e4

optmize get_batch

6bb4cdd

fix

18bbe29

add mooncake_store_bench_python.py

52040cc

fix

a844d34

fix

6898ad3

fix

1f8eba3

use put_batch for get op

74bdaa3

get no dup

e19ff0e

fix get

2d7575f

fix get

f0c45c3

fix get

96a8d38

fix get

17c9d0e

fix get

d24bc4a

zhaohaidao changed the title ~~(WIP) mooncake backend~~ mooncake backend Dec 29, 2025

zhaohaidao requested a review from 0oshowero0 December 29, 2025 06:39

0oshowero0 changed the title ~~mooncake backend~~ [Feat] Support MooncakeStore backend Dec 29, 2025

0oshowero0 requested review from Copilot and removed request for 0oshowero0 December 29, 2025 06:54

Copilot started reviewing on behalf of 0oshowero0 December 29, 2025 06:54 View session

0oshowero0 changed the title ~~[Feat] Support MooncakeStore backend~~ [Feat] Support Mooncake Store backend Dec 29, 2025

Copilot AI reviewed Dec 29, 2025

View reviewed changes

0oshowero0 reviewed Dec 29, 2025

View reviewed changes

Comment thread transfer_queue/storage/clients/mooncake_client.py Outdated

0oshowero0 reviewed Dec 29, 2025

View reviewed changes

Comment thread transfer_queue/storage/clients/mooncake_client.py

0oshowero0 reviewed Dec 29, 2025

View reviewed changes

dpj135 reviewed Dec 30, 2025

View reviewed changes

Comment thread transfer_queue/storage/managers/base.py Outdated

Comment thread transfer_queue/storage/managers/base.py Outdated

dpj135 reviewed Dec 30, 2025

View reviewed changes

Comment thread transfer_queue/storage/managers/mooncake_manager.py

0oshowero0 mentioned this pull request Dec 31, 2025

[TODO] TODO List #128

Open

22 tasks

赵海源 added 3 commits January 2, 2026 16:15

address comments

7bcadee

replace put_batch/get_batch with batch_put_tensor/batch_get_tensor

1b07b03

address comments

9a3fc7a

Merge branch 'dev' into mcr

9832a13

0oshowero0 approved these changes Jan 4, 2026

View reviewed changes

0oshowero0 merged commit 58f98e0 into dev Jan 4, 2026
3 checks passed

0oshowero0 mentioned this pull request Jan 7, 2026

[data] feat: TransferQueue - Update TransferQueue version and docs verl-project/verl#4829

Merged

8 tasks

0oshowero0 mentioned this pull request Jan 14, 2026

[TODO] Long-term TODO list Ascend/TransferQueue#2

Closed

22 tasks

	loop = asyncio.get_event_loop()
	loop = asyncio.get_running_loop()

Conversation

zhaohaidao commented Dec 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary by CodeRabbit

Release Notes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Copilot AI Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

zhaohaidao commented Jan 2, 2026

Uh oh!

Uh oh!

dpj135 commented Jan 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

zhaohaidao commented Dec 29, 2025 •

edited

Loading