372 lines
14 KiB
Rust
372 lines
14 KiB
Rust
//! ============================================================================
|
|
//! LEARNING GUIDE - PART 5: Blocking Operations
|
|
//! ============================================================================
|
|
//!
|
|
//! WHAT YOU'LL LEARN:
|
|
//! - How to use blocking operations in non-async code
|
|
//! - When to use listen_blocking() vs listen().await
|
|
//! - How to use timeouts with listen_timeout_blocking()
|
|
//! - Understanding deadlock risks and how to avoid them
|
|
//!
|
|
//! PREREQUISITES:
|
|
//! - Read examples 01-04 first
|
|
//! - Understand async vs blocking operations
|
|
//! - Understand threads vs async tasks
|
|
//!
|
|
//! KEY CONCEPTS:
|
|
//! - listen_blocking(): Blocks the THREAD (not executor)
|
|
//! - listen_timeout_blocking(): Returns error if timeout expires
|
|
//! - Deadlock risk: Dropping announcer without announcing
|
|
//! - Feature gates: Requires "std" feature (default)
|
|
//!
|
|
//! Run with: cargo run --example 05_blocking_operations
|
|
|
|
#[cfg(not(feature = "std"))]
|
|
fn main() {
|
|
eprintln!("ERROR: This example requires the 'std' feature.");
|
|
eprintln!("Run with: cargo run --example 05_blocking_operations");
|
|
std::process::exit(1);
|
|
}
|
|
|
|
#[cfg(feature = "std")]
|
|
use announcement::Announcement;
|
|
#[cfg(feature = "std")]
|
|
use std::thread;
|
|
#[cfg(feature = "std")]
|
|
use std::time::Duration;
|
|
|
|
#[cfg(feature = "std")]
|
|
fn main() {
|
|
println!("=== Blocking Operations Demo ===\n");
|
|
|
|
// INTRODUCTION: Async vs Blocking
|
|
// ================================
|
|
println!("Introduction: Understanding blocking operations");
|
|
println!("\n Async operations (listen().await):");
|
|
println!(" - Suspend the TASK (not the thread)");
|
|
println!(" - Allow other tasks to run on the same thread");
|
|
println!(" - Require async runtime (tokio, async-std, etc.)");
|
|
println!(" - Best for I/O-bound work");
|
|
|
|
println!("\n Blocking operations (listen_blocking()):");
|
|
println!(" - Block the entire THREAD");
|
|
println!(" - Thread sits idle waiting");
|
|
println!(" - No async runtime needed");
|
|
println!(" - Use in non-async code or dedicated threads");
|
|
|
|
println!("\n IMPORTANT: Don't use blocking operations in async tasks!");
|
|
println!(" They will block the executor thread and cause poor performance.\n");
|
|
|
|
// STEP 1: Basic blocking example
|
|
// ===============================
|
|
println!("Step 1: Basic blocking example with thread:");
|
|
|
|
let (announcer, announcement) = Announcement::<String>::new();
|
|
|
|
// Create a listener and move it to a separate thread
|
|
let listener = announcement.listener();
|
|
|
|
println!(" Spawning thread that will block...");
|
|
let handle = thread::spawn(move || {
|
|
println!(" [Thread] Started, blocking until value arrives...");
|
|
|
|
// IMPORTANT: This line BLOCKS the thread!
|
|
// ========================================
|
|
// The thread will sit here, waiting, consuming no CPU.
|
|
// When the value is announced, the thread wakes up.
|
|
//
|
|
// WARNING: If the announcer is dropped without announcing,
|
|
// this will block FOREVER (deadlock).
|
|
let value = listener.listen_blocking().unwrap();
|
|
|
|
println!(" [Thread] Received: {}", value);
|
|
value
|
|
});
|
|
|
|
println!(" Main thread: Sleeping for 100ms before announcing...");
|
|
thread::sleep(Duration::from_millis(100));
|
|
|
|
println!(" Main thread: Announcing value...");
|
|
announcer
|
|
.announce("Hello from blocking!".to_string())
|
|
.unwrap();
|
|
|
|
println!(" Main thread: Waiting for thread to complete...");
|
|
let result = handle.join().unwrap();
|
|
println!(" Thread result: {}\n", result);
|
|
|
|
// STEP 2: Blocking with timeout (SUCCESS case)
|
|
// =============================================
|
|
println!("Step 2: Blocking with timeout (SUCCESS case):");
|
|
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
println!(" Spawning thread with 1-second timeout...");
|
|
let handle = thread::spawn(move || {
|
|
println!(" [Thread] Waiting with 1-second timeout...");
|
|
|
|
// This will wait UP TO 1 second
|
|
// Returns Result<T, ListenError>
|
|
match listener.listen_timeout_blocking(Duration::from_secs(1)) {
|
|
Ok(value) => {
|
|
println!(" [Thread] Received: {}", value);
|
|
Some(value)
|
|
}
|
|
Err(err) => {
|
|
println!(" [Thread] Timeout! Error: {}", err);
|
|
None
|
|
}
|
|
}
|
|
});
|
|
|
|
// Announce quickly (within timeout)
|
|
println!(" Main thread: Sleeping 100ms...");
|
|
thread::sleep(Duration::from_millis(100));
|
|
|
|
println!(" Main thread: Announcing before timeout...");
|
|
announcer.announce(42).unwrap();
|
|
|
|
let result = handle.join().unwrap();
|
|
println!(" Thread received: {:?}", result);
|
|
println!(" ✓ SUCCESS - value received within timeout\n");
|
|
|
|
// STEP 3: Blocking with timeout (TIMEOUT case)
|
|
// =============================================
|
|
println!("Step 3: Blocking with timeout (TIMEOUT case):");
|
|
|
|
let (_announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
// NOTE: We DON'T announce - so the thread will timeout
|
|
println!(" Spawning thread with 100ms timeout...");
|
|
println!(" (We won't announce, so it will timeout)");
|
|
|
|
let handle = thread::spawn(move || {
|
|
println!(" [Thread] Waiting with 100ms timeout...");
|
|
|
|
match listener.listen_timeout_blocking(Duration::from_millis(100)) {
|
|
Ok(value) => {
|
|
println!(" [Thread] Received: {}", value);
|
|
Some(value)
|
|
}
|
|
Err(err) => {
|
|
println!(" [Thread] Timeout! Error: {}", err);
|
|
None
|
|
}
|
|
}
|
|
});
|
|
|
|
let result = handle.join().unwrap();
|
|
println!(" Thread result: {:?}", result);
|
|
println!(" ✓ TIMEOUT - no value within 100ms, error returned\n");
|
|
|
|
// STEP 4: Multiple blocking threads
|
|
// ==================================
|
|
println!("Step 4: Multiple threads all blocking:");
|
|
|
|
let (announcer, announcement) = Announcement::<String>::new();
|
|
|
|
let handles: Vec<_> = (0..5)
|
|
.map(|i| {
|
|
let listener = announcement.listener();
|
|
thread::spawn(move || {
|
|
println!(" [Thread {}] Blocking...", i);
|
|
let value = listener.listen_blocking().unwrap();
|
|
println!(" [Thread {}] Received: {}", i, value);
|
|
value
|
|
})
|
|
})
|
|
.collect();
|
|
|
|
println!(" Spawned 5 threads, all blocking");
|
|
println!(" Main thread: Sleeping 200ms...");
|
|
thread::sleep(Duration::from_millis(200));
|
|
|
|
println!(" Main thread: Announcing - all threads will wake!");
|
|
announcer
|
|
.announce("Broadcast to all threads!".to_string())
|
|
.unwrap();
|
|
|
|
println!(" Waiting for all threads...");
|
|
for (i, handle) in handles.into_iter().enumerate() {
|
|
let result = handle.join().unwrap();
|
|
println!(" Thread {} returned: {}", i, result);
|
|
}
|
|
println!(" ✓ All 5 threads received the value\n");
|
|
|
|
// STEP 5: DEADLOCK WARNING
|
|
// ========================
|
|
println!("Step 5: UNDERSTANDING DEADLOCK RISK:");
|
|
println!("\n DANGEROUS CODE (commented out to avoid hanging):");
|
|
println!(" ```");
|
|
println!(" let (announcer, announcement) = Announcement::<i32>::new();");
|
|
println!(" let listener = announcement.listener();");
|
|
println!(" ");
|
|
println!(" thread::spawn(move || {{");
|
|
println!(" listener.listen_blocking(); // BLOCKS FOREVER!");
|
|
println!(" }});");
|
|
println!(" ");
|
|
println!(" drop(announcer); // Dropped without announcing!");
|
|
println!(" // Thread is now stuck forever!");
|
|
println!(" ```");
|
|
|
|
println!("\n WHY THIS DEADLOCKS:");
|
|
println!(" 1. Thread calls listen_blocking()");
|
|
println!(" 2. Thread waits for announcement");
|
|
println!(" 3. Announcer is dropped WITHOUT announcing");
|
|
println!(" 4. No announcement ever comes");
|
|
println!(" 5. Thread waits FOREVER");
|
|
|
|
println!("\n HOW TO AVOID DEADLOCK:");
|
|
println!(" ✓ Always announce before dropping announcer");
|
|
println!(" ✓ OR use listen_timeout_blocking() with timeout");
|
|
println!(" ✓ OR use try_listen() in a loop with sleep");
|
|
println!(" ✓ OR use async listen().await (can be cancelled)\n");
|
|
|
|
// STEP 6: Safe pattern with timeout
|
|
// ==================================
|
|
println!("Step 6: Safe pattern with timeout:");
|
|
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
let handle = thread::spawn(move || {
|
|
println!(" [Thread] Using timeout for safety...");
|
|
|
|
// SAFE: Even if announcer is dropped, we'll timeout
|
|
match listener.listen_timeout_blocking(Duration::from_secs(2)) {
|
|
Ok(value) => {
|
|
println!(" [Thread] Received: {}", value);
|
|
Ok(value)
|
|
}
|
|
Err(err) => {
|
|
println!(" [Thread] Timed out: {}", err);
|
|
Err(err)
|
|
}
|
|
}
|
|
});
|
|
|
|
// Simulate: Announcer might be dropped here (e.g., error path)
|
|
// But thread will timeout instead of hanging forever
|
|
drop(announcer);
|
|
println!(" Dropped announcer WITHOUT announcing");
|
|
|
|
let result = handle.join().unwrap();
|
|
println!(" Thread result: {:?}", result);
|
|
println!(" ✓ Thread didn't hang - timeout saved us!\n");
|
|
|
|
// STEP 7: Alternative safe pattern - polling
|
|
// ===========================================
|
|
println!("Step 7: Alternative safe pattern - polling with try_listen():");
|
|
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
|
|
let handle = thread::spawn(move || {
|
|
println!(" [Thread] Polling with try_listen()...");
|
|
|
|
// Poll every 50ms, timeout after 1 second
|
|
let start = std::time::Instant::now();
|
|
let timeout = Duration::from_secs(1);
|
|
|
|
loop {
|
|
// Non-blocking check
|
|
if let Some(value) = listener.try_listen() {
|
|
println!(" [Thread] Got value: {}", value);
|
|
return Ok(value);
|
|
}
|
|
|
|
// Check timeout
|
|
if start.elapsed() > timeout {
|
|
println!(" [Thread] Polling timeout");
|
|
return Err("timeout");
|
|
}
|
|
|
|
// Sleep before next poll
|
|
thread::sleep(Duration::from_millis(50));
|
|
}
|
|
});
|
|
|
|
thread::sleep(Duration::from_millis(200));
|
|
announcer.announce(99).unwrap();
|
|
|
|
let result = handle.join().unwrap();
|
|
println!(" Thread result: {:?}", result);
|
|
println!(" ✓ Polling pattern is also safe from deadlock\n");
|
|
|
|
// STEP 8: When to use each method
|
|
// ================================
|
|
println!("Step 8: Decision guide - Which method to use:");
|
|
|
|
println!("\n Use try_listen() when:");
|
|
println!(" ✓ You don't want to wait");
|
|
println!(" ✓ Polling in a loop");
|
|
println!(" ✓ Checking if value is ready");
|
|
println!(" ✓ Non-blocking required");
|
|
|
|
println!("\n Use listen().await when:");
|
|
println!(" ✓ In async code");
|
|
println!(" ✓ Can wait efficiently");
|
|
println!(" ✓ Need cancellation support");
|
|
println!(" ✓ Using async runtime");
|
|
|
|
println!("\n Use listen_blocking() when:");
|
|
println!(" ✓ In non-async code");
|
|
println!(" ✓ Dedicated thread can block");
|
|
println!(" ✓ You control the announcer lifetime");
|
|
println!(" ✓ No async runtime available");
|
|
|
|
println!("\n Use listen_timeout_blocking() when:");
|
|
println!(" ✓ listen_blocking() but need timeout");
|
|
println!(" ✓ Can't guarantee announcement will happen");
|
|
println!(" ✓ Want to prevent deadlock");
|
|
println!(" ✓ Need to handle timeout case\n");
|
|
|
|
// STEP 9: Performance characteristics
|
|
// ====================================
|
|
println!("Step 9: Performance characteristics:");
|
|
println!("\n listen_blocking():");
|
|
println!(" - Fast path (already announced): ~5-10ns");
|
|
println!(" - Slow path (waiting): Thread sleeps, 0% CPU");
|
|
println!(" - Wakeup: ~100ns when announced");
|
|
|
|
println!("\n listen_timeout_blocking():");
|
|
println!(" - Same as listen_blocking()");
|
|
println!(" - Plus: Timeout check overhead (~50ns)");
|
|
println!(" - Plus: Extra wakeup for timeout");
|
|
|
|
println!("\n try_listen() polling:");
|
|
println!(" - Each poll: ~5-10ns");
|
|
println!(" - CPU usage: Depends on poll frequency");
|
|
println!(" - Example: Poll every 10ms = 0.0001% CPU\n");
|
|
|
|
// SUMMARY
|
|
// =======
|
|
println!("=== Summary ===");
|
|
println!("✓ Used listen_blocking() to block a thread");
|
|
println!("✓ Used listen_timeout_blocking() for safe timeouts");
|
|
println!("✓ Demonstrated timeout success and failure");
|
|
println!("✓ Understood deadlock risks and prevention");
|
|
println!("✓ Learned safe patterns: timeout and polling");
|
|
println!("\nKEY TAKEAWAYS:");
|
|
println!(" 1. listen_blocking() blocks the THREAD (not just task)");
|
|
println!(" 2. Always use timeout or guarantee announcement");
|
|
println!(" 3. Dropping announcer without announcing = deadlock");
|
|
println!(" 4. Prefer async listen().await in async code");
|
|
println!(" 5. Use blocking only in dedicated threads");
|
|
println!("\nCOMPLETED: Learning guide finished!");
|
|
println!("See shutdown.rs, config_broadcast.rs, lazy_init.rs for real-world patterns");
|
|
|
|
// EXPECTED OUTPUT:
|
|
// ================
|
|
// You should see:
|
|
// - Thread blocking and receiving value
|
|
// - Timeout success (value within timeout)
|
|
// - Timeout failure (no value, returns error)
|
|
// - Multiple threads all receiving broadcast
|
|
// - Deadlock explanation and prevention
|
|
// - Safe patterns with timeout and polling
|
|
// - Decision guide for choosing the right method
|
|
}
|