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

Jump to bottom

pd: speed up reconnecting to a pd leader #2207

Merged
merged 5 commits into from Aug 29, 2017
Merged

pd: speed up reconnecting to a pd leader #2207

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
914ebd6
pd: add connection timeout
overvenus Aug 22, 2017
d163587
pd: add timeout to all requests
overvenus Aug 22, 2017
97a2242
pd: try to connect to other members, then the previous leader.
overvenus Aug 24, 2017
6f4a227
address comments
overvenus Aug 29, 2017
1bb9f0f
Merge branch 'master' into ov/pd-conn-timeout
overvenus Aug 29, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
52 changes: 39 additions & 13 deletions src/pd/client.rs
View file
Open in desktop
Expand Up @@ -13,17 +13,18 @@

use std::fmt;
use std::sync::{Arc, RwLock};
use std::time::Duration;

use protobuf::RepeatedField;
use futures::{future, Future, Sink, Stream};
use futures::sync::mpsc::{self, UnboundedSender};
use grpc::{EnvBuilder, WriteFlags};
use grpc::{CallOption, EnvBuilder, WriteFlags};
use kvproto::metapb;
use kvproto::pdpb::{self, Member};

use util::{Either, HandyRwLock};
use pd::PdFuture;
use super::{Error, PdClient, RegionStat, Result};
use super::{Error, PdClient, RegionStat, Result, REQUEST_TIMEOUT};
use super::util::{check_resp_header, sync_request, validate_endpoints, Inner, LeaderClient};

