Repository: ray-project/ray
File: python/ray/data/_internal/cluster_autoscaler/default_autoscaling_coordinator.py (Python)

Code before:
        ] = lambda bundles: ray.autoscaler.sdk.request_resources(bundles=bundles),
        get_cluster_nodes: Callable[[], List[Dict]] = ray.nodes,
    ):
        self._get_current_time = get_current_time
        self._send_resources_request = send_resources_request
        self._get_cluster_nodes = get_cluster_nodes

        self._ongoing_reqs: Dict[str, OngoingRequest] = {}
        self._cluster_node_resources: List[ResourceDict] = []
        # Lock for thread-safe access to shared state from the background
        self._lock = threading.Lock()
        self._update_cluster_node_resources()

        # This is an actor, so the following check should always be True.
        # It's only needed for unit tests.
        if ray.is_initialized():
            # Start a thread to perform periodical operations.
            def tick_thread_run():
                while True:
                    time.sleep(self.TICK_INTERVAL_S)
                    self.tick()

            self._tick_thread = threading.Thread(target=tick_thread_run, daemon=True)
            self._tick_thread.start()

    def tick(self):
        """Used to perform periodical operations, e.g., purge expired requests,
        merge and send requests, check cluster resource updates, etc."""
        with self._lock:
            self._merge_and_send_requests()
            self._update_cluster_node_resources()
            self._reallocate_resources()

    def request_resources(
        self,
        requester_id: str,
        resources: List[ResourceDict],
        expire_after_s: float,
        request_remaining: bool = False,
        priority: ResourceRequestPriority = ResourceRequestPriority.MEDIUM,
    ) -> None:
        logger.debug("Received request from %s: %s.", requester_id, resources)
        with self._lock:
            # Round up the resource values to integers,
            # because the Autoscaler SDK only accepts integer values.
            for r in resources:
                for k in r:
                    r[k] = math.ceil(r[k])
            now = self._get_current_time()
            request_updated = False
            old_req = self._ongoing_reqs.get(requester_id)

Review comment (suggestion):
Nit: Since this an internal implementation detail, let's rename this with an underscore?
```suggestion
    def _tick(self):
```

Code after:
        ] = lambda bundles: ray.autoscaler.sdk.request_resources(bundles=bundles),
        get_cluster_nodes: Callable[[], List[Dict]] = ray.nodes,
    ):
        self._get_current_time = get_current_time
        self._send_resources_request = send_resources_request
        self._get_cluster_nodes = get_cluster_nodes

        self._ongoing_reqs: Dict[str, OngoingRequest] = {}
        self._cluster_node_resources: List[ResourceDict] = []
        # Lock for thread-safe access to shared state from the background
        self._lock = threading.Lock()
        self._update_cluster_node_resources()

        # This is an actor, so the following check should always be True.
        # It's only needed for unit tests.
        if ray.is_initialized():
            # Start a thread to perform periodical operations.
            def tick_thread_run():
                while True:
                    time.sleep(self.TICK_INTERVAL_S)
                    self._tick()

            self._tick_thread = threading.Thread(target=tick_thread_run, daemon=True)
            self._tick_thread.start()

    def _tick(self):
        """Used to perform periodical operations, e.g., purge expired requests,
        merge and send requests, check cluster resource updates, etc."""
        with self._lock:
            self._merge_and_send_requests()
            self._update_cluster_node_resources()
            self._reallocate_resources()

    def request_resources(
        self,
        requester_id: str,
        resources: List[ResourceDict],
        expire_after_s: float,
        request_remaining: bool = False,
        priority: ResourceRequestPriority = ResourceRequestPriority.MEDIUM,
    ) -> None:
        logger.debug("Received request from %s: %s.", requester_id, resources)
        with self._lock:
            # Round up the resource values to integers,
            # because the Autoscaler SDK only accepts integer values.
            for r in resources:
                for k in r:
                    r[k] = math.ceil(r[k])
            now = self._get_current_time()
            request_updated = False
            old_req = self._ongoing_reqs.get(requester_id)