Merge pull request #441 from weibocom/dev_mem

优化内存分配

Author: parabala

Cargo.toml

@@ -44,5 +44,5 @@ codegen-units = 256
 
 
 [workspace.dependencies]
-tokio = { version = "1.32", features = ["rt", "fs", "net", "rt-multi-thread", "time", "sync", "signal"], default-features = false }
+tokio = { version = "1.36", features = ["io-util", "rt", "fs", "net", "rt-multi-thread", "time", "sync", "signal"], default-features = false }
 serde = { version = "1.0.126", features = ["derive"], default-features = false }

agent/src/main.rs

@@ -1,4 +1,4 @@
-use ds::{chan::Receiver, BrzMalloc};
+use ds::BrzMalloc;
 #[global_allocator]
 static GLOBAL: BrzMalloc = BrzMalloc {};
 
@@ -13,10 +13,7 @@ mod init;
 use ds::time::{sleep, Duration};
 use rt::spawn;
 
-use protocol::{Parser, Result};
-use stream::{Backend, Request};
-type Endpoint = Backend<Request>;
-type Topology = endpoint::TopologyProtocol<Endpoint, Parser>;
+use protocol::Result;
 
 // 默认支持
 fn main() -> Result<()> {
@@ -58,7 +55,7 @@ async fn run() -> Result<()> {
     }
 }
 
-async fn discovery_init(ctx: &Context, rx: Receiver<TopologyWriteGuard<Topology>>) -> Result<()> {
+async fn discovery_init(ctx: &Context, rx: service::Receiver) -> Result<()> {
     // 将dns resolver的初始化放到外层，提前进行，避免并发场景下顺序错乱 fishermen
     let discovery = discovery::Discovery::from_url(&ctx.discovery);
     let snapshot = ctx.snapshot_path.to_string();

agent/src/prometheus.rs

@@ -54,10 +54,10 @@ pub(crate) fn register_target(ctx: &context::Context) {
   "target": "{local_ip}:{port}"
 }}"#
         );
+        let body: &'static str = Box::leak(body.into_boxed_str());
         let client = Client::new();
         let mut interval = interval(Duration::from_secs(60));
         loop {
-            let body = body.clone();
             let req = Request::builder()
                 .method("PUT")
                 .uri(&url)

agent/src/service.rs

@@ -5,7 +5,6 @@ use rt::spawn;
 use std::sync::Arc;
 
 use discovery::{TopologyReadGuard, TopologyWriteGuard};
-use ds::chan::Sender;
 use metrics::Path;
 use protocol::{Parser, Result};
 use stream::pipeline::copy_bidirectional;
@@ -16,15 +15,18 @@ type Topology = endpoint::TopologyProtocol<Endpoint, Parser>;
 use metrics::Status;
 // 一直侦听，直到成功侦听或者取消侦听（当前尚未支持取消侦听）
 // 1. 尝试侦听之前，先确保服务配置信息已经更新完成
+type Sender = ds::chan::Sender<(String, TopologyWriteGuard<Topology>)>;
+pub(super) type Receiver = ds::chan::Receiver<(String, TopologyWriteGuard<Topology>)>;
 pub(super) async fn process_one(
     quard: &Quadruple,
-    discovery: Sender<TopologyWriteGuard<Topology>>,
+    discovery: Sender,
 ) -> std::result::Result<(), Box<dyn std::error::Error>> {
     let p = Parser::try_from(&quard.protocol())?;
     let top = endpoint::TopologyProtocol::try_from(p.clone(), quard.endpoint())?;
     let (tx, rx) = discovery::topology(top, &quard.service());
     // 注册，定期更新配置
-    discovery.send(tx).await.map_err(|e| e.to_string())?;
+    let service = quard.service().to_string();
+    discovery.send((service, tx)).await?;
 
     let path = Path::new(vec![quard.protocol(), &quard.biz()]);
     let mut metrics = StreamMetrics::new(&path);

ci.sh

@@ -2,15 +2,11 @@
 brz_home="/data1/ci/breeze"
 mkdir -p $brz_home
 
-docker ps -a | grep breeze_ci_mysql4623 && docker rm -f breeze_ci_mysql4623
-docker ps -a | grep breeze_ci_mysql4624 && docker rm -f breeze_ci_mysql4624
+docker stop breeze_ci_mysql4623 breeze_ci_mysql4624 breeze_github_ci 2>/dev/null
+
 docker run --rm --name breeze_ci_mysql4623  -p 4623:3306 -d parabala/mysqlci_with_schema:v0.0.2
 docker run --rm --name breeze_ci_mysql4624  -p 4624:3306 -d parabala/mysqlci_with_schema:v0.0.2
-
 container_name=breeze_github_ci
-docker ps -a | grep "$container_name" && docker rm -f "$container_name"
-
-
 docker run --rm -d -v $brz_home:/data1/resource/breeze  --net="host"  --name "$container_name"  parabala/breeze:githubci108
 
 # rm -rf $brz_home/*
@@ -75,3 +71,4 @@ RUST_BACKTRACE=1 cargo test -p tests_integration --features github_workflow
 
 kill -9 $pid
 
+docker stop breeze_ci_mysql4623 breeze_ci_mysql4624 breeze_github_ci 

context/src/lib.rs

@@ -73,6 +73,11 @@ pub struct ContextOption {
     )]
     pub service_path: String,
 
+    #[clap(
+        long,
+        help("clean path for unix domain socket ."),
+        default_value("false")
+    )]
     #[clap(short, long, help("starting in upgrade mode"))]
     upgrade: bool,
 

discovery/Cargo.toml

@@ -28,7 +28,8 @@ serde_json = "1.0.65"
 rand = "0.8.4"
 md5 = "0.7"
 bs58 = "0.4"
-trust-dns-resolver = {version = "0.23.0"}
 once_cell = "1.14.0"
 noop-waker = "0.1.0"
 ctor = "*"
+dns-lookup = "*"
+libc = "*"