Repository: ray-project/ray
File: python/ray/data/_internal/execution/operators/actor_pool_map_operator.py (Python)

Code before:
                and (
                    time.time()
                    <= self._last_upscaled_at
                    + self._ACTOR_POOL_SCALE_DOWN_DEBOUNCE_PERIOD_S
                )
            ):
                # NOTE: To avoid spamming logs unnecessarily, debounce log is produced once
                #       per upscaling event
                if self._last_upscaled_at != self._last_downscaling_debounce_warning_ts:
                    logger.debug(
                        f"Ignoring scaling down request (request={config}; reason=debounced from scaling up at {self._last_upscaled_at})"
                    )
                    self._last_downscaling_debounce_warning_ts = self._last_upscaled_at

                return False

        return True

    def scale(self, req: ActorPoolScalingRequest) -> Optional[int]:
        # Verify request could be applied
        if not self._can_apply(req):
            return 0

        map_worker_cls_name = (
            (self.map_worker_cls_name + " ") if self.map_worker_cls_name else ""
        )

        if req.delta > 0:
            target_num_actors = req.delta
            logger.debug(
                f"Scaling up {map_worker_cls_name}actor pool by {target_num_actors} (reason={req.reason}, "
                f"{self.get_actor_info()})"
            )

            for _ in range(target_num_actors):
                actor, ready_ref = self._create_actor()
                self.add_pending_actor(actor, ready_ref)

            # Capture last scale up timestamp
            self._last_upscaled_at = time.time()

            return target_num_actors

        elif req.delta < 0:
            num_released = 0
            target_num_actors = abs(req.delta)

            for _ in range(target_num_actors):
                if self._remove_inactive_actor():
                    num_released += 1


Review comment (bug):
Felt a bit confused by this assignment because the variable is called "name" but it includes trailing whitespace. If I were to re-use this variable some months from now (e.g., adding another log), it could lead to unexpected results if I'm not aware of trailing whitespace.

If we make `map_worker_cls_name` a non-None `str`, I think we could just add the whitespace in the debug statement and keep the name as the name 

Code after:
            if (
                not config.force
                and self._last_upscaled_at is not None
                and (time.time() <= self._last_upscaled_at + self._debounce_period_s)
            ):
                # NOTE: To avoid spamming logs unnecessarily, debounce log is produced once
                #       per upscaling event
                if self._last_upscaled_at != self._last_downscaling_debounce_warning_ts:
                    logger.debug(
                        f"Ignoring scaling down request (request={config}; reason=debounced from scaling up at {self._last_upscaled_at})"
                    )
                    self._last_downscaling_debounce_warning_ts = self._last_upscaled_at

                return False

        return True

    def scale(self, req: ActorPoolScalingRequest) -> Optional[int]:
        # Verify request could be applied
        if not self._can_apply(req):
            return 0

        map_worker_cls_name = self.map_worker_cls_name

        if req.delta > 0:
            target_num_actors = req.delta
            logger.debug(
                f"Scaling up {map_worker_cls_name} actor pool by {target_num_actors} (reason={req.reason}, "
                f"{self.get_actor_info()})"
            )

            for _ in range(target_num_actors):
                actor, ready_ref = self._create_actor()
                self.add_pending_actor(actor, ready_ref)

            # Capture last scale up timestamp
            self._last_upscaled_at = time.time()

            return target_num_actors

        elif req.delta < 0:
            num_released = 0
            target_num_actors = abs(req.delta)

            for _ in range(target_num_actors):
                if self._remove_inactive_actor():
                    num_released += 1

            if num_released > 0:
                logger.debug(
                    f"Scaled down {map_worker_cls_name} actor pool by {num_released} "