TiKV 源码阅读三部曲（二）读流程

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

#[test_case(test_raftstore::must_new_cluster_and_kv_client)]
#[test_case(test_raftstore_v2::must_new_cluster_and_kv_client)]
fn test_rawkv()

let (_cluster, client, ctx) = new_cluster();
let v0 = b"v0".to_vec();
let v1 = b"v1".to_vec();
let (k, v) = (b"key".to_vec(), b"v2".to_vec());
    
// Raw get
let mut get_req = RawGetRequest::default();
get_req.set_context(ctx.clone());
get_req.key = k.clone();
let get_resp = client.raw_get(&get_req).unwrap();
assert_eq!(get_resp.value, v0);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

代码地址：
 https://github.com/pingcap/kvproto/blob/master/proto/tikvpb.proto
 https://tikv.github.io/doc/kvproto/tikvpb/trait.Tikv.html#tymethod.raw_get
 
service Tikv {
   // Raw commands; no transaction support.
   rpc RawGet(kvrpcpb.RawGetRequest) returns (kvrpcpb.RawGetResponse) {}
}
//语法回顾：
定义服务(Service)  如果想要将消息类型用在RPC(远程方法调用)系统中，可以在.proto文件中定义一个RPC服务接口

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// Service handles the RPC messages for the `Tikv` service.

// 这是一个文档注释，说明 Service 结构体用于处理 `Tikv` 服务的 RPC 消息。
pub struct Service<E: Engine, L: LockManager, F: KvFormat> {
    // 定义 Service 结构体，它包含泛型参数 E, L, F，分别代表不同的组件或功能。
    cluster_id: u64,
    // Service 结构体的字段，用于存储集群 ID。
    store_id: u64,
    // For handling KV requests.
    storage: Storage<E, L, F>,
    // Service 结构体的字段，用于处理键值（KV）请求。
    
}
 // 
