Merge pull request #58 from weibocom/dev

Dev

agent/Cargo.toml

@@ -17,12 +17,12 @@ elog = { path = "../log", package = "log" }
 stream = { path = "../stream" }
 metrics = { path = "../metrics" }
 rt = { path = "../rt" }
+ds = { path = "../ds" }
 
-crossbeam-channel = "0.5.1"
 log = "0.4.14"
-ds = { path = "../ds" }
 backtrace = "0.3.63"
 
 tokio = {version = "1.15.0", features = ["rt", "net", "rt-multi-thread", "time", "macros"]}
 mimalloc = { version = "*", default-features = false }
 
+rlimit = "0.3.0"

agent/src/main.rs

@@ -4,7 +4,6 @@ static GLOBAL: MiMalloc = MiMalloc;
 
 mod service;
 use context::Context;
-use crossbeam_channel::bounded;
 use discovery::*;
 
 use rt::spawn;
@@ -44,6 +43,10 @@ async fn main() -> Result<()> {
     }));
     let ctx = Context::from_os_args();
     ctx.check()?;
+    // set number of file
+    if let Err(e) = rlimit::setrlimit(rlimit::Resource::NOFILE, ctx.no_file, ctx.no_file) {
+        log::info!("set rlimit to {} failed:{:?}", ctx.no_file, e);
+    }
 
     let _l = service::listener_for_supervisor(ctx.port()).await?;
     elog::init(ctx.log_dir(), &ctx.log_level)?;
