Rust~tokio的AsyncRead

AsyncRead trait 定义

介绍

/// This trait is analogous to the [`std::io::Read`] trait, but integrates with
/// the asynchronous task system. In particular, the [`poll_read`] method,
/// unlike [`Read::read`], will automatically queue the current task for wakeup
/// and return if data is not yet available, rather than blocking the calling
/// thread.
// 此特征与 [std::io::Read] 特征类似，但它与异步任务系统集成
// 具体而言，与 [Read::read] 不同，[poll_read] 方法在数据尚不可用时，会自动将当前任务排入唤醒队列并返回，而不是阻塞调用线程
///
/// Specifically, this means that the `poll_read` function will return one of
/// the following:
///
/// * `Poll::Ready(Ok(()))` means that data was immediately read and placed into
///   the output buffer. The amount of data read can be determined by the
///   increase in the length of the slice returned by `ReadBuf::filled`. If the
///   difference is 0, either EOF has been reached, or the output buffer had zero
///   capacity (i.e. `buf.remaining()` == 0).
// * Poll::Ready(Ok(())) 表示数据已立即被读取并放入输出缓冲区。
// 读取的数据量可以通过 ReadBuf::filled 返回的切片长度的增加来确定。
// 如果差值为 0，则表示要么已到达文件末尾（EOF），要么输出缓冲区的容量为零（即 buf.remaining() 等于 0）
///
/// * `Poll::Pending` means that no data was read into the buffer
///   provided. The I/O object is not currently readable but may become readable
///   in the future. Most importantly, **the current future's task is scheduled
///   to get unparked when the object is readable**. This means that like
///   `Future::poll` you'll receive a notification when the I/O object is
///   readable again.
// * Poll::Pending 表示没有数据被读入提供的缓冲区。该 I/O 对象当前不可读，但未来可能变得可读
// 最重要的是，当该对象变得可读时，当前 Future 的任务将被安排唤醒
// 这意味着就像 Future::poll 一样，当 I/O 对象再次可读时，你将收到通知
///
///
/// * `Poll::Ready(Err(e))` for other errors are standard I/O errors coming from the
///   underlying object 底层对象.
// * Poll::Ready(Err(e)) 表示其他错误，这些是来自底层对象的标准 I/O 错误

特征定义

pub trait AsyncRead {/// Attempts to read from the `AsyncRead` into `buf`.////// On success, returns `Poll::Ready(Ok(()))` and places data in the/// unfilled portion of `buf`. If no data was read (`buf.filled().len()` is/// unchanged), it implies that EOF has been reached.////// If no data is available for reading, the method returns `Poll::Pending`/// and arranges for the current task (via `cx.waker()`) to receive a/// notification when the object becomes readable or is closed.// unfilled portion: 未填充的部分// 当该对象变得可读或被关闭时，（程序）会安排当前任务（通过 cx.waker()）接收通知// arranges for...to...：表示 “安排…… 去做……”，在这里是安排当前任务做某事fn poll_read(self: Pin<&mut Self>,cx: &mut Context<'_>,buf: &mut ReadBuf<'_>,) -> Poll<io::Result<()>>;
}

示例1 - 字节切片

为 &[u8] 类型实现 AsyncRead 特征。这意味着可以将字节切片的引用当作一个异步读取器来使用

