Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Draft] add unsafe_destroy_range & split_region method #387

Draft
wants to merge 5 commits into
base: master
Choose a base branch
from
+283 −108
Draft

[Draft] add unsafe_destroy_range & split_region method #387

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
646d2b4
add split region method
pingyu Jan 4, 2023
ca55424
fix RUST_PROTOBUF error
pingyu Jan 4, 2023
3748f01
add unsafe_destroy_range
pingyu Jan 4, 2023
f69ef93
add backoff
pingyu Jan 4, 2023
5c2a2e8
handle region error
pingyu Jan 5, 2023
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
199 changes: 97 additions & 102 deletions src/request/plan.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ where
match backoff.next_delay_duration() {
Some(duration) => {
let region_error_resolved =
Self::handle_region_error(pd_client.clone(), e, region_store).await?;
handle_region_error(pd_client.clone(), e, region_store).await?;
// don't sleep if we have resolved the region error
if !region_error_resolved {
futures_timer::Delay::new(duration).await;
Expand All @@ -161,107 +161,6 @@ where
Ok(vec![Ok(resp)])
}
}

// Returns
// 1. Ok(true): error has been resolved, retry immediately
// 2. Ok(false): backoff, and then retry
// 3. Err(Error): can't be resolved, return the error to upper level
async fn handle_region_error(
pd_client: Arc<PdC>,
mut e: errorpb::Error,
region_store: RegionStore,
) -> Result<bool> {
let ver_id = region_store.region_with_leader.ver_id();
if e.has_not_leader() {
let not_leader = e.get_not_leader();
if not_leader.has_leader() {
match pd_client
.update_leader(
region_store.region_with_leader.ver_id(),
not_leader.get_leader().clone(),
)
.await
{
Ok(_) => Ok(true),
Err(e) => {
pd_client.invalidate_region_cache(ver_id).await;
Err(e)
}
}
} else {
// The peer doesn't know who is the current leader. Generally it's because
// the Raft group is in an election, but it's possible that the peer is
// isolated and removed from the Raft group. So it's necessary to reload
// the region from PD.
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
}
} else if e.has_store_not_match() {
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
} else if e.has_epoch_not_match() {
Self::on_region_epoch_not_match(
pd_client.clone(),
region_store,
e.take_epoch_not_match(),
)
.await
} else if e.has_stale_command() || e.has_region_not_found() {
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
} else if e.has_server_is_busy()
|| e.has_raft_entry_too_large()
|| e.has_max_timestamp_not_synced()
{
Err(Error::RegionError(Box::new(e)))
} else {
// TODO: pass the logger around
// info!("unknwon region error: {:?}", e);
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
}
}

// Returns
// 1. Ok(true): error has been resolved, retry immediately
// 2. Ok(false): backoff, and then retry
// 3. Err(Error): can't be resolved, return the error to upper level
async fn on_region_epoch_not_match(
pd_client: Arc<PdC>,
region_store: RegionStore,
error: EpochNotMatch,
) -> Result<bool> {
let ver_id = region_store.region_with_leader.ver_id();
if error.get_current_regions().is_empty() {
pd_client.invalidate_region_cache(ver_id).await;
return Ok(true);
}

for r in error.get_current_regions() {
if r.get_id() == region_store.region_with_leader.id() {
let returned_conf_ver = r.get_region_epoch().get_conf_ver();
let returned_version = r.get_region_epoch().get_version();
let current_conf_ver = region_store
.region_with_leader
.region
.get_region_epoch()
.get_conf_ver();
let current_version = region_store
.region_with_leader
.region
.get_region_epoch()
.get_version();

// Find whether the current region is ahead of TiKV's. If so, backoff.
if returned_conf_ver < current_conf_ver || returned_version < current_version {
return Ok(false);
}
}
}
// TODO: finer grained processing
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
}
}

impl<P: Plan, PdC: PdClient> Clone for RetryableMultiRegion<P, PdC> {
Expand Down Expand Up @@ -298,6 +197,102 @@ where
}
}