@@ -57,7 +60,7 @@ async fn main() -> Result<()> {
     spawner.spawn(discovery::dns::start_dns_resolver_refresher());
     use discovery::watch_discovery;
     let discovery = Discovery::from_url(ctx.discovery());
-    let (tx, rx) = bounded(128);
+    let (tx, rx) = ds::chan::bounded(128);
     let snapshot = ctx.snapshot().to_string();
     let tick = ctx.tick();
     let mut fix = discovery::Fixed::default();

agent/src/service.rs

@@ -4,8 +4,8 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use context::Quadruple;
-use crossbeam_channel::Sender;
 use discovery::TopologyWriteGuard;
+use ds::chan::Sender;
 use metrics::Path;
 use protocol::callback::{Callback, CallbackPtr};
 use protocol::{Parser, Result};
@@ -52,7 +52,6 @@ pub(super) async fn process_one(
     }
     switcher.off();
 
-    // TODO 延迟一秒，释放top内存。
     // 因为回调，有可能在连接释放的时候，还在引用top。
     tokio::time::sleep(Duration::from_secs(3)).await;
     Ok(())
@@ -78,6 +77,7 @@ async fn _process_one(
         let p = p.clone();
         let cb = cb.clone();
         let metrics = StreamMetrics::new(path);
+        let path = format!("{:?}", path);
         log::debug!("connection established:{:?}", path);
         spawn(async move {
             use protocol::Topology;
@@ -87,7 +87,7 @@ async fn _process_one(
                 match e {
                     Error::Quit => {} // client发送quit协议退出
                     Error::ReadEof => {}
-                    e => log::info!("disconnected. {:?} ", e),
+                    e => log::info!("{:?} disconnected. {:?}", path, e),
                 }
             }
         });

context/src/lib.rs

@@ -16,6 +16,9 @@ pub struct Context {
     #[clap(long, help("port for suvervisor"), default_value("9984"))]
     port: u16,
 
+    #[clap(long, help("number of open file"), default_value("204800"))]
+    pub no_file: u64,
+
     #[clap(
         short,
         long,
@@ -144,7 +147,7 @@ impl ListenerIter {
                         continue;
                     }
                     if let Some(one) = Quadruple::parse(&name) {
-                        log::info!("service parsed :{}", one);
+                        log::debug!("service parsed :{}", one);
                         listeners.push(one);
                         self.processed.insert(name.to_string(), ());
                     }

discovery/Cargo.toml

@@ -27,7 +27,6 @@ serde_yaml = "0.8.17"
 serde_json = "1.0.65"
 rand = "0.8.4"
 futures = "0.3.16"
-crossbeam-channel = "0.5.1"
 md5 = "0.7"
 bs58 = "0.4"
 trust-dns-resolver = "0.20.0"

discovery/src/update.rs

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

ds/Cargo.toml

@@ -15,9 +15,8 @@ rand = "0.8.4"
 cache_line_size = "1.0.0"
 log = "0.4.14"
 crossbeam-queue = "0.3.2"
+crossbeam-channel = "0.5.1"
 cfg-if = "1"
 futures = "0.3.17"
 atomic-waker = "*"
-
 backtrace = "0.3.63"
-

ds/src/chan/mod.rs

@@ -0,0 +1,3 @@
+pub type Sender<T> = crossbeam_channel::Sender<T>;
+pub type Receiver<T> = crossbeam_channel::Receiver<T>;
+pub use crossbeam_channel::bounded;

ds/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod chan;
 mod cow;
 mod mem;
 pub mod queue;

ds/src/queue/mod.rs

@@ -0,0 +1,2 @@
+mod pinned;
+pub use pinned::PinnedQueue;

endpoint/src/cacheservice/config.rs

@@ -1,4 +1,5 @@
 use serde::{Deserialize, Serialize};
+use std::time::Duration;
 
 #[derive(Serialize, Deserialize, Clone, Debug, Default, Hash)]
 pub struct Namespace {
@@ -18,20 +19,37 @@ pub struct Namespace {
     pub slave: Vec<String>,
     #[serde(default)]
     pub slave_l1: Vec<Vec<String>>,
+
+    #[serde(default)]
+    pub timeout_ms_master: u32,
+    #[serde(default)]
+    pub timeout_ms_slave: u32,
 }
 
 impl Namespace {
-    pub(crate) fn parse<F: FnMut(Self)>(cfg: &str, namespace: &str, mut f: F) {
+    pub(crate) fn try_from(cfg: &str, namespace: &str) -> Option<Self> {
         log::debug!("namespace:{} cfg:{} updating", namespace, cfg);
         match serde_yaml::from_str::<Namespace>(cfg) {
-            Ok(ns) => {
-                f(ns);
-            }
             Err(e) => {
                 log::warn!("parse namespace error. {} msg:{:?}", namespace, e);
+                None
+            }
+            Ok(ns) => {
+                if ns.master.len() == 0 {
+                    log::info!("cache service master empty. namespace:{}", namespace);
+                    None
+                } else {
+                    Some(ns)
+                }
             }
         }
     }
+    pub(super) fn timeout_master(&self) -> Duration {
+        Duration::from_millis(250.max(self.timeout_ms_master as u64))
+    }
+    pub(super) fn timeout_slave(&self) -> Duration {
+        Duration::from_millis(100.max(self.timeout_ms_slave as u64))
+    }
 }
 
 impl Namespace {}

endpoint/src/cacheservice/topo.rs

@@ -159,11 +159,7 @@ where
 {
     #[inline]
     fn update(&mut self, namespace: &str, cfg: &str) {
-        super::config::Namespace::parse(cfg, namespace, |ns| {
-            if ns.master.len() == 0 {
-                log::info!("cache service master empty. namespace:{}", namespace);
-                return;
-            }
+        if let Some(ns) = super::config::Namespace::try_from(cfg, namespace) {
             self.hasher = Hasher::from(&ns.hash);
             let dist = &ns.distribution;
 
@@ -179,12 +175,13 @@ where
                     old.insert(e.1, e.0);
                 }
             }
-            let mto = Duration::from_millis(500);
+            let mto = ns.timeout_master();
+            let rto = ns.timeout_slave();
+
             // 准备master
             let master = self.build(old, ns.master, dist, namespace, mto);
             self.streams.push(master);
 
-            let rto = Duration::from_millis(150);
             // master_l1
             self.has_l1 = ns.master_l1.len() > 0;
             for l1 in ns.master_l1 {
@@ -203,8 +200,8 @@ where
                 let g = self.build(old, sl1, dist, namespace, rto);
                 self.streams.push(g);
             }
-            // old 会被dopped
-        });
+        }
+        // old 会被dopped
     }
     // 不同的业务共用一个配置。把不同的业务配置给拆分开
     #[inline]