use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex};
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant};
use std::sync::atomic::AtomicU64;
#[derive(Debug)]
pub struct NetworkBufferPool {
pools: Arc<Mutex<HashMap<usize, Vec<BufferEntry>>>>,
lru_tracker: Arc<Mutex<Vec<u64>>>,
config: BufferPoolConfig,
stats: Arc<Mutex<BufferPoolStats>>,
}
#[derive(Debug)]
struct BufferEntry {
id: u64,
data: Vec<u8>,
#[allow(dead_code)]
last_access: Instant,
#[allow(dead_code)]
ref_count: usize,
}
#[derive(Debug, Clone)]
pub struct BufferPoolConfig {
pub default_size: usize,
pub preallocate_count: usize,
pub max_pool_size: usize,
pub alignment: usize,
pub lru_threshold: Duration,
}
#[derive(Debug, Clone, Default)]
pub struct BufferPoolStats {
pub active_buffers: usize,
pub pooled_buffers: usize,
pub total_allocations: u64,
pub total_deallocations: u64,
pub pool_hits: u64,
pub pool_misses: u64,
pub hit_rate: f64,
pub avg_buffer_size: f64,
pub total_memory_usage: u64,
}
impl NetworkBufferPool {
pub fn new(config: BufferPoolConfig) -> Self {
let pool: _ = Self {
pools: Arc::new(Mutex::new(HashMap::new())),
lru_tracker: Arc::new(Mutex::new(Vec::new())),
config: config.clone(),
stats: Arc::new(Mutex::new(BufferPoolStats::default())),
};
pool.preallocate_buffers();
pool
}
pub fn default() -> Self {
let config: _ = BufferPoolConfig {
default_size: 64 * 1024, preallocate_count: 100, max_pool_size: 1000, alignment: 64, lru_threshold: Duration::from_secs(60), };
Self::new(config)
}
fn preallocate_buffers(&self) {
let mut pools = self.pools.lock().unwrap();
for _ in 0..self.config.preallocate_count {
let buffer: _ = self.create_buffer(self.config.default_size);
let size: _ = buffer.data.len();
if let Some(pool) = pools.get_mut(&size) {
pool.push(buffer);
} else {
pools.insert(size, vec![buffer]);
}
}
{
let mut stats = self.stats.lock().unwrap();
stats.pooled_buffers = self.config.preallocate_count;
stats.avg_buffer_size = self.config.default_size as f64;
}
}
fn create_buffer(&self, size: usize) -> BufferEntry {
let aligned_size: _ = self.align_size(size);
BufferEntry {
id: self.generate_id(),
data: vec![0u8; aligned_size],
last_access: Instant::now(),
ref_count: 0,
}
}
fn generate_id(&self) -> u64 {
static COUNTER: AtomicU64 = AtomicU64::new(1);
COUNTER.fetch_add(1, Ordering::SeqCst)
}
fn align_size(&self, size: usize) -> usize {
let alignment: _ = self.config.alignment;
let remainder: _ = size % alignment;
if remainder == 0 {
size
} else {
size + (alignment - remainder)
}
}
pub fn get_buffer(&self, size: usize) -> (Vec<u8>, usize) {
let aligned_size: _ = self.align_size(size);
{
let mut pools = self.pools.lock().unwrap();
if let Some(pool) = pools.get_mut(&aligned_size) {
if let Some(buffer) = pool.pop() {
{
let mut stats = self.stats.lock().unwrap();
stats.pool_hits += 1;
stats.active_buffers += 1;
stats.pooled_buffers = pool.len();
stats.hit_rate = stats.pool_hits as f64
/ (stats.pool_hits + stats.pool_misses) as f64;
}
self.update_lru(buffer.id);
return (buffer.data, aligned_size);
}
}
}
{
let mut stats = self.stats.lock().unwrap();
stats.pool_misses += 1;
stats.total_allocations += 1;
stats.active_buffers += 1;
stats.total_memory_usage += aligned_size as u64;
}
(vec![0u8; aligned_size], aligned_size)
}
pub fn return_buffer(&self, mut buffer: Vec<u8>, size: usize) {
let aligned_size: _ = self.align_size(size);
{
let stats: _ = self.stats.lock().unwrap();
if stats.pooled_buffers >= self.config.max_pool_size {
return;
}
}
buffer.clear();
let buffer_entry: _ = BufferEntry {
id: self.generate_id(),
data: buffer,
last_access: Instant::now(),
ref_count: 0,
};
{
let mut pools = self.pools.lock().unwrap();
if let Some(pool) = pools.get_mut(&aligned_size) {
pool.push(buffer_entry);
} else {
pools.insert(aligned_size, vec![buffer_entry]);
}
{
let mut stats = self.stats.lock().unwrap();
stats.pooled_buffers += 1;
stats.active_buffers -= 1;
stats.total_deallocations += 1;
}
}
self.cleanup_lru();
}
fn update_lru(&self, buffer_id: u64) {
let mut tracker = self.lru_tracker.lock().unwrap();
tracker.retain(|&id| id != buffer_id);
tracker.push(buffer_id);
}
fn cleanup_lru(&self) {
let mut pools = self.pools.lock().unwrap();
let mut tracker = self.lru_tracker.lock().unwrap();
while tracker.len() > self.config.max_pool_size {
if let Some(oldest_id) = tracker.first().cloned() {
for pool in pools.values_mut() {
pool.retain(|buffer| buffer.id != oldest_id);
}
tracker.remove(0);
}
}
let _now: _ = Instant::now();
tracker.retain(|&_id| {
true
});
}
pub fn get_stats(&self) -> BufferPoolStats {
self.stats.lock().unwrap().clone()
}
pub fn reset_stats(&self) {
let mut stats = self.stats.lock().unwrap();
*stats = BufferPoolStats::default();
}
pub fn warmup(&self, size: usize, count: usize) {
let aligned_size: _ = self.align_size(size);
let mut pools = self.pools.lock().unwrap();
for _ in 0..count {
let buffer: _ = self.create_buffer(aligned_size);
if let Some(pool) = pools.get_mut(&aligned_size) {
pool.push(buffer);
} else {
pools.insert(aligned_size, vec![buffer]);
}
}
{
let mut stats = self.stats.lock().unwrap();
stats.pooled_buffers += count;
stats.avg_buffer_size = (stats.avg_buffer_size + size as f64) / 2.0;
}
}
pub fn clear(&self) {
let mut pools = self.pools.lock().unwrap();
pools.clear();
let mut tracker = self.lru_tracker.lock().unwrap();
tracker.clear();
let mut stats = self.stats.lock().unwrap();
stats.pooled_buffers = 0;
stats.total_memory_usage = 0;
}
}
#[cfg(test)]
mod tests {
#[test]
fn test_zero_copy_network_buffer_pool_performance() {
let _pool: _ = NetworkBufferPool::default();
println!("NetworkBufferPool performance test placeholder");
println!("This test validates buffer pool allocation and deallocation performance");
}
}