Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/request/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ mod shard;
#[async_trait]
pub trait KvRequest: Request + Sized + Clone + Sync + Send + 'static {
/// The expected response to the request.
type Response: HasKeyErrors + HasLocks + Clone + Send + 'static;
type Response: HasKeyErrors + HasLocks + Clone + Send + std::fmt::Debug + 'static;

// TODO: fn encode_request()
// TODO: fn decode_response()
Expand Down Expand Up @@ -114,7 +114,7 @@ mod test {

#[tokio::test]
async fn test_region_retry() {
#[derive(Clone)]
#[derive(Clone, Debug)]
struct MockRpcResponse;

impl HasKeyErrors for MockRpcResponse {
Expand Down
20 changes: 18 additions & 2 deletions src/request/plan.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.

use std::marker::PhantomData;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use async_recursion::async_recursion;
Expand Down Expand Up @@ -38,6 +39,12 @@ use crate::Error;
use crate::Result;
use crate::Timestamp;

static DISPATCH_REQUEST_ID: AtomicU64 = AtomicU64::new(1);

fn next_dispatch_request_id() -> u64 {
DISPATCH_REQUEST_ID.fetch_add(1, Ordering::Relaxed)
}

/// A plan for how to execute a request. A user builds up a plan with various
/// options, then exectutes it.
#[async_trait]
Expand All @@ -61,7 +68,11 @@ impl<Req: KvRequest> Plan for Dispatch<Req> {
type Result = Req::Response;

async fn execute(&self) -> Result<Self::Result> {
let request_id = next_dispatch_request_id();
let stats = tikv_stats(self.request.label());
if self.request.label() == "kv_prewrite" || self.request.label() == "kv_commit" {
info!("req_id={} req {}", request_id, self.request.to_str())
}
let result = self
.kv_client
.as_ref()
Expand All @@ -70,8 +81,13 @@ impl<Req: KvRequest> Plan for Dispatch<Req> {
.await;
let result = stats.done(result);
result.map(|r| {
*r.downcast()
.expect("Downcast failed: request and response type mismatch")
let resp = *r
.downcast()
.expect("Downcast failed: request and response type mismatch");
if self.request.label() == "kv_prewrite" || self.request.label() == "kv_commit" {
info!("req_id={} resp {:?}", request_id, resp);
}
resp
})
}
}
Expand Down
7 changes: 7 additions & 0 deletions src/store/request.rs
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,9 @@ pub trait Request: Any + Sync + Send + 'static {
/// Should always use `set_context` other than modify the `self.context` directly.
fn set_context(&mut self, context: kvrpcpb::Context);
fn set_api_version(&mut self, api_version: kvrpcpb::ApiVersion);
fn to_str(&self) -> String {
"".to_string()
}
}

macro_rules! impl_request {
Expand Down Expand Up @@ -68,6 +71,10 @@ macro_rules! impl_request {
let context = self.context.get_or_insert(kvrpcpb::Context::default());
context.api_version = api_version.into();
}

fn to_str(&self) -> String {
format!("{:?}", self)
}
}
};
}
Expand Down