time: 自定义ds::time，后续替换std::time。默认使用std::time

agent/src/main.rs

@@ -13,7 +13,7 @@ use discovery::*;
 mod init;
 
 use rt::spawn;
-use std::time::Duration;
+use ds::time::Duration;
 
 use protocol::Result;
 

agent/src/prometheus.rs

@@ -19,7 +19,7 @@ pub struct PrometheusMetricsResponse {}
 
 use ds::lock::Lock;
 use lazy_static::lazy_static;
-use std::time::Instant;
+use ds::time::Instant;
 
 lazy_static! {
     static ref LAST: Lock<Instant> = Instant::now().into();
@@ -71,7 +71,7 @@ pub(crate) fn register_target(ctx: &context::Context) {
             port
         );
         let client = reqwest::Client::new();
-        let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));
+        let mut interval = tokio::time::interval(ds::time::Duration::from_secs(60));
         let mut q = vec![("refresh", true)];
         loop {
             let body = body.clone();

agent/src/service.rs

@@ -2,7 +2,7 @@ use context::Quadruple;
 use net::Listener;
 use rt::spawn;
 use std::sync::Arc;
-use std::time::Duration;
+use ds::time::Duration;
 
 use discovery::TopologyWriteGuard;
 use ds::chan::Sender;

context/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2021"
 
 [dependencies]
 log = { path = "../log" }
+ds = { path = "../ds" }
 
 clap = {version = "3.1.1", features = ["derive"] }
 url = "2.2.2"

context/src/lib.rs

@@ -134,9 +134,9 @@ impl ContextOption {
         Ok(())
     }
 
-    pub fn tick(&self) -> std::time::Duration {
+    pub fn tick(&self) -> ds::time::Duration {
         assert!(self.tick_sec >= 1 && self.tick_sec <= 60);
-        std::time::Duration::from_secs(self.tick_sec as u64)
+        ds::time::Duration::from_secs(self.tick_sec as u64)
     }
     // 如果是以升级模式启动，则会将原有的端口先关闭。
     pub fn listeners(&self) -> ListenerIter {

context/src/quadruple.rs

@@ -1,5 +1,5 @@
 use std::path::Path;
-use std::time::{Duration, Instant};
+use ds::time::{Duration, Instant};
 #[derive(Debug, Clone, Eq)]
 pub struct Quadruple {
     parsed_at: Instant,

discovery/src/cfg.rs

@@ -1,6 +1,6 @@
 use std::io::{Error, ErrorKind, Result};
 use std::path::PathBuf;
-use std::time::{Duration, Instant};
+use ds::time::{Duration, Instant};
 
 use tokio::fs::File;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};

discovery/src/dns.rs

@@ -166,7 +166,7 @@ pub async fn start_dns_resolver_refresher() {
     use std::task::Poll;
     let noop = noop_waker::noop_waker();
     let mut ctx = std::task::Context::from_waker(&noop);
-    use std::time::{Duration, Instant};
+    use ds::time::{Duration, Instant};
     const CYCLE: Duration = Duration::from_secs(79);
     let mut tick = tokio::time::interval(Duration::from_secs(1));
     let mut last = Instant::now(); // 上一次刷新的时间

discovery/src/update.rs

@@ -1,7 +1,7 @@
 // 定期更新discovery.
 use super::{Discover, ServiceId, TopologyWrite};
 use ds::chan::Receiver;
-use std::time::{Duration, Instant};
+use ds::time::{Duration, Instant};
 use tokio::time::interval;
 
 use crate::cache::DiscoveryCache;

ds/src/mem/policy.rs

@@ -1,5 +1,5 @@
 const BUF_MIN: usize = 2 * 1024;
-use std::time::Instant;
+use crate::time::Instant;
 // 内存需要缩容时的策略
 // 为了避免频繁的缩容，需要设置一个最小频繁，通常使用最小间隔时间
 pub struct MemPolicy {
@@ -108,8 +108,8 @@ impl Display for MemPolicy {
 
 #[cfg(debug_assertions)]
 mod trace {
+    use crate::time::Instant;
     use std::fmt::{self, Debug, Formatter};
-    use std::time::Instant;
     pub(super) struct Trace {
         direction: &'static str, // 方向: true为tx, false为rx. 打日志用
         id: usize,

endpoint/src/cacheservice/config.rs

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 use sharding::hash;
-//use std::time::Duration;
+//use ds::time::Duration;
 
 #[derive(Serialize, Deserialize, Clone, Debug, Default, Hash)]
 pub struct Namespace {

endpoint/src/cacheservice/topo.rs

@@ -3,7 +3,7 @@ use protocol::{Builder, Endpoint, Protocol, Request, Resource, Topology, TryNext
 use sharding::hash::Hasher;
 use sharding::Distance;
 use std::collections::HashMap;
-use std::time::Duration;
+use ds::time::Duration;
 
 use crate::TimeoutAdjust;
 use stream::Shards;

endpoint/src/lib.rs

@@ -24,7 +24,7 @@ trait TimeoutAdjust: Sized {
     }
 }
 
-use std::time::Duration;
+use ds::time::Duration;
 impl TimeoutAdjust for Duration {
     fn adjust(&mut self, ms: u32) {
         if ms > 0 {

endpoint/src/phantomservice/config.rs

@@ -1,4 +1,4 @@
-//use std::time::Duration;
+//use ds::time::Duration;
 
 use serde::{Deserialize, Serialize};
 

endpoint/src/redisservice/config.rs

@@ -1,5 +1,5 @@
 use std::collections::{HashMap, HashSet};
-//use std::time::Duration;
+//use ds::time::Duration;
 
 use serde::{Deserialize, Serialize};
 //use sharding::distribution::{DIST_ABS_MODULA, DIST_MODULA};

endpoint/src/redisservice/topo.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
-use std::time::Duration;
+use ds::time::Duration;
 
 use discovery::TopologyWrite;
 use protocol::{Builder, Endpoint, Protocol, Request, Resource, Single, Topology};

metrics/src/macros.rs

@@ -42,7 +42,7 @@ impl<T: ToNumber> MetricData for T {
         crate::register_cache(id, self.int());
     }
 }
-use std::time::Duration;
+use ds::time::Duration;
 impl MetricData for Duration {
     #[inline]
     fn incr_to(self, data: &ItemData) {

metrics/src/register.rs

@@ -168,7 +168,7 @@ pub struct MetricRegister {
 
 impl MetricRegister {
     fn new(rx: Receiver<(Arc<Id>, i64)>, metrics: CowWriteHandle<Metrics>) -> Self {
-        let mut tick = interval(std::time::Duration::from_secs(3));
+        let mut tick = interval(ds::time::Duration::from_secs(3));
         tick.set_missed_tick_behavior(MissedTickBehavior::Skip);
         Self {
             rx,

metrics/src/sender.rs

@@ -2,7 +2,7 @@
 //use std::future::Future;
 //use std::pin::Pin;
 //use std::task::{Context, Poll};
-//use std::time::{Duration, Instant};
+//use ds::time::{Duration, Instant};
 //
 //use std::task::ready;
 //use tokio::time::{interval, Interval, MissedTickBehavior};

metrics/src/types/host.rs

@@ -2,7 +2,7 @@ use psutil::process::Process;
 
 use crate::BASE_PATH;
 use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering};
-use std::time::Instant;
+use ds::time::Instant;
 
 static CPU_PERCENT: AtomicUsize = AtomicUsize::new(0);
 static MEMORY: AtomicI64 = AtomicI64::new(0);

metrics/src/types/rtt.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use ds::time::Duration;
 
 use crate::{Id, ItemWriter, NumberInner};
 pub const MAX: Duration = Duration::from_millis(30);

protocol/src/callback.rs

@@ -1,6 +1,6 @@
 use std::mem::MaybeUninit;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::time::Instant;
+use ds::time::Instant;
 
 use ds::AtomicWaker;
 

protocol/src/error.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use ds::time::Duration;
 
 #[derive(Debug)]
 pub enum Error {

protocol/src/lib.rs

@@ -47,7 +47,7 @@ impl Resource {
     }
 }
 
-use std::time::Duration;
+use ds::time::Duration;
 pub trait Builder<P, R, E> {
     fn build(addr: &str, parser: P, rsrc: Resource, service: &str, timeout: Duration) -> E
     where

protocol/src/req.rs

@@ -1,5 +1,5 @@
 use std::fmt::{Debug, Display};
-use std::time::Instant;
+use ds::time::Instant;
 
 use ds::RingSlice;
 

protocol/src/request.rs

@@ -7,7 +7,7 @@ pub struct Request {
 
 impl crate::Request for Request {
     #[inline]
-    fn start_at(&self) -> std::time::Instant {
+    fn start_at(&self) -> ds::time::Instant {
         self.ctx().start_at()
     }
 

rt/src/entry.rs

@@ -2,7 +2,7 @@ use std::fmt::Debug;
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use std::time::{Duration, Instant};
+use ds::time::{Duration, Instant};
 
 use metrics::{Metric, Path};
 use std::task::ready;

stream/src/builder.rs

@@ -17,7 +17,7 @@ pub struct BackendBuilder<P, R> {
     _marker: std::marker::PhantomData<(P, R)>,
 }
 
-use std::time::Duration;
+use ds::time::Duration;
 impl<P: Protocol, R: Request> protocol::Builder<P, R, Arc<Backend<R>>> for BackendBuilder<P, R> {
     fn build(
         addr: &str,

stream/src/checker.rs

@@ -1,5 +1,5 @@
 use std::sync::{atomic::AtomicBool, Arc};
-use std::time::Duration;
+use ds::time::Duration;
 
 use tokio::net::TcpStream;
 use tokio::time::timeout;

stream/src/pipeline.rs

@@ -3,7 +3,7 @@ use std::future::Future;
 use std::pin::Pin;
 use std::sync::atomic::{AtomicUsize, Ordering::*};
 use std::task::{ready, Context, Poll};
-use std::time::{Duration, Instant};
+use ds::time::{Duration, Instant};
 
 use tokio::io::{AsyncRead, AsyncWrite};
 

stream/src/reconn.rs

@@ -1,7 +1,7 @@
 use metrics::{Metric, Path};
 use std::sync::atomic::{AtomicBool, Ordering::Acquire};
 use std::sync::Arc;
-use std::time::Duration;
+use ds::time::Duration;
 pub(crate) struct ReconnPolicy {
     single: Arc<AtomicBool>,
     conns: usize,

tests/src/asserts.rs

@@ -0,0 +1,4 @@
+#[test]
+fn test_assert() {
+    ds::assert!(1 == 1, "assert true");
+}

tests/src/mem.rs

@@ -63,7 +63,7 @@ fn ring_buffer() {
     rrb.advance_write(1024);
 
     // 等待10ms。（默认是4ms）
-    std::thread::sleep(std::time::Duration::from_millis(10));
+    std::thread::sleep(ds::time::Duration::from_millis(10));
     let buf = rrb.as_mut_bytes();
     assert_eq!(buf.len(), 1024);
     rrb.advance_write(1024);

tests/src/ring_buffer.rs

@@ -1,7 +1,7 @@
 #[cfg(test)]
 mod tests {
     use ds::{RingBuffer, RingSlice};
-    use std::time::{Duration, Instant};
+    use ds::time::{Duration, Instant};
 
     fn rnd_bytes(size: usize) -> Vec<u8> {
         let data: Vec<u8> = (0..size).map(|_| rand::random::<u8>()).collect();
@@ -64,7 +64,7 @@ mod tests {
         assert_eq!(buf.len(), 0);
 
         // 等待10ms。（默认是4ms）
-        std::thread::sleep(std::time::Duration::from_millis(10));
+        std::thread::sleep(ds::time::Duration::from_millis(10));
         let buf = rrb.as_mut_bytes();
         assert_eq!(buf.len(), 1024);
         rrb.advance_write(1024);
@@ -76,7 +76,7 @@ mod tests {
         let ins = Instant::now();
         loop {
             rrb.advance_write(0);
-            std::thread::sleep(std::time::Duration::from_millis(3));
+            std::thread::sleep(ds::time::Duration::from_millis(3));
             if ins.elapsed() >= Duration::from_secs(70) {
                 break;
             }