// Returns
// 1. Ok(true): error has been resolved, retry immediately
// 2. Ok(false): backoff, and then retry
// 3. Err(Error): can't be resolved, return the error to upper level
pub async fn handle_region_error<PdC: PdClient>(
pd_client: Arc<PdC>,
mut e: errorpb::Error,
region_store: RegionStore,
) -> Result<bool> {
let ver_id = region_store.region_with_leader.ver_id();
if e.has_not_leader() {
let not_leader = e.get_not_leader();
if not_leader.has_leader() {
match pd_client
.update_leader(
region_store.region_with_leader.ver_id(),
not_leader.get_leader().clone(),
)
.await
{
Ok(_) => Ok(true),
Err(e) => {
pd_client.invalidate_region_cache(ver_id).await;
Err(e)
}
}
} else {
// The peer doesn't know who is the current leader. Generally it's because
// the Raft group is in an election, but it's possible that the peer is
// isolated and removed from the Raft group. So it's necessary to reload
// the region from PD.
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
}
} else if e.has_store_not_match() {
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
} else if e.has_epoch_not_match() {
on_region_epoch_not_match(pd_client.clone(), region_store, e.take_epoch_not_match()).await
} else if e.has_stale_command() || e.has_region_not_found() {
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
} else if e.has_server_is_busy()
|| e.has_raft_entry_too_large()
|| e.has_max_timestamp_not_synced()
{
Err(Error::RegionError(Box::new(e)))
} else {
// TODO: pass the logger around
// info!("unknwon region error: {:?}", e);
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
}
}

// Returns
// 1. Ok(true): error has been resolved, retry immediately
// 2. Ok(false): backoff, and then retry
// 3. Err(Error): can't be resolved, return the error to upper level
async fn on_region_epoch_not_match<PdC: PdClient>(
pd_client: Arc<PdC>,
region_store: RegionStore,
error: EpochNotMatch,
) -> Result<bool> {
let ver_id = region_store.region_with_leader.ver_id();
if error.get_current_regions().is_empty() {
pd_client.invalidate_region_cache(ver_id).await;
return Ok(true);
}

for r in error.get_current_regions() {
if r.get_id() == region_store.region_with_leader.id() {
let returned_conf_ver = r.get_region_epoch().get_conf_ver();
let returned_version = r.get_region_epoch().get_version();
let current_conf_ver = region_store
.region_with_leader
.region
.get_region_epoch()
.get_conf_ver();
let current_version = region_store
.region_with_leader
.region
.get_region_epoch()
.get_version();

// Find whether the current region is ahead of TiKV's. If so, backoff.
if returned_conf_ver < current_conf_ver || returned_version < current_version {
return Ok(false);
}
}
}
// TODO: finer grained processing
pd_client.invalidate_region_cache(ver_id).await;
Ok(false)
}

/// A technique for merging responses into a single result (with type `Out`).
pub trait Merge<In>: Sized + Clone + Send + Sync + 'static {
type Out: Send;
Expand Down
138 changes: 133 additions & 5 deletions src/transaction/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,16 +5,23 @@ use crate::{
backoff::DEFAULT_REGION_BACKOFF,
config::Config,
pd::{PdClient, PdRpcClient},
request::{plan::CleanupLocksResult, Plan},
request::{
plan::{handle_region_error, CleanupLocksResult},
Plan,
},
store::RegionStore,
timestamp::TimestampExt,
transaction::{
lock::ResolveLocksOptions, ResolveLocksContext, Snapshot, Transaction, TransactionOptions,
lock::ResolveLocksOptions, requests, requests::new_unsafe_destroy_range_request,
ResolveLocksContext, Snapshot, Transaction, TransactionOptions,
},
Backoff, Result,
Backoff, BoundRange, Result,
};
use futures::{future::try_join_all, StreamExt};
use slog::{Drain, Logger};
use std::{mem, sync::Arc};
use tikv_client_proto::pdpb::Timestamp;
use std::{collections::HashMap, mem, sync::Arc};
use tikv_client_common::Error;
use tikv_client_proto::{metapb::Region, pdpb::Timestamp};

// FIXME: cargo-culted value
const SCAN_LOCK_BATCH_SIZE: u32 = 1024;
Expand Down Expand Up @@ -301,4 +308,125 @@ impl Client {
let logger = self.logger.new(o!("child" => 1));
Transaction::new(timestamp, self.pd.clone(), options, logger)
}

