565 lines
19 KiB
Rust
565 lines
19 KiB
Rust
use announcement::Announcement;
|
|
use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
|
|
use std::hint::black_box;
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
// ============================================================================
|
|
// SHUTDOWN SIGNAL PATTERN - Notify all workers to shut down
|
|
// ============================================================================
|
|
|
|
fn bench_shutdown_signal(c: &mut Criterion) {
|
|
let mut group = c.benchmark_group("pattern/shutdown_signal");
|
|
|
|
// Simulate shutdown signal to various numbers of workers
|
|
for num_workers in [10, 100, 1_000, 10_000].iter() {
|
|
group.throughput(Throughput::Elements(*num_workers as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_workers),
|
|
num_workers,
|
|
|b, &n| {
|
|
b.iter(|| {
|
|
let (shutdown_tx, shutdown_announcement) = Announcement::new();
|
|
|
|
// Create N worker listeners
|
|
let mut workers = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
workers.push(shutdown_announcement.listener());
|
|
}
|
|
|
|
// Send shutdown signal
|
|
shutdown_tx.announce(Arc::new(())).unwrap();
|
|
|
|
// All workers check for shutdown
|
|
for worker in &workers {
|
|
black_box(worker.try_listen());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
fn bench_shutdown_with_tokio_tasks(c: &mut Criterion) {
|
|
let rt = tokio::runtime::Runtime::new().unwrap();
|
|
let mut group = c.benchmark_group("pattern/shutdown_async");
|
|
group.sample_size(10);
|
|
group.measurement_time(Duration::from_secs(10));
|
|
|
|
for num_workers in [10, 100].iter() {
|
|
group.throughput(Throughput::Elements(*num_workers as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_workers),
|
|
num_workers,
|
|
|b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (shutdown_tx, shutdown_announcement) = Announcement::new();
|
|
|
|
// Spawn N worker tasks
|
|
let mut handles = Vec::with_capacity(n);
|
|
for i in 0..n {
|
|
let listener = shutdown_announcement.listener();
|
|
handles.push(tokio::spawn(async move {
|
|
// Wait for shutdown
|
|
listener.listen().await.unwrap();
|
|
i
|
|
}));
|
|
}
|
|
|
|
// Small delay to let tasks start
|
|
tokio::time::sleep(Duration::from_millis(1)).await;
|
|
|
|
// Send shutdown signal
|
|
shutdown_tx.announce(Arc::new(())).unwrap();
|
|
|
|
// Wait for all workers to complete
|
|
for handle in handles {
|
|
black_box(handle.await.unwrap());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
// ============================================================================
|
|
// CONFIG BROADCAST PATTERN - Share configuration to all services
|
|
// ============================================================================
|
|
|
|
#[derive(Clone, Debug)]
|
|
#[allow(dead_code)]
|
|
struct Config {
|
|
database_url: String,
|
|
api_key: String,
|
|
max_connections: usize,
|
|
timeout_ms: u64,
|
|
}
|
|
|
|
fn bench_config_broadcast(c: &mut Criterion) {
|
|
let mut group = c.benchmark_group("pattern/config_broadcast");
|
|
|
|
for num_services in [10, 100].iter() {
|
|
group.throughput(Throughput::Elements(*num_services as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_services),
|
|
num_services,
|
|
|b, &n| {
|
|
b.iter(|| {
|
|
let (config_tx, config_announcement) = Announcement::new();
|
|
|
|
// Create N service listeners
|
|
let mut services = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
services.push(config_announcement.listener());
|
|
}
|
|
|
|
// Broadcast configuration
|
|
let config = Arc::new(Config {
|
|
database_url: "postgres://localhost/db".to_string(),
|
|
api_key: "secret-key-12345".to_string(),
|
|
max_connections: 100,
|
|
timeout_ms: 5000,
|
|
});
|
|
|
|
config_tx.announce(config).unwrap();
|
|
|
|
// All services receive config
|
|
for service in &services {
|
|
black_box(service.try_listen());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
fn bench_config_broadcast_async(c: &mut Criterion) {
|
|
let rt = tokio::runtime::Runtime::new().unwrap();
|
|
let mut group = c.benchmark_group("pattern/config_broadcast_async");
|
|
group.sample_size(10);
|
|
group.measurement_time(Duration::from_secs(10));
|
|
|
|
for num_services in [10, 100].iter() {
|
|
group.throughput(Throughput::Elements(*num_services as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_services),
|
|
num_services,
|
|
|b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (config_tx, config_announcement) = Announcement::new();
|
|
|
|
// Spawn N service tasks waiting for config
|
|
let mut handles = Vec::with_capacity(n);
|
|
for i in 0..n {
|
|
let listener = config_announcement.listener();
|
|
handles.push(tokio::spawn(async move {
|
|
// Wait for config
|
|
let config: Arc<Config> = listener.listen().await.unwrap();
|
|
// Use config
|
|
(i, config.max_connections)
|
|
}));
|
|
}
|
|
|
|
// Small delay to let tasks start
|
|
tokio::time::sleep(Duration::from_millis(1)).await;
|
|
|
|
// Broadcast configuration
|
|
let config = Arc::new(Config {
|
|
database_url: "postgres://localhost/db".to_string(),
|
|
api_key: "secret-key-12345".to_string(),
|
|
max_connections: 100,
|
|
timeout_ms: 5000,
|
|
});
|
|
|
|
config_tx.announce(config).unwrap();
|
|
|
|
// Wait for all services to process config
|
|
for handle in handles {
|
|
black_box(handle.await.unwrap());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
// ============================================================================
|
|
// EVENT FANOUT PATTERN - Broadcast events to many subscribers
|
|
// ============================================================================
|
|
|
|
#[derive(Clone, Debug)]
|
|
#[allow(dead_code)]
|
|
struct Event {
|
|
event_type: String,
|
|
timestamp: u64,
|
|
data: Vec<u8>,
|
|
}
|
|
|
|
fn bench_event_fanout(c: &mut Criterion) {
|
|
let mut group = c.benchmark_group("pattern/event_fanout");
|
|
|
|
for num_subscribers in [10, 100, 1_000].iter() {
|
|
group.throughput(Throughput::Elements(*num_subscribers as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_subscribers),
|
|
num_subscribers,
|
|
|b, &n| {
|
|
b.iter(|| {
|
|
let (event_tx, event_announcement) = Announcement::new();
|
|
|
|
// Create N subscriber listeners
|
|
let mut subscribers = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
subscribers.push(event_announcement.listener());
|
|
}
|
|
|
|
// Broadcast event
|
|
let event = Arc::new(Event {
|
|
event_type: "user.created".to_string(),
|
|
timestamp: 1234567890,
|
|
data: vec![1, 2, 3, 4, 5],
|
|
});
|
|
|
|
event_tx.announce(event).unwrap();
|
|
|
|
// All subscribers receive event
|
|
for subscriber in &subscribers {
|
|
black_box(subscriber.try_listen());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
fn bench_event_fanout_async(c: &mut Criterion) {
|
|
let rt = tokio::runtime::Runtime::new().unwrap();
|
|
let mut group = c.benchmark_group("pattern/event_fanout_async");
|
|
group.sample_size(10);
|
|
group.measurement_time(Duration::from_secs(10));
|
|
|
|
for num_subscribers in [10, 100].iter() {
|
|
group.throughput(Throughput::Elements(*num_subscribers as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_subscribers),
|
|
num_subscribers,
|
|
|b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (event_tx, event_announcement) = Announcement::new();
|
|
|
|
// Spawn N subscriber tasks
|
|
let mut handles = Vec::with_capacity(n);
|
|
for i in 0..n {
|
|
let listener = event_announcement.listener();
|
|
handles.push(tokio::spawn(async move {
|
|
let event: Arc<Event> = listener.listen().await.unwrap();
|
|
(i, event.timestamp)
|
|
}));
|
|
}
|
|
|
|
// Small delay
|
|
tokio::time::sleep(Duration::from_millis(1)).await;
|
|
|
|
// Broadcast event
|
|
let event = Arc::new(Event {
|
|
event_type: "user.created".to_string(),
|
|
timestamp: 1234567890,
|
|
data: vec![1, 2, 3, 4, 5],
|
|
});
|
|
|
|
event_tx.announce(event).unwrap();
|
|
|
|
// Wait for all subscribers
|
|
for handle in handles {
|
|
black_box(handle.await.unwrap());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
// ============================================================================
|
|
// REQUEST-RESPONSE MULTIPLEXING - Multiple waiters for single response
|
|
// ============================================================================
|
|
|
|
#[derive(Clone, Debug)]
|
|
#[allow(dead_code)]
|
|
struct Response {
|
|
status: u16,
|
|
body: String,
|
|
}
|
|
|
|
fn bench_request_response_multiplex(c: &mut Criterion) {
|
|
let mut group = c.benchmark_group("pattern/request_response_multiplex");
|
|
|
|
// Multiple tasks waiting for same HTTP response
|
|
for num_waiters in [2, 5, 10, 50, 100].iter() {
|
|
group.throughput(Throughput::Elements(*num_waiters as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_waiters),
|
|
num_waiters,
|
|
|b, &n| {
|
|
b.iter(|| {
|
|
let (response_tx, response_announcement) = Announcement::new();
|
|
|
|
// Create N waiters for the same response
|
|
let mut waiters = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
waiters.push(response_announcement.listener());
|
|
}
|
|
|
|
// Simulate response arrives
|
|
let response = Arc::new(Response {
|
|
status: 200,
|
|
body: "OK".to_string(),
|
|
});
|
|
|
|
response_tx.announce(response).unwrap();
|
|
|
|
// All waiters get response
|
|
for waiter in &waiters {
|
|
black_box(waiter.try_listen());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
fn bench_request_response_multiplex_async(c: &mut Criterion) {
|
|
let rt = tokio::runtime::Runtime::new().unwrap();
|
|
let mut group = c.benchmark_group("pattern/request_response_multiplex_async");
|
|
group.sample_size(10);
|
|
|
|
for num_waiters in [2, 5, 10, 50, 100].iter() {
|
|
group.throughput(Throughput::Elements(*num_waiters as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_waiters),
|
|
num_waiters,
|
|
|b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (response_tx, response_announcement) = Announcement::new();
|
|
|
|
// Spawn N tasks waiting for response
|
|
let mut handles = Vec::with_capacity(n);
|
|
for i in 0..n {
|
|
let listener = response_announcement.listener();
|
|
handles.push(tokio::spawn(async move {
|
|
let response: Arc<Response> = listener.listen().await.unwrap();
|
|
(i, response.status)
|
|
}));
|
|
}
|
|
|
|
// Response arrives
|
|
let response = Arc::new(Response {
|
|
status: 200,
|
|
body: "OK".to_string(),
|
|
});
|
|
|
|
response_tx.announce(response).unwrap();
|
|
|
|
// Wait for all waiters
|
|
for handle in handles {
|
|
black_box(handle.await.unwrap());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
// ============================================================================
|
|
// LAZY INITIALIZATION PATTERN - Signal when resource is ready
|
|
// ============================================================================
|
|
|
|
#[derive(Clone, Debug)]
|
|
#[allow(dead_code)]
|
|
struct Database {
|
|
connection_pool: Arc<Vec<String>>,
|
|
max_connections: usize,
|
|
}
|
|
|
|
fn bench_lazy_initialization(c: &mut Criterion) {
|
|
let mut group = c.benchmark_group("pattern/lazy_initialization");
|
|
|
|
for num_clients in [10, 100].iter() {
|
|
group.throughput(Throughput::Elements(*num_clients as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_clients),
|
|
num_clients,
|
|
|b, &n| {
|
|
b.iter(|| {
|
|
let (db_ready_tx, db_ready_announcement) = Announcement::new();
|
|
|
|
// N clients waiting for database to be ready
|
|
let mut clients = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
clients.push(db_ready_announcement.listener());
|
|
}
|
|
|
|
// Simulate database initialization
|
|
let db = Arc::new(Database {
|
|
connection_pool: Arc::new(vec![
|
|
"conn1".to_string(),
|
|
"conn2".to_string(),
|
|
"conn3".to_string(),
|
|
]),
|
|
max_connections: 10,
|
|
});
|
|
|
|
// Signal ready
|
|
db_ready_tx.announce(db).unwrap();
|
|
|
|
// All clients can now proceed
|
|
for client in &clients {
|
|
black_box(client.try_listen());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
fn bench_lazy_initialization_async(c: &mut Criterion) {
|
|
let rt = tokio::runtime::Runtime::new().unwrap();
|
|
let mut group = c.benchmark_group("pattern/lazy_initialization_async");
|
|
group.sample_size(10);
|
|
|
|
for num_clients in [10, 100].iter() {
|
|
group.throughput(Throughput::Elements(*num_clients as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_clients),
|
|
num_clients,
|
|
|b, &n| {
|
|
b.to_async(&rt).iter(|| async move {
|
|
let (db_ready_tx, db_ready_announcement) = Announcement::new();
|
|
|
|
// Spawn N client tasks
|
|
let mut handles = Vec::with_capacity(n);
|
|
for i in 0..n {
|
|
let listener = db_ready_announcement.listener();
|
|
handles.push(tokio::spawn(async move {
|
|
// Wait for database
|
|
let db: Arc<Database> = listener.listen().await.unwrap();
|
|
(i, db.max_connections)
|
|
}));
|
|
}
|
|
|
|
let db = Arc::new(Database {
|
|
connection_pool: Arc::new(vec![
|
|
"conn1".to_string(),
|
|
"conn2".to_string(),
|
|
"conn3".to_string(),
|
|
]),
|
|
max_connections: 10,
|
|
});
|
|
|
|
db_ready_tx.announce(db).unwrap();
|
|
|
|
// Wait for all clients
|
|
for handle in handles {
|
|
black_box(handle.await.unwrap());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
// ============================================================================
|
|
// BARRIER PATTERN - Multiple announcements that all must complete
|
|
// ============================================================================
|
|
|
|
fn bench_barrier_pattern(c: &mut Criterion) {
|
|
let mut group = c.benchmark_group("pattern/barrier");
|
|
|
|
// N tasks each announce completion, all wait for all to complete
|
|
for num_tasks in [5, 10, 25].iter() {
|
|
group.throughput(Throughput::Elements(*num_tasks as u64));
|
|
group.bench_with_input(
|
|
BenchmarkId::from_parameter(num_tasks),
|
|
num_tasks,
|
|
|b, &n| {
|
|
b.iter(|| {
|
|
let mut announcements = Vec::with_capacity(n);
|
|
let mut announcers = Vec::with_capacity(n);
|
|
|
|
// Create N announcement channels (one per task)
|
|
for _ in 0..n {
|
|
let (tx, announcement) = Announcement::<Arc<i32>>::new();
|
|
announcers.push(tx);
|
|
announcements.push(announcement);
|
|
}
|
|
|
|
// Each task listens to all other tasks
|
|
let mut all_listeners = Vec::with_capacity(n * n);
|
|
for announcement in &announcements {
|
|
for _ in 0..n {
|
|
all_listeners.push(announcement.listener());
|
|
}
|
|
}
|
|
|
|
// All tasks complete
|
|
for (i, announcer) in announcers.into_iter().enumerate() {
|
|
announcer.announce(Arc::new(i as i32)).unwrap();
|
|
}
|
|
|
|
// Verify all received
|
|
for listener in &all_listeners {
|
|
black_box(listener.try_listen());
|
|
}
|
|
});
|
|
},
|
|
);
|
|
}
|
|
|
|
group.finish();
|
|
}
|
|
|
|
criterion_group!(
|
|
shutdown,
|
|
bench_shutdown_signal,
|
|
bench_shutdown_with_tokio_tasks,
|
|
);
|
|
|
|
criterion_group!(config, bench_config_broadcast, bench_config_broadcast_async,);
|
|
|
|
criterion_group!(events, bench_event_fanout, bench_event_fanout_async,);
|
|
|
|
criterion_group!(
|
|
multiplex,
|
|
bench_request_response_multiplex,
|
|
bench_request_response_multiplex_async,
|
|
);
|
|
|
|
criterion_group!(
|
|
lazy_init,
|
|
bench_lazy_initialization,
|
|
bench_lazy_initialization_async,
|
|
);
|
|
|
|
criterion_group!(barrier, bench_barrier_pattern,);
|
|
|
|
criterion_main!(shutdown, config, events, multiplex, lazy_init, barrier);
|