From e4fb2ad74fead18752d8cd0ecfca578d01299af8 Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Mon, 10 Feb 2025 18:33:38 +0100 Subject: [PATCH] Something is ok (cleanup) --- src/web/proxyhttp.rs | 25 ++++++++++++++----------- src/web/start.rs | 35 +++++++++-------------------------- 2 files changed, 23 insertions(+), 37 deletions(-) diff --git a/src/web/proxyhttp.rs b/src/web/proxyhttp.rs index e247677..0788f8f 100644 --- a/src/web/proxyhttp.rs +++ b/src/web/proxyhttp.rs @@ -16,10 +16,12 @@ use tokio::sync::RwLock; use tokio::time::interval; pub struct LB { - pub upstreams_map: Arc, AtomicUsize)>>>, + pub upstreams: Arc, AtomicUsize)>>>, + // pub umap_full: Arc, AtomicUsize)>>>, } pub struct BGService { - pub upstreams_map: Arc, AtomicUsize)>>>, + pub upstreams: Arc, AtomicUsize)>>>, + // pub umap_full: Arc, AtomicUsize)>>>, } #[async_trait] @@ -33,7 +35,8 @@ impl BackgroundService for BGService { break; } _ = period.tick() => { - let map_write = self.upstreams_map.write().await; + let map_write = self.upstreams.write().await; + // let newups : DashMap, AtomicUsize)> = DashMap::new(); let newmap = discover(); if compare::dashmaps(&map_write, &newmap) { println!("DashMaps are equal. Chilling out."); @@ -44,7 +47,7 @@ impl BackgroundService for BGService { map_write.insert(k,v); } } - drop(map_write); // Important: Release the lock + drop(map_write); } } } @@ -52,7 +55,7 @@ impl BackgroundService for BGService { } fn discover() -> DashMap, AtomicUsize)> { - let upstreams_map: DashMap, AtomicUsize)> = DashMap::new(); + let upstreams: DashMap, AtomicUsize)> = DashMap::new(); let mut toreturn = vec![]; toreturn.push(("192.168.1.1".to_string(), 8000.to_owned())); toreturn.push(("192.168.1.10".to_string(), 8000.to_owned())); @@ -62,15 +65,15 @@ fn discover() -> DashMap, AtomicUsize)> { toreturn.push(("127.0.0.4".to_string(), 8000.to_owned())); toreturn.push(("127.0.0.5".to_string(), 8000.to_owned())); toreturn.push(("127.0.0.6".to_string(), 8000.to_owned())); - upstreams_map.insert("myip.netangels.net".to_string(), (toreturn, AtomicUsize::new(0))); + upstreams.insert("myip.netangels.net".to_string(), (toreturn, AtomicUsize::new(0))); let mut toreturn = vec![]; toreturn.push(("192.168.1.1".to_string(), 8000.to_owned())); toreturn.push(("192.168.1.10".to_string(), 8000.to_owned())); - upstreams_map.insert("polo.netangels.net".to_string(), (toreturn, AtomicUsize::new(0))); + upstreams.insert("polo.netangels.net".to_string(), (toreturn, AtomicUsize::new(0))); let mut toreturn = vec![]; toreturn.push(("192.168.1.20".to_string(), 8000.to_owned())); - upstreams_map.insert("glop.netangels.net".to_string(), (toreturn, AtomicUsize::new(0))); - upstreams_map + upstreams.insert("glop.netangels.net".to_string(), (toreturn, AtomicUsize::new(0))); + upstreams } #[async_trait] @@ -80,7 +83,7 @@ pub trait GetHost { #[async_trait] impl GetHost for LB { async fn get_host(&self, peer: &str) -> Option<(String, u16)> { - let map_read = self.upstreams_map.read().await; + let map_read = self.upstreams.read().await; let x = if let Some(entry) = map_read.get(peer) { let (servers, index) = entry.value(); // No clone here @@ -89,7 +92,7 @@ impl GetHost for LB { } let idx = index.fetch_add(1, Ordering::Relaxed) % servers.len(); println!("{} {:?} => len: {}, idx: {}", peer, servers[idx], servers.len(), idx); - Some(servers[idx].clone()) // Clone the server address + Some(servers[idx].clone()) } else { None }; diff --git a/src/web/start.rs b/src/web/start.rs index 80cf0f4..fa35da1 100644 --- a/src/web/start.rs +++ b/src/web/start.rs @@ -6,28 +6,6 @@ use std::sync::atomic::AtomicUsize; use std::sync::Arc; use tokio::sync::RwLock; -/* -pub fn run1() { - let mut upstreams = LoadBalancer::try_from_iter(vec!["192.168.1.10:8000", "192.168.1.1:8000", "127.0.0.1:8000"]).unwrap(); - env_logger::init(); - let hc = TcpHealthCheck::new(); - upstreams.set_health_check(hc); - upstreams.health_check_frequency = Some(Duration::from_secs(1)); - - let background = background_service("health check", upstreams); - let upstreams = background.task(); - let mut server = Server::new(None).unwrap(); - server.bootstrap(); - - let mut proxy = pingora_proxy::http_proxy_service(&server.configuration, proxyhttp::LB(upstreams)); - - proxy.add_tcp("0.0.0.0:6193"); - server.add_service(background); - server.add_service(proxy); - server.run_forever(); -} -*/ - pub fn run() { env_logger::init(); @@ -35,13 +13,18 @@ pub fn run() { server.bootstrap(); let upstreams_map: DashMap, AtomicUsize)> = DashMap::new(); - let config = Arc::new(RwLock::new(upstreams_map)); // Wrap in Arc> + let config = Arc::new(RwLock::new(upstreams_map)); - let lb = LB { upstreams_map: config.clone() }; // Share the Arc> - let bg_service = BGService { upstreams_map: config.clone() }; // Share the Arc> + let lb = LB { + upstreams: config.clone(), + // umap_full: config.clone(), + }; + let bg_service = BGService { + upstreams: config.clone(), + // umap_full: config.clone(), + }; let bg_srvc = background_service("bgsrvc", bg_service); - bg_srvc.task(); let mut proxy = pingora_proxy::http_proxy_service(&server.configuration, lb); proxy.add_tcp("0.0.0.0:6193");