impl AsyncRead for &[u8] {fn poll_read(// Pin 类型用于确保对象在内存中的位置不会改变，不过对于 &[u8] 这种不可变引用，Pin 在这里并没有实际作用// mut 表示可以修改 self 的值，这里主要是为了后续能更新切片的起始位置mut self: Pin<&mut Self>,// Context 包含了任务的唤醒器，用于在数据准备好时唤醒任务继续执行// 但由于 &[u8] 的数据已经在内存中，不需要等待，所以这里使用 _ 忽略该参数_cx: &mut Context<'_>,// 是一个用于存储读取数据的缓冲区，数据将从 &[u8] 中读取到这个缓冲区中buf: &mut ReadBuf<'_>,) -> Poll<io::Result<()>> {// 计算要读取的字节数// self.len() 表示当前字节切片的长度// buf.remaining() 表示 ReadBuf 缓冲区剩余的可用空间// 使用 std::cmp::min 函数取两者中的较小值，确保不会读取超过切片长度或缓冲区容量的数据let amt = std::cmp::min(self.len(), buf.remaining());// 将当前字节切片 self 从第 amt 个字节处分割成两个切片 a 和 blet (a, b) = self.split_at(amt);// 将切片 a 中的数据复制到 ReadBuf 缓冲区中buf.put_slice(a);// 将 self 更新为剩余的字节切片 b，这样下次调用 poll_read 时，将从剩余的数据开始读取*self = b;// 由于数据已经立即读取并放入缓冲区，所以返回 Poll::Ready(Ok(())) 表示操作已完成且成功// 字节切片的引用可以无缝集成到异步读取的代码中Poll::Ready(Ok(()))}
}

用到 AsyncRead trait 的结构体和方法

ReadToEnd 结构体

pub struct ReadToEnd<'a, R: ?Sized> {reader: &'a mut R,buf: VecWithInitialized<&'a mut Vec<u8>>,// The number of bytes appended to buf. This can be less than buf.len() if// the buffer was not empty when the operation was started.read: usize,// Make this future `!Unpin` for compatibility with async trait methods.#[pin]_pin: PhantomPinned,
}

read_to_end方法

