def read_async(self, ids: torch.Tensor):
        r"""Read the feature by index asynchronously.

        Parameters
        ----------
        ids : torch.Tensor
            The index of the feature. Only the specified indices of the
            feature are read.
        Returns
        -------
        A generator object.
            The returned generator object returns a future on
            ``read_async_num_stages(ids.device)``\ th invocation. The return result
            can be accessed by calling ``.wait()``. on the returned future object.
            It is undefined behavior to call ``.wait()`` more than once.

        Examples
        --------
        >>> import dgl.graphbolt as gb
        >>> feature = gb.Feature(...)
        >>> ids = torch.tensor([0, 2])
        >>> for stage, future in enumerate(feature.read_async(ids)):
        ...     pass
        >>> assert stage + 1 == feature.read_async_num_stages(ids.device)
        >>> result = future.wait()  # result contains the read values.
        """
        future = self._feature.query(
            ids if self._offset == 0 else ids + self._offset, async_op=True
        )

        yield

        values, missing_index, missing_keys = future.wait()

        fallback_reader = self._fallback_feature.read_async(
            missing_keys if self._offset == 0 else missing_keys - self._offset
        )
        fallback_num_stages = self._fallback_feature.read_async_num_stages(
            missing_keys.device
        )
        for i in range(fallback_num_stages):
            missing_values_future = next(fallback_reader, None)
            if i < fallback_num_stages - 1:
                yield  # fallback feature stages.

        class _Waiter:
            def __init__(
                self,
                feature,
                values,
                missing_index,
                missing_keys,
                missing_values_future,
            ):
                self.feature = feature
                self.values = values
                self.missing_index = missing_index
                self.missing_keys = missing_keys
                self.missing_values_future = missing_values_future

            def wait(self):
                """Returns the stored value when invoked."""
                missing_values = self.missing_values_future.wait()
                self.feature.replace(self.missing_keys, missing_values)
                self.values[self.missing_index] = missing_values
                values = self.values
                # Ensure there is no memory leak.
                self.feature = self.values = self.missing_index = None
                self.missing_keys = self.missing_values_future = None
                return values

        yield _Waiter(
            self._feature,
            values,
            missing_index,
            missing_keys,
            missing_values_future,
        )
Read the feature by index asynchronously.

        Parameters
        ----------
        ids : torch.Tensor
            The index of the feature. Only the specified indices of the
            feature are read.
        Returns
        -------
        A generator object.
            The returned generator object returns a future on
            ``read_async_num_stages(ids.device)``\ th invocation. The return result
            can be accessed by calling ``.wait()``. on the returned future object.
            It is undefined behavior to call ``.wait()`` more than once.

        Examples
        --------
        >>> import dgl.graphbolt as gb
        >>> feature = gb.Feature(...)
        >>> ids = torch.tensor([0, 2])
        >>> for stage, future in enumerate(feature.read_async(ids)):
        ...     pass
        >>> assert stage + 1 == feature.read_async_num_stages(ids.device)
        >>> result = future.wait()  # result contains the read values.
        
python/dgl/graphbolt/impl/gpu_cached_feature.py