Merge pull request #112 from sticnarf/remove-tokio-timer

Replace tokio-timer with futures_timer

Cargo.toml

@@ -28,7 +28,6 @@ log = "0.3.9"
 regex = "1"
 serde = "1.0"
 serde_derive = "1.0"
-tokio-timer = "0.2"
 futures-timer = "0.3"
 
 [dependencies.prometheus]

src/pd/retry.rs

@@ -5,18 +5,17 @@
 use std::{
     fmt,
     sync::{Arc, RwLock},
-    time::{Duration, Instant},
+    time::Duration,
 };
 
-use futures::compat::Compat01As03;
 use futures::future::BoxFuture;
 use futures::prelude::*;
 use futures::ready;
 use futures::task::{Context, Poll};
+use futures_timer::Delay;
 use grpcio::Environment;
 use kvproto::metapb;
 use std::pin::Pin;
-use tokio_timer::timer::Handle;
 
 use crate::{
     pd::{
@@ -25,7 +24,6 @@ use crate::{
     },
     security::SecurityManager,
     transaction::Timestamp,
-    util::GLOBAL_TIMER_HANDLE,
     Result,
 };
 
@@ -164,7 +162,6 @@ struct Request<Func> {
     request_sent: usize,
 
     client: Arc<RetryClient>,
-    timer: Handle,
 
     // A function which makes an async request.
     func: Func,
@@ -202,7 +199,6 @@ where
         Request {
             request_sent: 0,
             client,
-            timer: GLOBAL_TIMER_HANDLE.clone(),
             func,
         }
     }
@@ -219,11 +215,7 @@ where
                 future::Either::Left(future::ok(()))
             }
             Err(_) => future::Either::Right(
-                Compat01As03::new(
-                    self.timer
-                        .delay(Instant::now() + Duration::from_secs(RECONNECT_INTERVAL_SEC)),
-                )
-                .map(|_| Err(())),
+                Delay::new(Duration::from_secs(RECONNECT_INTERVAL_SEC)).map(|_| Err(())),
             ),
         }
     }

src/util.rs

@@ -1,7 +1,6 @@
 // Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
 
-use std::{sync::mpsc, thread, time::Duration};
-use tokio_timer::{self, timer::Handle};
+use std::{thread, time::Duration};
 
 macro_rules! internal_err {
     ($e:expr) => ({
@@ -38,37 +37,3 @@ pub fn duration_to_sec(d: Duration) -> f64 {
     // In most cases, we can't have so large Duration, so here just panic if overflow now.
     d.as_secs() as f64 + (nanos / 1_000_000_000.0)
 }
-
-lazy_static! {
-    pub static ref GLOBAL_TIMER_HANDLE: Handle = start_global_timer();
-}
-
-fn start_global_timer() -> Handle {
-    let (tx, rx) = mpsc::channel();
-    thread::Builder::new()
-        .name(thread_name!("timer"))
-        .spawn(move || {
-            let mut timer = tokio_timer::Timer::default();
-            tx.send(timer.handle()).unwrap();
-            loop {
-                timer.turn(None).unwrap();
-            }
-        })
-        .unwrap();
-    rx.recv().unwrap()
-}
-
-#[cfg(test)]
-mod tests {
-    use futures::compat::Compat01As03;
-
-    #[test]
-    fn test_global_timer() {
-        let handle = super::GLOBAL_TIMER_HANDLE.clone();
-        let delay =
-            handle.delay(::std::time::Instant::now() + ::std::time::Duration::from_millis(100));
-        let timer = ::std::time::Instant::now();
-        futures::executor::block_on(Compat01As03::new(delay)).unwrap();
-        assert!(timer.elapsed() >= ::std::time::Duration::from_millis(100));
-    }
-}