// 名为 read_to_end 的公共（在当前 crate 内部可见）函数，以在 crate 内的其他模块中使用，但对外部 crate 不可见
// 创建一个 ReadToEnd 结构体的实例，用于异步地将一个实现了 AsyncRead 特征的读取器中的所有数据读取到一个 Vec<u8> 缓冲区中
// 'a：生命周期参数，用于确保 reader 和 buffer 的引用在 ReadToEnd 结构体的生命周期内有效
// R：泛型类型参数，代表实现了 AsyncRead 特征的读取器类型
// reader: &'a mut R：可变引用，指向实现了 AsyncRead 特征的读取器
// buffer: &'a mut Vec<u8>：可变引用，指向用于存储读取数据的 Vec<u8> 缓冲区
// ReadToEnd<'a, R>：函数的返回类型，包含了读取器和缓冲区的引用，以及一些用于跟踪读取状态的信息
pub(crate) fn read_to_end<'a, R>(reader: &'a mut R, buffer: &'a mut Vec<u8>) -> ReadToEnd<'a, R>
where
// R: AsyncRead：要求泛型类型 R 必须实现 AsyncRead 特征，意味着 R 类型的对象可以进行异步读取操作
// R: Unpin：要求 R 类型是可以被移动的，即不依赖于固定的内存位置。Unpin 用于支持异步任务的安全移动
// R: ?Sized：表示 R 可以是一个动态大小的类型，即类型的大小在编译时不需要确定R: AsyncRead + Unpin + ?Sized,
{ReadToEnd {reader, // 将传入的读取器引用存储在 ReadToEnd 结构体中buf: VecWithInitialized::new(buffer), // 将传入的 buffer 包装成 VecWithInitialized 类型的对象read: 0, // 初始化一个计数器，用于记录已经读取的字节数_pin: PhantomPinned, // 零大小的标记类型，用于表示 ReadToEnd 结构体可能包含固定位置的字段}
}

创建一个 ReadToEnd 结构体的实例，用于异步地将一个实现了 AsyncRead 特征的读取器中的所有数据读取到一个 Vec<u8> 缓冲区中

Future

为 ReadToEnd 结构体实现了 Future 特征，使得 ReadToEnd 类型的实例可以作为异步操作来使用，用于将数据从实现了 AsyncRead 特征的读取器中读取到缓冲区，直到文件末尾

impl<A> Future for ReadToEnd<'_, A>
whereA: AsyncRead + ?Sized + Unpin,
{// 返回的是一个 io::Result<usize>// 可能是一个成功的 usize 值（表示读取的字节数），也可能是一个 io::Error 错误type Output = io::Result<usize>;// 返回值是 Poll<Self::Output>// Poll 是一个枚举，有 Poll::Ready 和 Poll::Pending 两个变体，分别表示操作已完成和操作还未完成fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {// project 方法通常用于将 Pin<&mut Self> 解包为结构体的各个字段的固定引用// 投影let me = self.project();// me.reader 的类型是 & mut R，使用 Pin::new(*me.reader) 制作 Pin<&mut R> 类型read_to_end_internal(me.buf, Pin::new(*me.reader), me.read, cx)}
}// pub(super)：表示该函数在当前模块的父模块中是公共的
// V 必须实现 VecU8 特征
pub(super) fn read_to_end_internal<V: VecU8, R: AsyncRead + ?Sized>(buf: &mut VecWithInitialized<V>,mut reader: Pin<&mut R>,num_read: &mut usize,cx: &mut Context<'_>,
) -> Poll<io::Result<usize>> {loop {// 调用 poll_read_to_end 函数尝试读取数据// ready! 宏用于处理 Poll 枚举// 如果结果是 Poll::Pending，则当前任务会让出控制权// 如果结果是 Poll::Ready，则会继续执行后续代码// 这里为什么要用 Pin 的 as_mut?// impl<Ptr: DerefMut> Pin<Ptr> {//     pub fn as_mut(&mut self) -> Pin<&mut Ptr::Target> {// 			SAFETY: see documentation on this function// 			unsafe { Pin::new_unchecked(&mut *self.__pointer) }// 		}//     ...// }// as_mut 方法会返回一个新的 Pin 类型的值，具体为 Pin<&mut R>，不过它是基于当前 reader 的可变引用构建的// 返回的引用仍然保持了原有的固定性，即对象在内存中的位置不会被改变// 这对于处理 !Unpin 类型（不能被随意移动的类型）尤为重要，确保在后续操作中自引用的有效性// 返回的引用是可变的，这允许在不改变对象固定状态的前提下，对对象进行修改操作// 例如，可以调用对象的可变方法或者修改对象的可变字段let ret = ready!(poll_read_to_end(buf, reader.as_mut(), cx));match ret {Err(err) => return Poll::Ready(Err(err)),Ok(0) => return Poll::Ready(Ok(mem::replace(num_read, 0))),Ok(num) => {*num_read += num;}}}
}fn poll_read_to_end<V: VecU8, R: AsyncRead + ?Sized>(buf: &mut VecWithInitialized<V>,read: Pin<&mut R>,cx: &mut Context<'_>,
) -> Poll<io::Result<usize>> {// This uses an adaptive system to extend the vector when it fills. We want to// avoid paying to allocate and zero a huge chunk of memory if the reader only// has 4 bytes while still making large reads if the reader does have a ton// of data to return. Simply tacking on an extra DEFAULT_BUF_SIZE space every// time is 4,500 times (!) slower than this if the reader has a very small// amount of data to return. When the vector is full with its starting// capacity, we first try to read into a small buffer to see if we reached// an EOF. This only happens when the starting capacity is >= NUM_BYTES, since// we allocate at least NUM_BYTES each time. This avoids the unnecessary// allocation that we attempt before reading into the vector.// 这里采用了一种自适应系统（adaptive system），在向量（vector）填满时对其进行扩容。// 我们希望避免出现以下情况：// 当读取器（reader）只有 4 字节数据时，却要为分配并清零一大块内存而付出代价；// 同时，当读取器有大量数据要返回时，仍能进行大规模的读取操作。// 如果读取器要返回的数据量非常小，每次简单地额外增加默认缓冲区大小（DEFAULT_BUF_SIZE）的空间，// 其速度比这种自适应系统要慢 4500 倍// 当 vector 达到其初始容量时，我们会先尝试将数据读取到一个小缓冲区中，以此判断是否已到达文件末尾（EOF）// 这种做法仅在初始容量大于或等于 NUM_BYTES 时才会进行// 因为我们每次至少会分配 NUM_BYTES 的空间// 这样做避免了在将数据读取到 vector 之前进行不必要的内存分配const NUM_BYTES: usize = 32;let try_small_read = buf.try_small_read_first(NUM_BYTES);// Get a ReadBuf into the vector.let mut read_buf;let poll_result;// 如果 try_small_read 为 true，则先使用一个小缓冲区 small_buf 进行读取，将读取的数据存储到 small_read_buf 中// 然后将读取的数据复制到 read_buf 中// 如果 read_buf 空间不足，调用 buf.reserve(NUM_BYTES) 扩展缓冲区// 最后返回读取的字节数 to_write.len()// // 如果 try_small_read 为 false，则直接调用 buf.reserve(NUM_BYTES) 扩展缓冲区// 然后将数据直接读取到 read_buf 中，计算读取的字节数并返回let n = if try_small_read {// Read some bytes using a small read.let mut small_buf: [MaybeUninit<u8>; NUM_BYTES] = [MaybeUninit::uninit(); NUM_BYTES];let mut small_read_buf = ReadBuf::uninit(&mut small_buf);poll_result = read.poll_read(cx, &mut small_read_buf);let to_write = small_read_buf.filled();// Ensure we have enough space to fill our vector with what we read.read_buf = buf.get_read_buf();if to_write.len() > read_buf.remaining() {buf.reserve(NUM_BYTES);read_buf = buf.get_read_buf();}read_buf.put_slice(to_write);to_write.len()} else {// Ensure we have enough space for reading.buf.reserve(NUM_BYTES);read_buf = buf.get_read_buf();// Read data directly into vector.let filled_before = read_buf.filled().len();poll_result = read.poll_read(cx, &mut read_buf);// Compute the number of bytes read.read_buf.filled().len() - filled_before};// Update the length of the vector using the result of poll_read.let read_buf_parts = into_read_buf_parts(read_buf);buf.apply_read_buf(read_buf_parts);// 如果是 Poll::Pending，表示读取操作还未完成，当前任务让出控制权，返回 Poll::Pending// 如果是 Poll::Ready(Err(err))，表示读取过程中出现错误，返回 Poll::Ready(Err(err))// 如果是 Poll::Ready(Ok(()))，表示读取操作成功，返回 Poll::Ready(Ok(n))，其中 n 是读取的字节数match poll_result {Poll::Pending => {debug_assert_eq!(n, 0);Poll::Pending}Poll::Ready(Err(err)) => {debug_assert_eq!(n, 0);Poll::Ready(Err(err))}Poll::Ready(Ok(())) => Poll::Ready(Ok(n)),}
}

附录

ready!

// pub：表示这个宏是公共的，意味着它可以在定义它的 crate 外部被使用
// $e:expr：这是一个宏参数，$e 是参数名，expr 表示该参数是一个表达式
// 在使用这个宏时，需要传入一个表达式，该表达式的结果应该是 Poll 枚举类型
pub macro ready($e:expr) {match $e {$crate::task::Poll::Ready(t) => t,$crate::task::Poll::Pending => {return $crate::task::Poll::Pending;}}
}

Unpin 和 !Unpin

Unpin：Unpin 是一个自动实现的特征，类型默认实现 Unpin 特征意味着它可以安全地被移动。而没有实现 Unpin 特征的类型则是 !Unpin，即不能被随意移动。

PhantomPinned 是一个零大小的标记类型，它的主要作用是让结构体成为 !Unpin 类型，也就是不能被随意移动的类型

虽然从 ReadToEnd 结构体的定义来看，并没有明显的自引用，但在未来的实现中可能会引入自引用。添加 PhantomPinned 可以提前为可能的自引用情况做好准备。自引用结构体的一个重要特性是其内存位置不能改变，否则自引用会失效。通过使用 PhantomPinned 让结构体成为 !Unpin 类型，可以确保在异步操作过程中，结构体的内存位置不会被意外移动，从而保证自引用的有效性。

Pin 和 !Unpin

Pin 和 !Unpin 不能互相替换，在 Rust 异步编程中有着不同的概念和用途：

Pin

Pin 是一个类型构造器，定义在 std::pin::Pin。它用于固定一个值的内存位置，确保该值在内存中不会被移动。其核心作用是处理自引用结构体，因为自引用结构体依赖于内存位置的稳定性，一旦被移动，自引用就会失效：

use std::pin::Pin;struct SelfReferential {data: String,pointer: *const String,
}impl SelfReferential {fn new(data: String) -> Self {let mut this = Self {data,pointer: std::ptr::null(),};this.pointer = &this.data;this}
}fn main() {let mut value = SelfReferential::new("hello".to_string());let pinned: Pin<&mut SelfReferential> = Pin::new(&mut value);// 此时 pinned 中的 value 不能被移动
}

Unpin

Unpin 是一个自动实现的特征（auto trait）。如果一个类型实现了 Unpin 特征，意味着它可以安全地被移动；反之，!Unpin 表示该类型不能被随意移动。大多数简单类型（如 i32、String 等）都自动实现了 Unpin 特征，而包含自引用的类型通常是 !Unpin 的。

为什么不能互相替换

用途不同
Pin 是一个具体的类型，用于对值进行包装，以固定其内存位置。它提供了一种操作方式，让你可以在代码中明确地控制某个值是否可以被移动。
Unpin 和 !Unpin 是特征层面的概念，用于描述类型的移动性。它们决定了一个类型在异步操作中是否可以被安全地移动。

功能不同
Pin 关注的是对单个值的内存位置的固定，它可以将一个原本可以移动的 Unpin 类型的值包装起来，使其在特定的上下文中不能被移动；也可以处理 !Unpin 类型的值
Unpin 和 !Unpin 主要影响类型在异步编程中的使用方式。例如，Future 特征的 poll 方法在处理 !Unpin 类型的 Future 时，需要使用 Pin 来确保其内存位置固定。

为什么说 Future 特征的 poll 方法在处理 !Unpin 类型的 Future 时，需要使用 Pin 来确保其内存位置固定

（1）为了保证自引用的有效性、避免数据竞争以及满足异步编程的语义要求
!Unpin 类型通常是包含自引用的类型。自引用意味着类型内部的某个字段持有指向自身其他部分的引用。例如：

struct SelfReferential {data: String,reference: *const String,
}impl SelfReferential {fn new(data: String) -> Self {let mut this = Self { data, reference: std::ptr::null() };this.reference = &this.data;this}
}

Pin 类型可以固定一个值的内存位置，防止其被移动。当 Future 是 !Unpin 类型时，使用 Pin 来包装它，就可以确保在 poll 方法执行过程中，Future 的内存位置不会改变，从而保证自引用的有效性。例如：

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};struct SelfReferentialFuture {state: u32,reference: *const u32,
}impl SelfReferentialFuture {fn new(state: u32) -> Self {let mut this = Self { state, reference: std::ptr::null() };this.reference = &this.state;this}
}impl Future for SelfReferentialFuture {type Output = u32;fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {// 由于使用了 Pin，这里可以安全地使用自引用let state = unsafe { *self.get_unchecked_mut().reference };Poll::Ready(state)}
}

（2）避免数据竞争
在异步编程中，Future 可能会在多个任务调度中被暂停和恢复。如果 Future 是 !Unpin 类型且没有固定内存位置，在暂停和恢复的过程中，可能会出现数据竞争的问题。例如，一个 Future 可能会在 poll 方法中更新自身的状态，如果在更新过程中 Future 被移动，就可能导致数据不一致。
Pin 确保一致性
使用 Pin 固定 Future 的内存位置，可以保证在整个异步操作的生命周期内，Future 的状态是稳定的，避免了因移动而导致的数据竞争问题。这样，poll 方法可以安全地访问和修改 Future 的内部状态。

（3）满足异步编程的语义要求
Rust 的异步 trait 方法设计上考虑了 !Unpin 类型的 Future。为了确保异步操作的正确性和安全性，Future 特征的 poll 方法要求处理 !Unpin 类型的 Future 时使用 Pin。这是一种约定，保证了不同的异步实现之间的一致性和兼容性。

（4）遵循编程范式
遵循这种约定可以让开发者编写的异步代码更加规范和易于理解。当其他开发者看到 poll 方法接受 Pin<&mut Self> 类型的参数时，就知道这个 Future 可能是 !Unpin 类型，需要特别处理自引用的情况

综上所述，使用 Pin 来固定 !Unpin 类型的 Future 的内存位置，是为了保证自引用的有效性、避免数据竞争以及满足异步编程的语义要求，从而确保异步操作的正确性和安全性。

如何确定一个结构体是否是 Unpin 还是 !Unpin

use std::marker::PhantomPinned;// 定义一个泛型函数，要求 T 实现 Unpin 特征
fn check_unpin<T: Unpin>(_: T) {}// 一个简单的结构体，自动实现 Unpin 特征
struct SimpleStruct {data: u32,
}// 一个包含自引用的结构体，通常是 !Unpin 的
struct SelfReferential {data: String,pointer: *const String,_pin: PhantomPinned,
}impl SelfReferential {fn new(data: String) -> Self {let mut this = Self {data,pointer: std::ptr::null(),};this.pointer = &this.data;this}
}fn main() {let simple = SimpleStruct { data: 42 };// 编译通过，说明 SimpleStruct 实现了 Unpin 特征check_unpin(simple);let self_ref = SelfReferential::new("hello".to_string());// 编译报错，说明 SelfReferential 是 !Unpin 的check_unpin(self_ref);
}

self: Pin<&mut Self> 类型的参数，用到了 self.project()

self: Pin<&mut Self> 经常出现在 Future 等异步特征的 poll 方法中，用于处理那些不能被随意移动的类型（!Unpin 类型）。
而 self.project() 是一种用于解构 Pin<&mut Self> 的方式

self.project() 是基于 Pin 的投影（Projection）机制实现的。投影机制允许将一个 Pin<&mut Self> 解构为其内部字段的 Pin 包装的引用。

实现 Deref 和 DerefMut 的结构体

对于一个实现了 Deref 和 DerefMut 的结构体，Rust 提供了一种方便的方式来访问其内部字段：

use std::pin::Pin;struct MyStruct {field1: u32,field2: String,
}impl MyStruct {fn some_method(self: Pin<&mut Self>) {// 通过 Deref 和 DerefMut 访问字段let field1 = self.field1;let field2 = &mut self.field2;}
}

但这种方式只适用于 Unpin 类型的字段。对于 !Unpin 类型的字段，需要使用投影机制。

Pin 的投影机制

为了支持投影，结构体需要实现 Unpin 或者使用 #[pin_project] 等工具。
#[pin_project] 是 pin-project 库提供的一个宏，它可以自动为结构体生成投影方法

use pin_project::pin_project;#[pin_project]
struct MyAsyncStruct {#[pin]async_field: MyAsyncType,sync_field: u32,
}impl MyAsyncStruct {fn some_async_method(self: Pin<&mut Self>) {let this = self.project();// this.async_field 是 Pin<&mut MyAsyncType> 类型// this.sync_field 是 &mut u32 类型}
}

在这个例子中，#[pin_project] 宏会为 MyAsyncStruct 生成一个 project 方法。当调用 self.project() 时，它会返回一个包含各个字段投影的结构体，其中标记了 #[pin] 的字段会被包装成 Pin<&mut T> 类型，而其他字段则是普通的可变引用。

self.project() 的使用场景

处理 !Unpin 字段：当结构体中包含 !Unpin 类型的字段时，使用 self.project() 可以安全地访问和操作这些字段，同时保持它们的固定性。
异步操作：在异步编程中，Future 的 poll 方法通常接收 self: Pin<&mut Self> 类型的参数。通过 self.project()可以解构 self，并对内部的 !Unpin 字段进行异步操作，例如调用 poll 方法。

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use pin_project::pin_project;#[pin_project]
struct MyFuture {#[pin]inner_future: AnotherFuture,state: u32,
}impl Future for MyFuture {type Output = ();fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {let this = self.project();match this.inner_future.poll(cx) {Poll::Ready(_) => {// 处理完成状态this.state += 1;Poll::Ready(())}Poll::Pending => Poll::Pending,}}
}

在这个例子中，self.project() 用于解构 MyFuture，并对 inner_future 这个 !Unpin 类型的字段调用 poll 方法，同时可以安全地修改 state 字段。
self.project() 用于在处理 !Unpin 类型的结构体时，安全地解构 Pin<&mut Self> 并访问其内部字段，确保异步操作的正确性和安全性。

Pin的pin_project

pin_project! {pub struct ReadToEnd<'a, R: ?Sized> {reader: &'a mut R,buf: VecWithInitialized<&'a mut Vec<u8>>,read: usize,// Make this future `!Unpin` for compatibility with async trait methods.#[pin]_pin: PhantomPinned,}
}