Repository: ray-project/ray
File: python/ray/serve/_private/request_router/request_router.py (Python)

Code before:
        self._pending_requests_by_id[internal_id] = pending_request
        model_id = pending_request.metadata.multiplexed_model_id
        if model_id:
            self._pending_requests_by_model_id[model_id].append(pending_request)

    def _remove_pending_request_from_indices(self, pending_request: PendingRequest):
        """Remove a pending request from the dict indices."""
        internal_id = pending_request.metadata.internal_request_id
        self._pending_requests_by_id.pop(internal_id, None)
        # Note: We don't eagerly remove from _pending_requests_by_model_id here
        # because finding the item in the list would be O(n). Instead, we use lazy
        # cleanup in _get_pending_request_matching_multiplexed_model_id() which
        # removes done entries during lookup, amortizing the cleanup cost.

    def _insert_pending_request_sorted(
        self, queue: Deque[PendingRequest], pending_request: PendingRequest
    ):
        """Insert a pending request into the queue maintaining sorted order by created_at.

        Optimized for the common case where retries have recent timestamps and belong
        near the end of the queue. Uses O(1) append when possible, otherwise O(n) insert.
        """
        # Fast path: if queue is empty or request is newer than the last item, append.
        # This is O(1) and handles the common case where retries are recent.
        if len(queue) == 0 or pending_request.created_at >= queue[-1].created_at:
            queue.append(pending_request)
            return

        # Slow path: linear scan to find insertion point. This is O(n) but retries
        # requiring mid-queue insertion are rare, so we keep the simple approach.
        index = 0
        for pr in queue:
            if pending_request.created_at < pr.created_at:
                break
            index += 1
        queue.insert(index, pending_request)

    def on_new_queue_len_info(
        self, replica_id: ReplicaID, queue_len_info: ReplicaQueueLengthInfo
    ):
        """Update queue length cache with new info received from replica."""
        if self._use_replica_queue_len_cache:
            self._replica_queue_len_cache.update(
                replica_id, queue_len_info.num_ongoing_requests
            )
            self._update_router_queue_len_gauge(
                replica_id, queue_len_info.num_ongoing_requests
            )

    def on_send_request(self, replica_id: ReplicaID):
        """Increment queue length cache when a request is sent to a replica."""

Review comment (performance):
if these queues are sorted by `pending_request.created_at`, we should apply something like binary search on them as its complexity O(log n) would be more beneficial to us.

Code after:
        # because finding the item in the list would be O(n). Instead, we use lazy
        # cleanup in _get_pending_request_matching_multiplexed_model_id() which
        # removes done entries during lookup, amortizing the cleanup cost.

    def _insert_pending_request_sorted(
        self, queue: Deque[PendingRequest], pending_request: PendingRequest
    ):
        """Insert a pending request into the queue maintaining sorted order by created_at.

        Optimized for the common case where retries have recent timestamps and belong
        near the end of the queue. Uses O(1) append when possible, otherwise O(n) insert.
        """
        # Fast path: if queue is empty or request is newer than the last item, append.
        # This is O(1) and handles the common case where retries are recent.
        if len(queue) == 0 or pending_request.created_at >= queue[-1].created_at:
            queue.append(pending_request)
            return

        # Slow path: linear scan to find insertion point. This is O(n) but retries
        # requiring mid-queue insertion are rare, so we keep the simple approach.
        index = 0
        for pr in queue:
            if pending_request.created_at < pr.created_at:
                break
            index += 1
        queue.insert(index, pending_request)

    def on_new_queue_len_info(
        self, replica_id: ReplicaID, queue_len_info: ReplicaQueueLengthInfo
    ):
        """Update queue length cache with new info received from replica."""
        if self._use_replica_queue_len_cache:
            self._replica_queue_len_cache.update(
                replica_id, queue_len_info.num_ongoing_requests
            )
            self._update_router_queue_len_gauge(
                replica_id, queue_len_info.num_ongoing_requests
            )

    def on_send_request(self, replica_id: ReplicaID):
        """Increment queue length cache when a request is sent to a replica."""
        if self._use_replica_queue_len_cache:
            num_ongoing_requests = self._replica_queue_len_cache.get(replica_id) or 0
            new_queue_len = num_ongoing_requests + 1
            self._replica_queue_len_cache.update(replica_id, new_queue_len)
            self._update_router_queue_len_gauge(replica_id, new_queue_len)

    def update_replicas(self, replicas: List[RunningReplica]):
        """Update the set of available replicas to be considered for routing.

        When the set of replicas changes, we may spawn additional routing tasks