use std::collections::hash_map::Entry;
use std::collections::{HashMap, VecDeque};
use std::io::{self, Read};
use crate::stream::Stream;
use crate::unit::Unit;
use crate::Proxy;
use url::Url;
const DEFAULT_MAX_IDLE_CONNECTIONS: usize = 100;
const DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST: usize = 1;
#[derive(Debug)]
pub(crate) struct ConnectionPool {
recycle: HashMap<PoolKey, VecDeque<Stream>>,
lru: VecDeque<PoolKey>,
max_idle_connections: usize,
max_idle_connections_per_host: usize,
}
fn remove_first_match(list: &mut VecDeque<PoolKey>, key: &PoolKey) -> Option<PoolKey> {
match list.iter().position(|x| x == key) {
Some(i) => list.remove(i),
None => None,
}
}
fn remove_last_match(list: &mut VecDeque<PoolKey>, key: &PoolKey) -> Option<PoolKey> {
match list.iter().rposition(|x| x == key) {
Some(i) => list.remove(i),
None => None,
}
}
impl Default for ConnectionPool {
fn default() -> Self {
Self {
max_idle_connections: DEFAULT_MAX_IDLE_CONNECTIONS,
max_idle_connections_per_host: DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST,
recycle: HashMap::default(),
lru: VecDeque::default(),
}
}
}
impl ConnectionPool {
pub fn set_max_idle_connections(&mut self, max_connections: usize) {
if self.max_idle_connections == max_connections {
return;
}
while self.lru.len() > max_connections {
self.remove_oldest();
}
self.max_idle_connections = max_connections;
}
fn noop(&self) -> bool {
self.max_idle_connections == 0 || self.max_idle_connections_per_host == 0
}
pub fn set_max_idle_connections_per_host(&mut self, max_connections: usize) {
if self.max_idle_connections_per_host == max_connections {
return;
}
if max_connections == 0 {
self.lru.clear();
self.recycle.clear();
return;
}
for (key, val) in self.recycle.iter_mut() {
while val.len() > max_connections {
val.pop_front();
remove_first_match(&mut self.lru, key)
.expect("invariant failed: key in recycle but not in lru");
}
}
self.max_idle_connections_per_host = max_connections;
}
pub fn try_get_connection(&mut self, url: &Url, proxy: &Option<Proxy>) -> Option<Stream> {
let key = PoolKey::new(url, proxy);
self.remove(&key)
}
fn remove(&mut self, key: &PoolKey) -> Option<Stream> {
match self.recycle.entry(key.clone()) {
Entry::Occupied(mut occupied_entry) => {
let streams = occupied_entry.get_mut();
let stream = streams.pop_back();
let stream = stream.expect("invariant failed: empty VecDeque in `recycle`");
if streams.len() == 0 {
occupied_entry.remove();
}
remove_last_match(&mut self.lru, &key)
.expect("invariant failed: key in recycle but not in lru");
Some(stream)
}
Entry::Vacant(_) => None,
}
}
fn add(&mut self, key: PoolKey, stream: Stream) {
if self.noop() {
return;
}
match self.recycle.entry(key.clone()) {
Entry::Occupied(mut occupied_entry) => {
let streams = occupied_entry.get_mut();
streams.push_back(stream);
if streams.len() > self.max_idle_connections_per_host {
streams.pop_front();
remove_first_match(&mut self.lru, &key)
.expect("invariant failed: key in recycle but not in lru");
}
}
Entry::Vacant(vacant_entry) => {
vacant_entry.insert(vec![stream].into());
}
}
self.lru.push_back(key);
if self.lru.len() > self.max_idle_connections {
self.remove_oldest()
}
}
fn remove_oldest(&mut self) {
assert!(!self.noop(), "remove_oldest called on Pool with max of 0");
let key = self.lru.pop_front();
let key = key.expect("tried to remove oldest but no entries found!");
match self.recycle.entry(key) {
Entry::Occupied(mut occupied_entry) => {
let streams = occupied_entry.get_mut();
streams
.pop_front()
.expect("invariant failed: key existed in recycle but no streams available");
if streams.len() == 0 {
occupied_entry.remove();
}
}
Entry::Vacant(_) => panic!("invariant failed: key existed in lru but not in recycle"),
}
}
#[cfg(test)]
pub fn len(&self) -> usize {
self.lru.len()
}
}
#[derive(PartialEq, Clone, Eq, Hash)]
struct PoolKey {
scheme: String,
hostname: String,
port: Option<u16>,
proxy: Option<Proxy>,
}
use std::fmt;
impl fmt::Debug for PoolKey {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_fmt(format_args!(
"{}|{}|{}",
self.scheme,
self.hostname,
self.port.unwrap_or(0)
))
}
}
impl PoolKey {
fn new(url: &Url, proxy: &Option<Proxy>) -> Self {
let port = url.port_or_known_default();
PoolKey {
scheme: url.scheme().to_string(),
hostname: url.host_str().unwrap_or("").to_string(),
port,
proxy: proxy.clone(),
}
}
}
#[test]
fn poolkey_new() {
PoolKey::new(&Url::parse("zzz:///example.com").unwrap(), &None);
}
#[test]
fn pool_connections_limit() {
let mut pool = ConnectionPool::default();
let hostnames = (0..DEFAULT_MAX_IDLE_CONNECTIONS * 2).map(|i| format!("{}.example", i));
let poolkeys = hostnames.map(|hostname| PoolKey {
scheme: "https".to_string(),
hostname,
port: Some(999),
proxy: None,
});
for key in poolkeys.clone() {
pool.add(key, Stream::Cursor(std::io::Cursor::new(vec![])));
}
assert_eq!(pool.len(), DEFAULT_MAX_IDLE_CONNECTIONS);
for key in poolkeys.skip(DEFAULT_MAX_IDLE_CONNECTIONS) {
let result = pool.remove(&key);
assert!(result.is_some(), "expected key was not in pool");
}
assert_eq!(pool.len(), 0)
}
#[test]
fn pool_per_host_connections_limit() {
let mut pool = ConnectionPool::default();
let poolkey = PoolKey {
scheme: "https".to_string(),
hostname: "example.com".to_string(),
port: Some(999),
proxy: None,
};
for _ in 0..pool.max_idle_connections_per_host * 2 {
pool.add(
poolkey.clone(),
Stream::Cursor(std::io::Cursor::new(vec![])),
);
}
assert_eq!(pool.len(), DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST);
for _ in 0..DEFAULT_MAX_IDLE_CONNECTIONS_PER_HOST {
let result = pool.remove(&poolkey);
assert!(result.is_some(), "expected key was not in pool");
}
assert_eq!(pool.len(), 0);
}
#[test]
fn pool_update_connection_limit() {
let mut pool = ConnectionPool::default();
pool.set_max_idle_connections(50);
let hostnames = (0..pool.max_idle_connections).map(|i| format!("{}.example", i));
let poolkeys = hostnames.map(|hostname| PoolKey {
scheme: "https".to_string(),
hostname,
port: Some(999),
proxy: None,
});
for key in poolkeys.clone() {
pool.add(key, Stream::Cursor(std::io::Cursor::new(vec![])));
}
assert_eq!(pool.len(), 50);
pool.set_max_idle_connections(25);
assert_eq!(pool.len(), 25);
}
#[test]
fn pool_update_per_host_connection_limit() {
let mut pool = ConnectionPool::default();
pool.set_max_idle_connections(50);
pool.set_max_idle_connections_per_host(50);
let poolkey = PoolKey {
scheme: "https".to_string(),
hostname: "example.com".to_string(),
port: Some(999),
proxy: None,
};
for _ in 0..50 {
pool.add(
poolkey.clone(),
Stream::Cursor(std::io::Cursor::new(vec![])),
);
}
assert_eq!(pool.len(), 50);
pool.set_max_idle_connections_per_host(25);
assert_eq!(pool.len(), 25);
}
#[test]
fn pool_checks_proxy() {
let mut pool = ConnectionPool::default();
let url = Url::parse("zzz:///example.com").unwrap();
pool.add(
PoolKey::new(&url, &None),
Stream::Cursor(std::io::Cursor::new(vec![])),
);
assert_eq!(pool.len(), 1);
pool.add(
PoolKey::new(&url, &Some(Proxy::new("localhost:9999").unwrap())),
Stream::Cursor(std::io::Cursor::new(vec![])),
);
assert_eq!(pool.len(), 2);
pool.add(
PoolKey::new(
&url,
&Some(Proxy::new("user:password@localhost:9999").unwrap()),
),
Stream::Cursor(std::io::Cursor::new(vec![])),
);
assert_eq!(pool.len(), 3);
}
pub(crate) struct PoolReturnRead<R: Read + Sized + Into<Stream>> {
unit: Option<Unit>,
reader: Option<R>,
}
impl<R: Read + Sized + Into<Stream>> PoolReturnRead<R> {
pub fn new(unit: Option<Unit>, reader: R) -> Self {
PoolReturnRead {
unit,
reader: Some(reader),
}
}
fn return_connection(&mut self) -> io::Result<()> {
if let (Some(unit), Some(reader)) = (self.unit.take(), self.reader.take()) {
let state = &mut unit.req.agent.lock().unwrap();
let mut stream = reader.into();
if !stream.is_poolable() {
return Ok(());
}
stream.reset()?;
let key = PoolKey::new(&unit.url, &unit.req.proxy);
state.pool().add(key, stream);
}
Ok(())
}
fn do_read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
match self.reader.as_mut() {
None => Ok(0),
Some(reader) => reader.read(buf),
}
}
}
impl<R: Read + Sized + Into<Stream>> Read for PoolReturnRead<R> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let amount = self.do_read(buf)?;
if amount == 0 {
self.return_connection()?;
}
Ok(amount)
}
}