Repository: ray-project/ray
File: python/ray/train/v2/_internal/execution/worker_group/worker_group.py (Python)

Code before:
            for callback in self._callbacks:
                callback.before_worker_group_start(worker_group_context)

            pg_handle = self._create_placement_group(
                self._train_run_context.scaling_config, worker_group_context
            )

            logger.info(
                f"Attempting to start training worker group of size {worker_group_context.num_workers} with "
                f"the following resources: [{worker_group_context.resources_per_worker}] * {worker_group_context.num_workers}"
            )

            # Wait for the placement group to be ready before proceeding
            # to create actors.
            # This could hang if the resources are not available, so we should
            # time out if this hangs for a while to try again with a different size.
            # For example, the controller may try to set a worker group size
            # based on stale information about cluster resources.
            pg_wait_start = time_monotonic()
            if not pg_handle.wait(self._worker_group_start_timeout_s):
                pg_handle.shutdown()
                raise WorkerGroupStartupTimeoutError(
                    num_workers=worker_group_context.num_workers
                )
            logger.debug(
                "[Train Worker Initialization] Placement group ready in "
                f"{time_monotonic() - pg_wait_start:.2f}s."
            )

            # TODO: Figure out ordering between these different calls/callbacks.
            worker_group_state_builder.with_placement_group_handle(pg_handle)

            # Initialize the synchronization actor on the driver node
            sync_actor = SynchronizationActor.options(
                scheduling_strategy=NodeAffinitySchedulingStrategy(
                    node_id=ray.get_runtime_context().get_node_id(),
                    soft=False,
                )
            ).remote(
                timeout_s=self._collective_timeout_s,
                warn_interval_s=self._collective_warn_interval_s,
            )
            worker_group_state_builder.with_sync_actor(sync_actor)

            create_workers_start = time_monotonic()
            workers = self._create_workers(
                worker_group_context.num_workers,
                pg_handle.placement_group,
                worker_group_context.resources_per_worker,
            )
            worker_group_state_builder.with_workers(workers)

Review comment (suggestion):
```suggestion
                "[Worker Group Initialization] Placement group created in "
```

Code after:
        ):
            for callback in self._callbacks:
                callback.before_worker_group_start(worker_group_context)

            pg_handle = self._create_placement_group(
                self._train_run_context.scaling_config, worker_group_context
            )

            logger.info(
                f"Attempting to start training worker group of size {worker_group_context.num_workers} with "
                f"the following resources: [{worker_group_context.resources_per_worker}] * {worker_group_context.num_workers}"
            )

            # Wait for the placement group to be ready before proceeding
            # to create actors.
            # This could hang if the resources are not available, so we should
            # time out if this hangs for a while to try again with a different size.
            # For example, the controller may try to set a worker group size
            # based on stale information about cluster resources.
            pg_wait_start = time_monotonic()
            if not pg_handle.wait(self._worker_group_start_timeout_s):
                pg_handle.shutdown()
                raise WorkerGroupStartupTimeoutError(
                    num_workers=worker_group_context.num_workers
                )
            logger.debug(
                "[Worker Group Initialization] Placement group ready in "
                f"{time_monotonic() - pg_wait_start:.2f}s."
            )

            # TODO: Figure out ordering between these different calls/callbacks.
            worker_group_state_builder.with_placement_group_handle(pg_handle)

            # Initialize the synchronization actor on the driver node
            sync_actor = SynchronizationActor.options(
                scheduling_strategy=NodeAffinitySchedulingStrategy(
                    node_id=ray.get_runtime_context().get_node_id(),
                    soft=False,
                )
            ).remote(
                timeout_s=self._collective_timeout_s,
                warn_interval_s=self._collective_warn_interval_s,
            )
            worker_group_state_builder.with_sync_actor(sync_actor)

            create_workers_start = time_monotonic()
            workers = self._create_workers(
                worker_group_context.num_workers,
                pg_handle.placement_group,
                worker_group_context.resources_per_worker,
            )