Improve readability of utils

Also removes two tests which afaict are testing the std lib, not this crate

Signed-off-by: Nick Cameron <nrc@ncameron.org>

Author: nrc

src/rpc/client.rs

@@ -24,7 +24,6 @@ use crate::{
         pd::{PdClient, PdTimestamp, Region, RegionId, RegionVerId, Store, StoreId},
         security::SecurityManager,
         tikv::KvClient,
-        util::HandyRwLock,
     },
     Config, Error, Key, KvPair, Result, Value,
 };
@@ -45,7 +44,7 @@ impl RpcClientInner {
         let env = Arc::new(
             EnvBuilder::new()
                 .cq_count(CQ_COUNT)
-                .name_prefix(thd_name!(CLIENT_PREFIX))
+                .name_prefix(thread_name!(CLIENT_PREFIX))
                 .build(),
         );
         let security_mgr = Arc::new(
@@ -112,7 +111,7 @@ impl RpcClientInner {
     }
 
     fn kv_client(&self, context: RegionContext) -> Result<(RegionContext, Arc<KvClient>)> {
-        if let Some(conn) = self.tikv.rl().get(context.address()) {
+        if let Some(conn) = self.tikv.read().unwrap().get(context.address()) {
             return Ok((context, Arc::clone(conn)));
         };
         info!("connect to tikv endpoint: {:?}", context.address());
@@ -125,7 +124,8 @@ impl RpcClientInner {
         )
         .map(Arc::new)
         .map(|c| {
-            tikv.wl()
+            tikv.write()
+                .unwrap()
                 .insert(context.address().to_owned(), Arc::clone(&c));
             (context, c)
         })

src/rpc/pd/client.rs

@@ -21,7 +21,6 @@ use crate::{
             PdTimestamp, Region, RegionId, Store, StoreId,
         },
         security::SecurityManager,
-        util::HandyRwLock,
     },
     Error, ErrorKind, Result,
 };
@@ -64,7 +63,7 @@ impl PdClient {
         timeout: Duration,
     ) -> Result<PdClient> {
         let leader = LeaderClient::connect(env, endpoints, security_mgr, timeout)?;
-        let cluster_id = leader.rl().cluster_id();
+        let cluster_id = leader.read().unwrap().cluster_id();
 
         Ok(PdClient {
             cluster_id,
@@ -74,7 +73,7 @@ impl PdClient {
     }
 
     fn get_leader(&self) -> pdpb::Member {
-        self.leader.rl().members.get_leader().clone()
+        self.leader.read().unwrap().members.get_leader().clone()
     }
 
     fn get_region_and_leader(
@@ -148,7 +147,7 @@ impl PdClient {
         let mut executor = context.executor();
         let wrapper = move |cli: &RwLock<LeaderClient>| {
             let option = CallOption::default().timeout(timeout);
-            let cli = &cli.rl().client;
+            let cli = &cli.read().unwrap().client;
             executor(cli, option).unwrap().map(|r| match r {
                 Err(e) => Err(ErrorKind::Grpc(e))?,
                 Ok(r) => {
@@ -209,7 +208,7 @@ impl PdClient {
     }
 
     pub fn get_ts(&self) -> impl Future<Output = Result<PdTimestamp>> {
-        self.leader.wl().get_ts()
+        self.leader.write().unwrap().get_ts()
     }
 }
 

src/rpc/pd/leader.rs

@@ -31,7 +31,6 @@ use crate::{
             PdTimestamp,
         },
         security::SecurityManager,
-        util::HandyRwLock,
     },
     Error, Result,
 };
@@ -131,10 +130,10 @@ impl PdReactor {
 
     fn init(client: &Arc<RwLock<LeaderClient>>, handle: &OtherHandle) {
         let client = Arc::clone(client);
-        let (tx, rx) = client.wl().client.tso().unwrap();
+        let (tx, rx) = client.write().unwrap().client.tso().unwrap();
         let tx = Compat01As03Sink::new(tx);
         let rx = Compat01As03::new(rx);
-        let tso_rx = client.wl().reactor.tso_rx.take().unwrap(); // Receiver<TsoRequest>: Stream
+        let tso_rx = client.write().unwrap().reactor.tso_rx.take().unwrap(); // Receiver<TsoRequest>: Stream
 
         handle.spawn(
             tx.sink_map_err(Into::into)
@@ -158,7 +157,7 @@ impl PdReactor {
 
         handle.spawn(
             rx.try_for_each(move |resp| {
-                let mut client = client.wl();
+                let mut client = client.write().unwrap();
                 let reactor = &mut client.reactor;
                 let tso_pending = reactor.tso_pending.take().unwrap();
                 reactor.schedule(PdTask::Response(tso_pending, resp));
@@ -174,7 +173,7 @@ impl PdReactor {
     }
 
     fn tso_request(client: &Arc<RwLock<LeaderClient>>) {
-        let mut client = client.wl();
+        let mut client = client.write().unwrap();
         let cluster_id = client.cluster_id;
         let reactor = &mut client.reactor;
         let mut tso_batch = reactor.tso_buffer.take().unwrap();
@@ -203,7 +202,7 @@ impl PdReactor {
                 })
                 .unwrap();
         }
-        client.wl().reactor.tso_buffer = Some(requests);
+        client.write().unwrap().reactor.tso_buffer = Some(requests);
     }
 
     fn dispatch(client: &Arc<RwLock<LeaderClient>>, task: PdTask, handle: &OtherHandle) {
@@ -264,7 +263,7 @@ impl LeaderClient {
             timeout,
         }));
 
-        client.wl().reactor.start(Arc::clone(&client));
+        client.write().unwrap().reactor.start(Arc::clone(&client));
         Ok(client)
     }
 
@@ -276,7 +275,7 @@ impl LeaderClient {
     pub fn reconnect(leader: &Arc<RwLock<LeaderClient>>, interval: u64) -> Result<()> {
         warn!("updating pd client, blocking the tokio core");
         let ((client, members), start) = {
-            let leader = leader.rl();
+            let leader = leader.read().unwrap();
             if leader.last_update.elapsed() < Duration::from_secs(interval) {
                 // Avoid unnecessary updating.
                 return Ok(());
@@ -292,7 +291,7 @@ impl LeaderClient {
 
         {
             let leader_clone = Arc::clone(leader);
-            let mut leader = leader.wl();
+            let mut leader = leader.write().unwrap();
             leader.client = client;
             leader.members = members;
             leader.last_update = Instant::now();

src/rpc/util.rs

@@ -1,10 +1,6 @@
 // Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
 
-use std::{
-    sync::{mpsc, RwLock, RwLockReadGuard, RwLockWriteGuard},
-    thread,
-    time::Duration,
-};
+use std::{sync::mpsc, thread, time::Duration};
 
 use lazy_static::*;
 use tokio_timer::{self, timer::Handle};
@@ -22,31 +18,14 @@ macro_rules! internal_err {
 }
 
 /// make a thread name with additional tag inheriting from current thread.
-macro_rules! thd_name {
+macro_rules! thread_name {
     ($name:expr) => {{
         $crate::rpc::util::get_tag_from_thread_name()
             .map(|tag| format!("{}::{}", $name, tag))
             .unwrap_or_else(|| $name.to_owned())
     }};
 }
 
-/// A handy shortcut to replace `RwLock` write/read().unwrap() pattern to
-/// shortcut wl and rl.
-pub trait HandyRwLock<T> {
-    fn wl(&self) -> RwLockWriteGuard<T>;
-    fn rl(&self) -> RwLockReadGuard<T>;
-}
-
-impl<T> HandyRwLock<T> for RwLock<T> {
-    fn wl(&self) -> RwLockWriteGuard<T> {
-        self.write().unwrap()
-    }
-
-    fn rl(&self) -> RwLockReadGuard<T> {
-        self.read().unwrap()
-    }
-}
-
 pub fn get_tag_from_thread_name() -> Option<String> {
     thread::current()
         .name()
@@ -69,7 +48,7 @@ lazy_static! {
 fn start_global_timer() -> Handle {
     let (tx, rx) = mpsc::channel();
     thread::Builder::new()
-        .name(thd_name!("timer"))
+        .name(thread_name!("timer"))
         .spawn(move || {
             let mut timer = tokio_timer::Timer::default();
             tx.send(timer.handle()).unwrap();
@@ -83,46 +62,7 @@ fn start_global_timer() -> Handle {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use futures::compat::Compat01As03;
-    use std::*;
-
-    #[test]
-    fn test_rwlock_deadlock() {
-        // If the test runs over 60s, then there is a deadlock.
-        let mu = RwLock::new(Some(1));
-        {
-            let _clone = foo(&mu.rl());
-            let mut data = mu.wl();
-            assert!(data.is_some());
-            *data = None;
-        }
-
-        {
-            match foo(&mu.rl()) {
-                Some(_) | None => {
-                    let res = mu.try_write();
-                    assert!(res.is_err());
-                }
-            }
-        }
-
-        #[cfg_attr(feature = "cargo-clippy", allow(clippy::clone_on_copy))]
-        fn foo(a: &Option<usize>) -> Option<usize> {
-            a.clone()
-        }
-    }
-
-    #[test]
-    fn test_internal_error() {
-        let file_name = file!();
-        let line_number = line!();
-        let e = internal_err!("{}", "hi");
-        assert_eq!(
-            format!("{}", e),
-            format!("[{}:{}]: hi", file_name, line_number + 1)
-        );
-    }
 
     #[test]
     fn test_global_timer() {