mirror of
https://github.com/sadoyan/aralez.git
synced 2026-04-29 22:38:36 +08:00
Removed unnecessary locks dashmaps
This commit is contained in:
@@ -1,9 +1,8 @@
|
||||
use crate::utils::tools::*;
|
||||
use std::collections::HashSet;
|
||||
use tokio::sync::RwLockReadGuard;
|
||||
|
||||
// #[allow(dead_code)]
|
||||
pub fn dm(map1: &RwLockReadGuard<UpstreamMap>, map2: &UpstreamMap) -> bool {
|
||||
pub fn dm(map1: &UpstreamMap, map2: &UpstreamMap) -> bool {
|
||||
if map1.len() != map2.len() {
|
||||
return false; // Different number of keys
|
||||
}
|
||||
|
||||
@@ -3,10 +3,9 @@ use dashmap::DashMap;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::RwLock;
|
||||
use tokio::time::interval;
|
||||
|
||||
pub async fn hc(upslist: Arc<RwLock<UpstreamMap>>, fullist: Arc<RwLock<UpstreamMap>>) {
|
||||
pub async fn hc(upslist: Arc<UpstreamMap>, fullist: Arc<UpstreamMap>) {
|
||||
let mut period = interval(Duration::from_secs(2));
|
||||
|
||||
loop {
|
||||
@@ -18,8 +17,7 @@ pub async fn hc(upslist: Arc<RwLock<UpstreamMap>>, fullist: Arc<RwLock<UpstreamM
|
||||
// println!("\nElapsed dash: {:.2?}", before.elapsed());
|
||||
// let before = Instant::now();
|
||||
{
|
||||
let full = fullist.read().await;
|
||||
for v in full.iter() {
|
||||
for v in fullist.iter() {
|
||||
fclone.insert(v.key().clone(), (v.value().0.clone(), AtomicUsize::new(0)));
|
||||
}
|
||||
} // lock releases when scope ends
|
||||
@@ -43,16 +41,15 @@ pub async fn hc(upslist: Arc<RwLock<UpstreamMap>>, fullist: Arc<RwLock<UpstreamM
|
||||
}
|
||||
// let before = Instant::now();
|
||||
{
|
||||
let upsl = upslist.read().await;
|
||||
if !crate::utils::compare::dm(&upsl, &totest) {
|
||||
if !crate::utils::compare::dm(&upslist, &totest) {
|
||||
println!("Dashmaps not matched, synchronizing");
|
||||
upsl.clear();
|
||||
upslist.clear();
|
||||
for (k, v) in totest { // loop takes the ownership
|
||||
println!("Host: {}", k);
|
||||
for vv in &v.0 {
|
||||
println!(" :===> {:?}", vv);
|
||||
}
|
||||
upsl.insert(k, v);
|
||||
upslist.insert(k, v);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,12 +13,11 @@ use pingora_http::{RequestHeader, ResponseHeader};
|
||||
use pingora_proxy::{ProxyHttp, Session};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::RwLock;
|
||||
// use tokio::time::Instant;
|
||||
|
||||
pub struct LB {
|
||||
pub upstreams: Arc<RwLock<UpstreamMap>>,
|
||||
pub umap_full: Arc<RwLock<UpstreamMap>>,
|
||||
pub upstreams: Arc<UpstreamMap>,
|
||||
pub umap_full: Arc<UpstreamMap>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -49,31 +48,21 @@ impl BackgroundService for LB {
|
||||
val = rx.next() => {
|
||||
match val {
|
||||
Some(newmap) => {
|
||||
let umap_work = self.upstreams.read().await;
|
||||
let umap_full = self.umap_full.read().await;
|
||||
match compare::dm(&umap_full, &newmap) {
|
||||
match compare::dm(&self.umap_full, &newmap) {
|
||||
false => {
|
||||
drop(umap_full);
|
||||
drop(umap_work);
|
||||
let work = self.upstreams.write().await;
|
||||
let full = self.umap_full.write().await;
|
||||
work.clear();
|
||||
full.clear();
|
||||
self.upstreams.clear();
|
||||
self.umap_full.clear();
|
||||
for (k,v) in newmap {
|
||||
println!("Host: {}", k);
|
||||
// <UpstreamMap
|
||||
for vv in v.0.clone() {
|
||||
println!(" ===> {:?}", vv);
|
||||
}
|
||||
work.insert(k.clone(), (v.0.clone(), AtomicUsize::new(0))); // No need for extra vec!
|
||||
full.insert(k, (v.0, AtomicUsize::new(0))); // Use `value.0` directly
|
||||
self.upstreams.insert(k.clone(), (v.0.clone(), AtomicUsize::new(0))); // No need for extra vec!
|
||||
self.umap_full.insert(k, (v.0, AtomicUsize::new(0))); // Use `value.0` directly
|
||||
}
|
||||
drop(full);
|
||||
drop(work);
|
||||
}
|
||||
true => {
|
||||
drop(umap_full);
|
||||
drop(umap_work);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -92,8 +81,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.read().await;
|
||||
let x = if let Some(entry) = map_read.get(peer) {
|
||||
let x = if let Some(entry) = self.upstreams.get(peer) {
|
||||
let (servers, index) = entry.value();
|
||||
if servers.is_empty() {
|
||||
return None;
|
||||
@@ -104,7 +92,6 @@ impl GetHost for LB {
|
||||
} else {
|
||||
None
|
||||
};
|
||||
drop(map_read);
|
||||
x
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ use dashmap::DashMap;
|
||||
use pingora_core::prelude::background_service;
|
||||
use pingora_core::server::Server;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
pub fn run() {
|
||||
env_logger::init();
|
||||
@@ -13,10 +12,10 @@ pub fn run() {
|
||||
server.bootstrap();
|
||||
|
||||
let upstreams_map: UpstreamMap = DashMap::new();
|
||||
let config = Arc::new(RwLock::new(upstreams_map));
|
||||
let config = Arc::new(upstreams_map);
|
||||
|
||||
let umap_full: UpstreamMap = DashMap::new();
|
||||
let fconfig = Arc::new(RwLock::new(umap_full));
|
||||
let fconfig = Arc::new(umap_full);
|
||||
|
||||
let lb = LB {
|
||||
upstreams: config.clone(),
|
||||
|
||||
Reference in New Issue
Block a user