use bytes::{Bytes, BytesMut};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
#[derive(Clone)]
pub struct BytesPool {
pool: Arc<Mutex<HashMap<usize, Vec<BytesMut>>>>,
stats: Arc<Mutex<PoolStats>>,
}
impl Default for BytesPool {
fn default() -> Self {
Self::new()
}
}
impl BytesPool {
pub fn new() -> Self {
Self {
pool: Arc::new(Mutex::new(HashMap::new())),
stats: Arc::new(Mutex::new(PoolStats::default())),
}
}
pub fn get(&self, capacity: usize) -> BytesMut {
let bucket = Self::capacity_bucket(capacity);
let mut pool = self.pool.lock().unwrap_or_else(|e| e.into_inner());
let mut stats = self.stats.lock().unwrap_or_else(|e| e.into_inner());
if let Some(buffers) = pool.get_mut(&bucket) {
if let Some(mut buf) = buffers.pop() {
buf.clear();
stats.hits += 1;
return buf;
}
}
stats.misses += 1;
stats.allocations += 1;
BytesMut::with_capacity(bucket)
}
pub fn put(&self, mut buf: BytesMut) {
if buf.capacity() > 1024 * 1024 * 4 {
return;
}
buf.clear();
let bucket = Self::capacity_bucket(buf.capacity());
let mut pool = self.pool.lock().unwrap_or_else(|e| e.into_inner());
let buffers = pool.entry(bucket).or_default();
if buffers.len() < 100 {
buffers.push(buf);
}
}
pub fn stats(&self) -> PoolStats {
*self.stats.lock().unwrap_or_else(|e| e.into_inner())
}
pub fn clear(&self) {
self.pool.lock().unwrap_or_else(|e| e.into_inner()).clear();
}
fn capacity_bucket(capacity: usize) -> usize {
if capacity == 0 {
return 1024; }
capacity.next_power_of_two().max(1024)
}
}
#[derive(Clone)]
pub struct CidStringPool {
pool: Arc<Mutex<HashMap<String, Arc<str>>>>,
stats: Arc<Mutex<PoolStats>>,
}
impl Default for CidStringPool {
fn default() -> Self {
Self::new()
}
}
impl CidStringPool {
pub fn new() -> Self {
Self {
pool: Arc::new(Mutex::new(HashMap::new())),
stats: Arc::new(Mutex::new(PoolStats::default())),
}
}
pub fn intern(&self, s: &str) -> Arc<str> {
let mut pool = self.pool.lock().unwrap_or_else(|e| e.into_inner());
let mut stats = self.stats.lock().unwrap_or_else(|e| e.into_inner());
if let Some(existing) = pool.get(s) {
stats.hits += 1;
return Arc::clone(existing);
}
stats.misses += 1;
let arc: Arc<str> = Arc::from(s);
pool.insert(s.to_string(), Arc::clone(&arc));
arc
}
pub fn stats(&self) -> PoolStats {
*self.stats.lock().unwrap_or_else(|e| e.into_inner())
}
pub fn clear(&self) {
self.pool.lock().unwrap_or_else(|e| e.into_inner()).clear();
}
pub fn len(&self) -> usize {
self.pool.lock().unwrap_or_else(|e| e.into_inner()).len()
}
pub fn is_empty(&self) -> bool {
self.pool
.lock()
.unwrap_or_else(|e| e.into_inner())
.is_empty()
}
}
#[derive(Debug, Clone, Copy, Default)]
pub struct PoolStats {
pub hits: u64,
pub misses: u64,
pub allocations: u64,
}
impl PoolStats {
pub fn hit_rate(&self) -> f64 {
let total = self.hits + self.misses;
if total == 0 {
return 0.0;
}
self.hits as f64 / total as f64
}
pub fn miss_rate(&self) -> f64 {
1.0 - self.hit_rate()
}
}
static GLOBAL_BYTES_POOL: once_cell::sync::Lazy<BytesPool> =
once_cell::sync::Lazy::new(BytesPool::new);
static GLOBAL_CID_STRING_POOL: once_cell::sync::Lazy<CidStringPool> =
once_cell::sync::Lazy::new(CidStringPool::new);
pub fn global_bytes_pool() -> &'static BytesPool {
&GLOBAL_BYTES_POOL
}
pub fn global_cid_string_pool() -> &'static CidStringPool {
&GLOBAL_CID_STRING_POOL
}
pub fn freeze_bytes(buf: BytesMut) -> Bytes {
buf.freeze()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_bytes_pool_basic() {
let pool = BytesPool::new();
let buf1 = pool.get(1024);
assert!(buf1.capacity() >= 1024);
pool.put(buf1);
let buf2 = pool.get(1024);
assert!(buf2.capacity() >= 1024);
let stats = pool.stats();
assert_eq!(stats.hits, 1);
assert_eq!(stats.misses, 1);
}
#[test]
fn test_bytes_pool_capacity_bucketing() {
let pool = BytesPool::new();
let buf1 = pool.get(100);
let buf2 = pool.get(1000);
let buf3 = pool.get(2000);
assert!(buf1.capacity() >= 100);
assert!(buf2.capacity() >= 1000);
assert!(buf3.capacity() >= 2000);
pool.put(buf1);
pool.put(buf2);
pool.put(buf3);
let buf4 = pool.get(150); let buf5 = pool.get(1100);
assert!(buf4.capacity() >= 150);
assert!(buf5.capacity() >= 1100);
}
#[test]
fn test_cid_string_pool_basic() {
let pool = CidStringPool::new();
let s1 = pool.intern("QmTest123");
let s2 = pool.intern("QmTest123");
assert_eq!(s1.as_ref(), s2.as_ref());
assert!(Arc::ptr_eq(&s1, &s2));
let stats = pool.stats();
assert_eq!(stats.hits, 1);
assert_eq!(stats.misses, 1);
}
#[test]
fn test_cid_string_pool_different_strings() {
let pool = CidStringPool::new();
let s1 = pool.intern("QmTest1");
let s2 = pool.intern("QmTest2");
assert_ne!(s1.as_ref(), s2.as_ref());
assert!(!Arc::ptr_eq(&s1, &s2));
let stats = pool.stats();
assert_eq!(stats.hits, 0);
assert_eq!(stats.misses, 2);
}
#[test]
fn test_pool_stats_hit_rate() {
let stats = PoolStats {
hits: 80,
misses: 20,
allocations: 20,
};
assert!((stats.hit_rate() - 0.8).abs() < 0.001);
assert!((stats.miss_rate() - 0.2).abs() < 0.001);
}
#[test]
fn test_pool_stats_empty() {
let stats = PoolStats::default();
assert_eq!(stats.hit_rate(), 0.0);
assert_eq!(stats.miss_rate(), 1.0);
}
#[test]
fn test_bytes_pool_clear() {
let pool = BytesPool::new();
let buf = pool.get(1024);
pool.put(buf);
pool.clear();
let _buf2 = pool.get(1024);
let stats = pool.stats();
assert_eq!(stats.misses, 2); }
#[test]
fn test_cid_string_pool_len() {
let pool = CidStringPool::new();
assert_eq!(pool.len(), 0);
assert!(pool.is_empty());
pool.intern("QmTest1");
assert_eq!(pool.len(), 1);
assert!(!pool.is_empty());
pool.intern("QmTest2");
assert_eq!(pool.len(), 2);
pool.intern("QmTest1"); assert_eq!(pool.len(), 2); }
#[test]
fn test_bytes_pool_size_limit() {
let pool = BytesPool::new();
let large_buf = BytesMut::with_capacity(10 * 1024 * 1024);
pool.put(large_buf);
let _buf = pool.get(10 * 1024 * 1024);
let stats = pool.stats();
assert!(stats.misses >= 1);
}
#[test]
fn test_global_pools() {
let bytes_pool = global_bytes_pool();
let cid_pool = global_cid_string_pool();
let _buf = bytes_pool.get(1024);
let _s = cid_pool.intern("QmTest");
}
#[test]
fn test_freeze_bytes() {
let mut buf = BytesMut::with_capacity(1024);
buf.extend_from_slice(b"Hello, world!");
let bytes = freeze_bytes(buf);
assert_eq!(&bytes[..], b"Hello, world!");
}
}