move file

Author: pengwu22

tests/utils/test_bucketed_weight_transfer.py

@@ -55,7 +55,7 @@ def _generate_weights(weight_specs, seed):
 # ---------------------------------------------------------------------------
 def _sender_fn(zmq_handle, weight_specs, seed, bucket_size_mb, use_shm):
     """Sender process: generate weights, move to GPU, send."""
-    from verl.workers.rollout.vllm_rollout.bucketed_weight_transfer import BucketedWeightSender
+    from verl.workers.rollout.bucketed_weight_transfer import BucketedWeightSender
 
     weights = _generate_weights(weight_specs, seed)
     sender = BucketedWeightSender(
@@ -68,7 +68,7 @@ def _sender_fn(zmq_handle, weight_specs, seed, bucket_size_mb, use_shm):
 
 def _receiver_fn(zmq_handle, use_shm, result_queue):
     """Receiver process: receive weights, send back (name, dtype, shape, checksum)."""
-    from verl.workers.rollout.vllm_rollout.bucketed_weight_transfer import BucketedWeightReceiver
+    from verl.workers.rollout.bucketed_weight_transfer import BucketedWeightReceiver
 
     device = torch.device("cuda:0")
     receiver = BucketedWeightReceiver(

tests/utils/test_shared_memory.py

@@ -18,7 +18,7 @@
 
 import torch
 
-from verl.workers.rollout.vllm_rollout.bucketed_weight_transfer import create_shared_memory, rebuild_shared_memory
+from verl.workers.rollout.bucketed_weight_transfer import create_shared_memory, rebuild_shared_memory
 
 
 class TestSharedMemory(unittest.TestCase):

…vllm_rollout/bucketed_weight_transfer.py …kers/rollout/bucketed_weight_transfer.pyverl/workers/rollout/vllm_rollout/bucketed_weight_transfer.py renamed to verl/workers/rollout/bucketed_weight_transfer.py verl/workers/rollout/vllm_rollout/bucketed_weight_transfer.py renamed to verl/workers/rollout/bucketed_weight_transfer.py

@@ -169,7 +169,7 @@ def _init_buffer(self):
 
             # Create unique name for shared memory
             shm_name = f"verl_weights_{uuid.uuid4().hex}"
-            shm = shared_memory.SharedMemory(name=shm_name, create=True, size=self.bucket_size)
+            shm = create_shared_memory(self.bucket_size, shm_name)
             buffer = torch.frombuffer(shm.buf, dtype=torch.uint8)
 
             comm_metadata = {"name": shm_name, "size": self.bucket_size}

verl/workers/rollout/vllm_rollout/utils.py

@@ -156,7 +156,7 @@ def update_weights_from_ipc(self, peft_config: dict = None, base_sync_done=False
         """Update the weights of the rollout model."""
         from vllm.platforms import current_platform
 
-        from verl.workers.rollout.vllm_rollout.bucketed_weight_transfer import BucketedWeightReceiver
+        from verl.workers.rollout.bucketed_weight_transfer import BucketedWeightReceiver
 
         if current_platform.device_type == "npu" and self.device is None:
             self.device = torch.device(f"npu:{self.local_rank}")

verl/workers/rollout/vllm_rollout/vllm_rollout.py

@@ -41,7 +41,7 @@
 from verl.utils.device import get_device_id, is_support_ipc
 from verl.workers.config import HFModelConfig, RolloutConfig
 from verl.workers.rollout.base import BaseRollout
-from verl.workers.rollout.vllm_rollout.bucketed_weight_transfer import BucketedWeightSender
+from verl.workers.rollout.bucketed_weight_transfer import BucketedWeightSender
 from verl.workers.rollout.vllm_rollout.utils import get_device_uuid
 
 logger = logging.getLogger(__file__)