net-pool 0.5.1

A set of types for network connection pool.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

use crate::backend::{Address, BackendState};
use crate::error::Error;
use crate::strategy::Strategy;
use std::sync::Arc;
use std::sync::atomic::Ordering::Relaxed;
use std::sync::atomic::{AtomicU64, AtomicUsize};
use std::time::Duration;

pub trait Pool: Send + Sync {
    /// id
    fn get_id(&self) -> &str;

    /// 设置id
    fn set_id(&mut self, id: &str);
    
    /// 设置最大连接数
    fn set_max_conn(&self, max: Option<usize>);

    /// 获取最大连接数
    fn get_max_conn(&self) -> Option<usize>;

    /// 获取当前的连接数
    fn get_cur_conn(&self) -> usize;

    /// 设置空闲连接保留时长
    fn set_keepalive(&self, _: Option<Duration>) {}

    /// 获取空闲连接保留时长
    fn get_keepalive(&self) -> Option<Duration> {
        None
    }

    /// 获取转发策略
    fn get_strategy(&self) -> Arc<dyn Strategy>;

    /// 设置策略
    fn set_strategy(&mut self, strategy: Arc<dyn Strategy>);
    
    /// 添加一个后端地址
    fn add_backend(&self, id: Option<u32>, addr: Address) {
        self.get_strategy().add_backend(id, addr)
    }

    /// 根据策略获取一个后端
    fn get_backend(&self, key: &str) -> Option<BackendState> {
        self.get_strategy().get_backend(key)
    }

    /// 移除一个后端地址
    fn remove_backend(&self, addr: &Address) -> bool {
        self.get_strategy().remove_backend(addr)
    }

    /// 获取所有后端地址切片
    fn get_backends(&self) -> Vec<BackendState> {
        self.get_strategy().get_backends()
    }

    fn get_backend_by_id(&self, id: u32) -> Option<BackendState> {
        self.get_strategy().get_backend_by_id(id)
    }

    fn get_backend_by_code(&self, code: u64) -> Option<BackendState> {
        self.get_strategy().get_backend_by_code(code)
    }

    // 是否使用tls
    fn use_tls(&self, _: bool) {}

    fn tls(&self) -> bool {
        false
    }
}

pub struct BaseState {
    /// id标识
    pub id: String,
    /// usize::MAX表示无设置
    pub max_conn: AtomicUsize, 
    pub cur_conn: AtomicUsize,
    /// 精度只到秒级别, u64::MAX表示无设置
    pub keepalive: AtomicU64, 
    pub lb_strategy: Arc<dyn Strategy>,
}

impl BaseState {
    pub fn new(strategy: Arc<dyn Strategy>) -> Self {
        BaseState {
            id: String::new(),
            max_conn: AtomicUsize::new(usize::MAX),
            cur_conn: AtomicUsize::new(0),
            keepalive: AtomicU64::new(u64::MAX),
            lb_strategy: strategy,
        }
    }
}

pub fn increase_current(max: &AtomicUsize, cur: &AtomicUsize) -> Result<(), Error> {
    let m = max.load(Relaxed);
    if m == usize::MAX {
        cur.fetch_add(1, Relaxed);
        Ok(())
    } else {
        let mut c = cur.load(Relaxed);
        loop {
            if c == m {
                break Err(Error::PoolFull);
            }

            match cur.compare_exchange(c, c + 1, Relaxed, Relaxed) {
                Ok(_) => break Ok(()),
                Err(s) => c = s,
            }
        }
    }
}