const CQ_COUNT: usize = 1;
Expand Down Expand Up @@ -86,7 +87,10 @@ impl PdClient for RpcClient {
let resp = try!(sync_request(
&self.leader_client,
LEADER_CHANGE_RETRY,
|client| client.bootstrap(req.clone())
|client| {
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
client.bootstrap_opt(req.clone(), option)
}
));
try!(check_resp_header(resp.get_header()));
Ok(())
Expand All @@ -99,7 +103,10 @@ impl PdClient for RpcClient {
let resp = try!(sync_request(
&self.leader_client,
LEADER_CHANGE_RETRY,
|client| client.is_bootstrapped(req.clone())
|client| {
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
client.is_bootstrapped_opt(req.clone(), option)
}
));
try!(check_resp_header(resp.get_header()));

Expand All @@ -113,7 +120,10 @@ impl PdClient for RpcClient {
let resp = try!(sync_request(
&self.leader_client,
LEADER_CHANGE_RETRY,
|client| client.alloc_id(req.clone())
|client| {
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
client.alloc_id_opt(req.clone(), option)
}
));
try!(check_resp_header(resp.get_header()));

Expand All @@ -128,7 +138,10 @@ impl PdClient for RpcClient {
let resp = try!(sync_request(
&self.leader_client,
LEADER_CHANGE_RETRY,
|client| client.put_store(req.clone())
|client| {
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
client.put_store_opt(req.clone(), option)
}
));
try!(check_resp_header(resp.get_header()));

Expand All @@ -143,7 +156,10 @@ impl PdClient for RpcClient {
let mut resp = try!(sync_request(
&self.leader_client,
LEADER_CHANGE_RETRY,
|client| client.get_store(req.clone())
|client| {
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
client.get_store_opt(req.clone(), option)
}
));
try!(check_resp_header(resp.get_header()));

Expand All @@ -157,7 +173,10 @@ impl PdClient for RpcClient {
let mut resp = try!(sync_request(
&self.leader_client,
LEADER_CHANGE_RETRY,
|client| client.get_cluster_config(req.clone())
|client| {
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
client.get_cluster_config_opt(req.clone(), option)
}
));
try!(check_resp_header(resp.get_header()));

Expand All @@ -172,7 +191,10 @@ impl PdClient for RpcClient {
let mut resp = try!(sync_request(
&self.leader_client,
LEADER_CHANGE_RETRY,
|client| client.get_region(req.clone())
|client| {
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
client.get_region_opt(req.clone(), option)
}
));
try!(check_resp_header(resp.get_header()));

Expand All @@ -185,7 +207,8 @@ impl PdClient for RpcClient {
req.set_region_id(region_id);

let executor = |client: &RwLock<Inner>, req: pdpb::GetRegionByIDRequest| {
let handler = client.rl().client.get_region_by_id_async(req);
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
let handler = client.rl().client.get_region_by_id_async_opt(req, option);
handler
.map_err(Error::Grpc)
.and_then(|mut resp| {
Expand Down Expand Up @@ -268,7 +291,8 @@ impl PdClient for RpcClient {
req.set_region(region);

let executor = |client: &RwLock<Inner>, req: pdpb::AskSplitRequest| {
let handler = client.rl().client.ask_split_async(req);
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
let handler = client.rl().client.ask_split_async_opt(req, option);
handler
.map_err(Error::Grpc)
.and_then(|resp| {
Expand All @@ -289,7 +313,8 @@ impl PdClient for RpcClient {
req.set_stats(stats);

let executor = |client: &RwLock<Inner>, req: pdpb::StoreHeartbeatRequest| {
let handler = client.rl().client.store_heartbeat_async(req);
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
let handler = client.rl().client.store_heartbeat_async_opt(req, option);
handler
.map_err(Error::Grpc)
.and_then(|resp| {
Expand All @@ -311,7 +336,8 @@ impl PdClient for RpcClient {
req.set_right(right);

let executor = |client: &RwLock<Inner>, req: pdpb::ReportSplitRequest| {
let handler = client.rl().client.report_split_async(req);
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
let handler = client.rl().client.report_split_async_opt(req, option);
handler
.map_err(Error::Grpc)
.and_then(|resp| {
Expand Down
2 changes: 2 additions & 0 deletions src/pd/mod.rs
View file
Open in desktop
Expand Up @@ -137,3 +137,5 @@ pub trait PdClient: Send + Sync {
// Report pd the split region.
fn report_split(&self, left: metapb::Region, right: metapb::Region) -> PdFuture<()>;
}

const REQUEST_TIMEOUT: u64 = 2; // 2s
27 changes: 14 additions & 13 deletions src/pd/util.rs
View file
Open in desktop
Expand Up @@ -22,18 +22,17 @@ use futures::{task, Async, BoxFuture, Future, Poll, Stream};
use futures::task::Task;
use futures::future::{loop_fn, ok, Loop};
use futures::sync::mpsc::UnboundedSender;
use grpc::{ChannelBuilder, ClientDuplexReceiver, ClientDuplexSender, Environment,
use grpc::{CallOption, ChannelBuilder, ClientDuplexReceiver, ClientDuplexSender, Environment,
Result as GrpcResult};
use tokio_timer::Timer;
use rand::{self, Rng};
use kvproto::pdpb::{ErrorType, GetMembersRequest, GetMembersResponse, Member,
RegionHeartbeatRequest, RegionHeartbeatResponse, ResponseHeader};
use kvproto::pdpb_grpc::PdClient;
use prometheus::HistogramTimer;

use util::{Either, HandyRwLock};
use pd::{Error, PdFuture, Result};
use pd::metrics::PD_SEND_MSG_HISTOGRAM;
use super::{Error, PdFuture, Result, REQUEST_TIMEOUT};
use super::metrics::PD_SEND_MSG_HISTOGRAM;

pub struct Inner {
env: Arc<Environment>,
Expand Down Expand Up @@ -199,7 +198,7 @@ pub struct Request<Req, Resp, F> {
timer: Option<HistogramTimer>,
}

const MAX_REQUEST_COUNT: usize = 5;
const MAX_REQUEST_COUNT: usize = 3;

impl<Req, Resp, F> Request<Req, Resp, F>
where
Expand Down Expand Up @@ -378,7 +377,8 @@ fn connect(env: Arc<Environment>, addr: &str) -> Result<(PdClient, GetMembersRes
let addr = addr.trim_left_matches("http://");
let channel = ChannelBuilder::new(env).connect(addr);
let client = PdClient::new(channel);
match client.get_members(GetMembersRequest::new()) {
let option = CallOption::default().timeout(Duration::from_secs(REQUEST_TIMEOUT));
match client.get_members_opt(GetMembersRequest::new(), option) {
Ok(resp) => Ok((client, resp)),
Err(e) => Err(Error::Grpc(e)),
}
Expand All @@ -388,16 +388,17 @@ pub fn try_connect_leader(
env: Arc<Environment>,
previous: &GetMembersResponse,
) -> Result<(PdClient, GetMembersResponse)> {
// Try to connect other members.
// Randomize endpoints.
let previous_leader = previous.get_leader();
let members = previous.get_members();
let mut indexes: Vec<usize> = (0..members.len()).collect();
rand::thread_rng().shuffle(&mut indexes);

let cluster_id = previous.get_header().get_cluster_id();
let mut resp = None;
'outer: for i in indexes {
for ep in members[i].get_client_urls() {
// Try to connect to other members, then the previous leader.
'outer: for m in members
.into_iter()
.filter(|m| *m != previous_leader)
.chain(&[previous_leader.clone()])
{
for ep in m.get_client_urls() {
match connect(env.clone(), ep.as_str()) {
Ok((_, r)) => {
let new_cluster_id = r.get_header().get_cluster_id();
Expand Down