impl<E: Engine, L: LockManager, F: KvFormat> Tikv for Service<E, L, F> {
    handle_request!(kv_get, future_get, GetRequest, GetResponse, has_time_detail);
    handle_request!(raw_get, future_raw_get, RawGetRequest, RawGetResponse);

1
2
3
4
5
6
7

// 定义一个宏 `handle_request`，它接受一组模式匹配的参数。
macro_rules! handle_request {
    // 第一个模式匹配，当提供四个参数时使用。
    ($fn_name: ident, $future_name: ident, $req_ty: ident, $resp_ty: ident) => {

    };
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// 定义一个泛型异步函数 future_raw_get，它接受三个泛型参数 E, L, F，分别约束为 Engine, LockManager, KvFormat。
fn future_raw_get<E: Engine, L: LockManager, F: KvFormat>(
    // 函数接受一个对 Storage<E, L, F> 类型的引用和一个 RawGetRequest 类型的可变引用。
    storage: &Storage<E, L, F>,
    mut req: RawGetRequest,
) -> impl Future<Output = ServerResult<RawGetResponse>> {
    // 调用 storage 的 raw_get 方法，传入 req 的上下文、列族和键，并获取返回值。
    let v = storage.raw_get(req.take_context(), req.take_cf(), req.take_key());

    // 定义一个异步闭包，它将异步执行。
    // https://rust-lang.github.io/async-book/03_async_await/01_chapter.html
    async move {
        // 等待 v 的 Future 完成并获取结果。
        let v = v.await;
        // 创建一个 RawGetResponse 的默认实例。
        let mut resp = RawGetResponse::default();
        // 尝试从 v 中提取区域错误，如果有，则设置到响应中。
        if let Some(err) = extract_region_error(&v) {
            resp.set_region_error(err);
        } else {
            // 根据 v 的结果进行不同的处理。
            match v {
                // 如果获取到值，将其设置到响应的值字段中。
                //https://doc.rust-lang.org/std/result/
                Ok(Some(val)) => resp.set_value(val),
                // 如果没有找到值，设置响应的 not_found 字段为 true。
                Ok(None) => resp.set_not_found(true),
                // 如果发生错误，将错误信息格式化后设置到响应的 error 字段中。
                Err(e) => resp.set_error(format!("{}", e)),
            }
        }
        // 返回构建好的响应。
        Ok(resp)
    }
}

1
2
3
4
5
6
7

/// Get the value of a raw key.
    pub fn raw_get(
        &self,
        ctx: Context,
        cf: String,
        key: Vec<u8>,
    ) -> impl Future<Output = Result<Option<Vec<u8>>>> s

1
2
3
4
5
6
7

+TiKV::kv_get (grpc-poll-thread)
 +future_get
  +Storage::get
   +Storage::snapshot (readpool-thread)
   +SnapshotStore::get
     +PointGetterBuilder::build
     +PointGetter::get

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

/// [`Storage`](Storage) implements transactional KV APIs and raw KV APIs on a
/// given [`Engine`]. An [`Engine`] provides low level KV functionality.

/// [`Engine`] has multiple implementations. When a TiKV server is running, a
/// [`RaftKv`](crate::server::raftkv::RaftKv) will be the underlying [`Engine`]
/// of [`Storage`]. The other two types of engines are for test purpose.
///
/// [`Storage`] is reference counted and cloning [`Storage`] will just increase
/// the reference counter. Storage resources (i.e. threads, engine) will be
/// released when all references are dropped.
///
/// Notice that read and write methods may not be performed over full data in
/// most cases, i.e. when underlying engine is
/// [`RaftKv`](crate::server::raftkv::RaftKv), which limits data access in the
/// range of a single region according to specified `ctx` parameter. However,
/// [`unsafe_destroy_range`](crate::server::gc_worker::GcTask::
/// UnsafeDestroyRange) is the only exception. It's always performed on the
/// whole TiKV.
///
/// Operations of [`Storage`](Storage) can be divided into two types: MVCC
/// operations and raw operations. MVCC operations uses MVCC keys, which usually
/// consist of several physical keys in different CFs. In default CF and write
/// CF, the key will be memcomparable-encoded and append the timestamp to it, so
/// that multiple versions can be saved at the same time. Raw operations use raw
/// keys, which are saved directly to the engine without memcomparable- encoding
/// and appending timestamp.

pub struct Storage<E: Engine, L: LockManager, F: KvFormat> {
    // TODO: Too many Arcs, would be slow when clone.
    engine: E,
    sched: TxnScheduler<E, L>,
    /// The thread pool used to run most read operations.
    //读取线程池，所有只读 KV 请求，包括事务的和非事务的，如 raw get、txn kv get 等最终都会在这个线程池内执行
    read_pool: ReadPoolHandle,
}

1
2
3
4
5
6
7

/// Get the value of a raw key.
pub fn raw_get(
    &self,
    ctx: Context,
    cf: String,
    key: Vec<u8>,
) -> impl Future<Output = Result<Option<Vec<u8>>>> {

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

/// 获取原始键的值。
/// 
/// 该方法接收一个上下文、列族cf和键key，返回一个Future，
/// 输出是一个包含结果的Option<Vec<u8>>类型。
pub fn raw_get(
    &self,
    ctx: Context,
    cf: String,
    key: Vec<u8>,
) -> impl Future<Output = Result<Option<Vec<u8>>>> {
 
    // 创建并启动带忙碌检查的读取池
    self.read_pool_spawn_with_busy_check(
        busy_threshold,
        async move {
            // 增加特定命令的计数
            KV_COMMAND_COUNTER_VEC_STATIC.get(CMD).inc();
            SCHED_COMMANDS_PRI_COUNTER_VEC_STATIC
                .get(priority_tag)
                .inc();

            // 检查API版本是否匹配
            Self::check_api_version(api_version, ctx.api_version, CMD, [&key])?;

            // 记录命令持续时间
            let command_duration = Instant::now();

            // 创建快照上下文
            let snap_ctx = SnapContext {
                pb_ctx: &ctx,
                ..Default::default()
            };

            // 获取快照
            let snapshot =
                Self::with_tls_engine(|engine| Self::snapshot(engine, snap_ctx)).await?;

            // 获取桶信息
            let buckets = snapshot.ext().get_buckets();

            // 创建原始存储
            let store = RawStore::new(snapshot, api_version);

            // 获取列族信息
            let cf = Self::rawkv_cf(&cf, api_version)?;

            // 内部命令执行块
            {
                // 记录起始时间
                let begin_instant = Instant::now();
                let mut stats = Statistics::default();

                // 编码原始键
                let key = F::encode_raw_key_owned(key, None);

                // 获取键值
                let r = store
                    .raw_get_key_value(cf, &key, &mut stats)
                    .map_err(Error::from);

}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

 // 获取快照
let snapshot = Self::with_tls_engine(|engine| Self::snapshot(engine, snap_ctx)).await?;
await? 用于等待异步操作完成并检查是否有错误发生。如果有错误，它将返回该错误，并使用.await?引发适当的错误处理

pub type Result<T> = std::result::Result<T, Error>;
pub type Callback<T> = Box<dyn FnOnce(Result<T>) + Send>;

#[inline]
fn with_tls_engine<R>(f: impl FnOnce(&mut E) -> R) -> R {
    // Safety: the read pools ensure that a TLS engine exists.
    unsafe { with_tls_engine(f) }
}

/// Get a snapshot of `engine`.
    fn snapshot(
        engine: &mut E,
        ctx: SnapContext<'_>,
    ) -> impl std::future::Future<Output = Result<E::Snap>> {
        kv::snapshot(engine, ctx)
            .map_err(txn::Error::from)
            .map_err(Error::from)
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// 定义一个泛型函数 read_pool_spawn_with_busy_check，它接受以下参数：
fn read_pool_spawn_with_busy_check<Fut, T>(
    &self, // &self 表示这是一个方法，它借用了当前实例。
    busy_threshold: Duration, // busy_threshold 参数，表示繁忙阈值。
    future: Fut, // future 参数，是一个异步任务。
    priority: CommandPri, // priority 参数，表示任务的优先级。
    task_id: u64, // task_id 参数，表示任务的唯一标识符。
    metadata: TaskMetadata<'_>, // metadata 参数，包含任务的元数据。
    resource_limiter: Option<Arc<ResourceLimiter>>, // resource_limiter 参数，表示资源限制器的可选引用。
) -> impl Future<Output = Result<T>> // 返回一个实现了 Future trait 的类型，其输出是 Result<T> 类型。
where
    Fut: Future<Output = Result<T>> + Send + 'static, // Fut 必须是可发送的（Send）并且 'static 生命周期，且其输出必须是 Result<T>。
    T: Send + 'static, // T 必须是可发送的（Send）并且 'static 生命周期。
{
    // 检查繁忙阈值，如果返回 Err，则构造一个错误并立即返回。
    if let Err(busy_err) = self.read_pool.check_busy_threshold(busy_threshold) {
        let mut err = kvproto::errorpb::Error::default();
        err.set_server_is_busy(busy_err); // 设置错误为服务器繁忙。
        return Either::Left(future::err(Error::from(ErrorInner::Kv(err.into()))));
    }

    // 如果检查通过，则将 future 任务提交给读取池。
    Either::Right(
        self.read_pool
            .spawn_handle(
                future, // 提交的异步任务。
                priority, // 任务的优先级。
                task_id, // 任务的唯一标识符。
                metadata, // 任务的元数据。
                resource_limiter, // 资源限制器的可选引用。
            )
            .map_err(|_| Error::from(ErrorInner::SchedTooBusy)) // 将 spawn_handle 可能的错误转换为统一的错误类型。
            .and_then(|res| future::ready(res)), // 使用 and_then 来处理 spawn_handle 的结果。
    )
}

1
2
3
4
5
6
7

+TiKV::kv_get (grpc-poll-thread)
 +future_get
  +Storage::get
   +Storage::snapshot (readpool-thread)
   +SnapshotStore::get
     +PointGetterBuilder::build
     +PointGetter::get