// FIXME: make the loop a plan.
pub async fn split_region_with_retry(
&self,
#[allow(clippy::ptr_arg)] key: &Vec<u8>,
split_keys: Vec<Vec<u8>>,
is_raw_kv: bool,
) -> Result<Vec<Region>> {
debug!(self.logger, "invoking split region with retry");
let mut backoff = DEFAULT_REGION_BACKOFF;
let mut i = 0;
'retry: loop {
i += 1;
debug!(self.logger, "split region: attempt {}", i);
let store = self.pd.clone().store_for_key(key.into()).await?;
let request = requests::new_split_region_request(split_keys.clone(), is_raw_kv);
let plan = crate::request::PlanBuilder::new(self.pd.clone(), request)
.single_region_with_store(store.clone())
.await?
.extract_error()
.plan();
match plan.execute().await {
Ok(mut resp) => return Ok(resp.take_regions().into()),
Err(Error::ExtractedErrors(mut errors)) => match errors.pop() {
Some(Error::RegionError(e)) => match backoff.next_delay_duration() {
Some(duration) => {
let region_error_resolved =
handle_region_error(self.pd.clone(), *e, store).await?;
if !region_error_resolved {
futures_timer::Delay::new(duration).await;
}
continue 'retry;
}
None => return Err(Error::RegionError(e)),
},
Some(e) => return Err(e),
None => unreachable!(),
},
Err(e) => return Err(e),
}
}
}

// FIXME: make list stores and retry a plan.
pub async fn unsafe_destroy_range(&self, start_key: Vec<u8>, end_key: Vec<u8>) -> Result<()> {
debug!(self.logger, "invoking unsafe destroy range");
let backoff = DEFAULT_REGION_BACKOFF;
let stores = self
.list_stores_for_unsafe_destroy(start_key.clone(), end_key.clone())
.await?;
let mut handles = Vec::with_capacity(stores.len());
for store in stores.into_values() {
let logger = self.logger.clone();
let start_key = start_key.clone();
let end_key = end_key.clone();
let pd = self.pd.clone();
let mut backoff = backoff.clone();
let task = async move {
let mut i = 0;
'retry: loop {
i += 1;
debug!(logger, "unsafe destroy range: attempt {}", i);
let request =
new_unsafe_destroy_range_request(start_key.clone(), end_key.clone());
let plan = crate::request::PlanBuilder::new(pd.clone(), request)
.single_region_with_store(store.clone())
.await?
.extract_error()
.plan();
match plan.execute().await {
Ok(_) => return Ok(()),
Err(e) => {
warn!(logger, "unsafe destroy range error: {:?}", e);
match backoff.next_delay_duration() {
Some(duration) => {
futures_timer::Delay::new(duration).await;
continue 'retry;
}
None => return Err(e),
}
}
}
}
};
handles.push(tokio::spawn(task));
}

let results = try_join_all(handles).await?;
match results.into_iter().find(|x| x.is_err()) {
Some(r) => r,
None => Ok(()),
}
}

async fn list_stores_for_unsafe_destroy(
&self,
start_key: Vec<u8>,
end_key: Vec<u8>,
) -> Result<HashMap<u64, RegionStore>> {
let mut stores = HashMap::new();
let bnd_range = BoundRange::from((start_key, end_key));
self.pd
.clone()
.stores_for_range(bnd_range)
.map(|store| -> Result<()> {
let store = store?;
let store_id = store
.region_with_leader
.leader
.as_ref()
.unwrap()
.get_store_id();
stores.entry(store_id).or_insert(store);
Ok(())
})
.collect::<Vec<_>>()
.await
.into_iter()
.collect::<Result<_>>()?;
Ok(stores)
}
}
1 change: 0 additions & 1 deletion src/transaction/lock.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,6 @@ async fn resolve_lock_with_retry(
let store = pd_client.clone().store_for_key(key.into()).await?;
let ver_id = store.region_with_leader.ver_id();
let request = requests::new_resolve_lock_request(start_version, commit_version);
// The only place where single-region is used
let plan = crate::request::PlanBuilder::new(pd_client.clone(), request)
.single_region_with_store(store)
.await?
Expand Down
Loading