[megatron] fix: fix megatron sync_weights oom on user_trainer_do_validate mode (verl-project#4944)

### 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.

fix user_trainer_do_validate=True run megatron+VLLM process error. In
megatron case, this also can recive some improvments

<img width="1452" height="1162" alt="image"
src="https://github.com/user-attachments/assets/ac6c2d01-a48c-48a2-a5d6-0aab73e9d3e4"
/>
<img width="984" height="672" alt="image"
src="https://github.com/user-attachments/assets/ee3f99ca-0264-41e5-a80f-0d31783c21e7"
/>
<img width="424" height="724" alt="image"
src="https://github.com/user-attachments/assets/217a6090-b132-453a-805d-22f414fb2286"
/>
<img width="1160" height="616" alt="image"
src="https://github.com/user-attachments/assets/3110b313-024f-46c5-a08e-0b81a4a0f117"
/>

<img width="1574" height="1268" alt="image"
src="https://github.com/user-attachments/assets/3e12a53c-6dc7-4f33-b7d8-9016585d80ed"
/>


### 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`, `veomni`, `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`.

Author: chenjiaoAngel

verl/experimental/fully_async_policy/megatron_worker.py

@@ -57,13 +57,13 @@ def sync_rollout_weights(self, sync_group_name="actor_rollout"):
         assert (self._is_actor or self._is_rollout) and not self.config.hybrid_engine
         assert hasattr(self, "_weights_info") and self._weights_info is not None
         if self._is_actor and self._is_offload_param:
-            load_megatron_model_to_gpu(self.actor_module)
+            load_megatron_model_to_gpu(self.actor_module, False)
         params_generator = self._get_actor_params_generator() if self._is_actor else None
         params = {key: tensor for key, tensor in params_generator} if params_generator is not None else None
 
         rollout_name = self.config.rollout.name
         inference_model = None
-        if self._is_rollout:
+        if self._is_rollout and (not self._is_actor):
             if rollout_name == "vllm":
                 inference_model = BaseDetachNcclSync.get_inference_model(self.rollout)
                 from verl.utils.vllm.patch import patch_vllm_moe_model_weight_loader
@@ -136,7 +136,7 @@ def sync_rollout_weights_by_checkpoint(self, sync_group_name="actor_rollout"):
         # Load model to GPU
         load_start_time = time.time()
         if self._is_actor and self._is_offload_param:
-            load_megatron_model_to_gpu(self.actor_module)
+            load_megatron_model_to_gpu(self.actor_module, False)
         load_duration = time.time() - load_start_time
 
         from ray.util.collective import collective
@@ -158,7 +158,7 @@ def sync_rollout_weights_by_checkpoint(self, sync_group_name="actor_rollout"):
 
         rollout_name = self.config.rollout.name
         inference_model = None
-        if self._is_rollout:
+        if self._is_rollout and (not self._is_actor):
             if rollout_name == "vllm":
                 inference_model = BaseDetachNcclSync.get_inference_model(self.rollout)
                 from verl.utils.vllm.patch import patch_vllm_moe_model_weight_loader
@@ -243,7 +243,7 @@ def get_actor_weights_info(self):
         if hasattr(self, "_weights_info"):
             return self._weights_info
         if self._is_offload_param:
-            load_megatron_model_to_gpu(self.actor_module)
+            load_megatron_model_to_gpu(self.actor_module, False)
         params_generator = self._get_actor_params_generator()
         ret = []
         for key, tensor in params_generator: