python/ray/util/client/dataclient.py
def _process_response(self, response: Any) -> None:
        """
        Process responses from the data servicer.
        """
        if response.req_id == 0:
            # This is not being waited for.
            logger.debug(f"Got unawaited response {response}")
            return
        if response.req_id in self.asyncio_waiting_data:
            can_remove = True
            try:
                callback = self.asyncio_waiting_data[response.req_id]
                if isinstance(callback, ChunkCollector):
                    can_remove = callback(response)
                elif callback:
                    callback(response)
                if can_remove:
                    # NOTE: calling del self.asyncio_waiting_data results
                    # in the destructor of ClientObjectRef running, which
                    # calls ReleaseObject(). So self.asyncio_waiting_data
                    # is accessed without holding self.lock. Holding the
                    # lock shouldn't be necessary either.
                    del self.asyncio_waiting_data[response.req_id]
            except Exception:
                logger.exception("Callback error:")
            with self.lock:
                # Update outstanding requests
                if response.req_id in self.outstanding_requests and can_remove:
                    del self.outstanding_requests[response.req_id]
                    # Acknowledge response
                    self._acknowledge(response.req_id)
        else:
            with self.lock:
                self.ready_data[response.req_id] = response
                self.cv.notify_all()

        Process responses from the data servicer.
        
Process responses from the data servicer.
Process responses from the data servicer .
def _process_response(self, response: Any) -> None:
        if response.req_id == 0:
            logger.debug(f"Got unawaited response {response}")
            return
        if response.req_id in self.asyncio_waiting_data:
            can_remove = True
            try:
                callback = self.asyncio_waiting_data[response.req_id]
                if isinstance(callback, ChunkCollector):
                    can_remove = callback(response)
                elif callback:
                    callback(response)
                if can_remove:
                    del self.asyncio_waiting_data[response.req_id]
            except Exception:
                logger.exception("Callback error:")
            with self.lock:
                if response.req_id in self.outstanding_requests and can_remove:
                    del self.outstanding_requests[response.req_id]
                    self._acknowledge(response.req_id)
        else:
            with self.lock:
                self.ready_data[response.req_id] = response
                self.cv.notify_all()
def _process_response ( self , response : Any ) -> None : if response . req_id == 0 : logger . debug ( f"Got unawaited response {response}" ) return if response . req_id in self . asyncio_waiting_data : can_remove = True try : callback = self . asyncio_waiting_data [ response . req_id ] if isinstance ( callback , ChunkCollector ) : can_remove = callback ( response ) elif callback : callback ( response ) if can_remove : del self . asyncio_waiting_data [ response . req_id ] except Exception : logger . exception ( "Callback error:" ) with self . lock : if response . req_id in self . outstanding_requests and can_remove : del self . outstanding_requests [ response . req_id ] self . _acknowledge ( response . req_id ) else : with self . lock : self . ready_data [ response . req_id ] = response self . cv . notify_all ( )
Process responses from the data servicer.
Process responses from the data servicer .
"""
        Process responses from the data servicer.
        """ # This is not being waited for. # NOTE: calling del self.asyncio_waiting_data results # in the destructor of ClientObjectRef running, which # calls ReleaseObject(). So self.asyncio_waiting_data # is accessed without holding self.lock. Holding the # lock shouldn't be necessary either. # Update outstanding requests # Acknowledge response
self
response
Any