Add: windows mvp - transparent bugs not fixed
This commit is contained in:
DaZuo0122
409
skills/m07-concurrency/patterns/async-patterns.md
Normal file
409
skills/m07-concurrency/patterns/async-patterns.md
Normal file
@@ -0,0 +1,409 @@
|
||||
# Async Patterns in Rust
|
||||
|
||||
## Task Spawning
|
||||
|
||||
### Basic Spawn
|
||||
```rust
|
||||
use tokio::task;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
// Spawn a task that runs concurrently
|
||||
let handle = task::spawn(async {
|
||||
expensive_computation().await
|
||||
});
|
||||
|
||||
// Do other work while task runs
|
||||
other_work().await;
|
||||
|
||||
// Wait for result
|
||||
let result = handle.await.unwrap();
|
||||
}
|
||||
```
|
||||
|
||||
### Spawn with Shared State
|
||||
```rust
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
async fn process_with_state() {
|
||||
let state = Arc::new(Mutex::new(vec![]));
|
||||
|
||||
let handles: Vec<_> = (0..10)
|
||||
.map(|i| {
|
||||
let state = Arc::clone(&state);
|
||||
tokio::spawn(async move {
|
||||
let mut guard = state.lock().await;
|
||||
guard.push(i);
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Wait for all tasks
|
||||
for handle in handles {
|
||||
handle.await.unwrap();
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Select Pattern
|
||||
|
||||
### Racing Multiple Futures
|
||||
```rust
|
||||
use tokio::select;
|
||||
use tokio::time::{sleep, Duration};
|
||||
|
||||
async fn first_response() {
|
||||
select! {
|
||||
result = fetch_from_server_a() => {
|
||||
println!("A responded first: {:?}", result);
|
||||
}
|
||||
result = fetch_from_server_b() => {
|
||||
println!("B responded first: {:?}", result);
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Select with Timeout
|
||||
```rust
|
||||
use tokio::time::timeout;
|
||||
|
||||
async fn with_timeout() -> Result<Data, Error> {
|
||||
select! {
|
||||
result = fetch_data() => result,
|
||||
_ = sleep(Duration::from_secs(5)) => {
|
||||
Err(Error::Timeout)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Or use timeout directly
|
||||
async fn with_timeout2() -> Result<Data, Error> {
|
||||
timeout(Duration::from_secs(5), fetch_data())
|
||||
.await
|
||||
.map_err(|_| Error::Timeout)?
|
||||
}
|
||||
```
|
||||
|
||||
### Select with Channel
|
||||
```rust
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
async fn process_messages(mut rx: mpsc::Receiver<Message>) {
|
||||
loop {
|
||||
select! {
|
||||
Some(msg) = rx.recv() => {
|
||||
handle_message(msg).await;
|
||||
}
|
||||
_ = tokio::signal::ctrl_c() => {
|
||||
println!("Shutting down...");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Channel Patterns
|
||||
|
||||
### MPSC (Multi-Producer, Single-Consumer)
|
||||
```rust
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
async fn producer_consumer() {
|
||||
let (tx, mut rx) = mpsc::channel(100);
|
||||
|
||||
// Spawn producers
|
||||
for i in 0..3 {
|
||||
let tx = tx.clone();
|
||||
tokio::spawn(async move {
|
||||
tx.send(format!("Message from {}", i)).await.unwrap();
|
||||
});
|
||||
}
|
||||
|
||||
// Drop original sender so channel closes
|
||||
drop(tx);
|
||||
|
||||
// Consume
|
||||
while let Some(msg) = rx.recv().await {
|
||||
println!("Received: {}", msg);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Oneshot (Single-Shot Response)
|
||||
```rust
|
||||
use tokio::sync::oneshot;
|
||||
|
||||
async fn request_response() {
|
||||
let (tx, rx) = oneshot::channel();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let result = compute_something().await;
|
||||
tx.send(result).unwrap();
|
||||
});
|
||||
|
||||
// Wait for response
|
||||
let response = rx.await.unwrap();
|
||||
}
|
||||
```
|
||||
|
||||
### Broadcast (Multi-Consumer)
|
||||
```rust
|
||||
use tokio::sync::broadcast;
|
||||
|
||||
async fn pub_sub() {
|
||||
let (tx, _) = broadcast::channel(16);
|
||||
|
||||
// Subscribe multiple consumers
|
||||
let mut rx1 = tx.subscribe();
|
||||
let mut rx2 = tx.subscribe();
|
||||
|
||||
tokio::spawn(async move {
|
||||
while let Ok(msg) = rx1.recv().await {
|
||||
println!("Consumer 1: {}", msg);
|
||||
}
|
||||
});
|
||||
|
||||
tokio::spawn(async move {
|
||||
while let Ok(msg) = rx2.recv().await {
|
||||
println!("Consumer 2: {}", msg);
|
||||
}
|
||||
});
|
||||
|
||||
// Publish
|
||||
tx.send("Hello").unwrap();
|
||||
}
|
||||
```
|
||||
|
||||
### Watch (Single Latest Value)
|
||||
```rust
|
||||
use tokio::sync::watch;
|
||||
|
||||
async fn config_updates() {
|
||||
let (tx, mut rx) = watch::channel(Config::default());
|
||||
|
||||
// Consumer watches for changes
|
||||
tokio::spawn(async move {
|
||||
while rx.changed().await.is_ok() {
|
||||
let config = rx.borrow();
|
||||
apply_config(&config);
|
||||
}
|
||||
});
|
||||
|
||||
// Update config
|
||||
tx.send(Config::new()).unwrap();
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Structured Concurrency
|
||||
|
||||
### JoinSet for Task Groups
|
||||
```rust
|
||||
use tokio::task::JoinSet;
|
||||
|
||||
async fn parallel_fetch(urls: Vec<String>) -> Vec<Result<Response, Error>> {
|
||||
let mut set = JoinSet::new();
|
||||
|
||||
for url in urls {
|
||||
set.spawn(async move {
|
||||
fetch(&url).await
|
||||
});
|
||||
}
|
||||
|
||||
let mut results = vec![];
|
||||
while let Some(res) = set.join_next().await {
|
||||
results.push(res.unwrap());
|
||||
}
|
||||
results
|
||||
}
|
||||
```
|
||||
|
||||
### Scoped Tasks (no 'static)
|
||||
```rust
|
||||
// Using tokio-scoped or async-scoped crate
|
||||
use async_scoped::TokioScope;
|
||||
|
||||
async fn scoped_example(data: &[u32]) {
|
||||
let results = TokioScope::scope_and_block(|scope| {
|
||||
for item in data {
|
||||
scope.spawn(async move {
|
||||
process(item).await
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Cancellation Patterns
|
||||
|
||||
### Using CancellationToken
|
||||
```rust
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
async fn cancellable_task(token: CancellationToken) {
|
||||
loop {
|
||||
select! {
|
||||
_ = token.cancelled() => {
|
||||
println!("Task cancelled");
|
||||
break;
|
||||
}
|
||||
_ = do_work() => {
|
||||
// Continue working
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn main_with_cancellation() {
|
||||
let token = CancellationToken::new();
|
||||
let task_token = token.clone();
|
||||
|
||||
let handle = tokio::spawn(cancellable_task(task_token));
|
||||
|
||||
// Cancel after some condition
|
||||
tokio::time::sleep(Duration::from_secs(5)).await;
|
||||
token.cancel();
|
||||
|
||||
handle.await.unwrap();
|
||||
}
|
||||
```
|
||||
|
||||
### Graceful Shutdown
|
||||
```rust
|
||||
async fn serve_with_shutdown(shutdown: impl Future) {
|
||||
let server = TcpListener::bind("0.0.0.0:8080").await.unwrap();
|
||||
|
||||
loop {
|
||||
select! {
|
||||
Ok((socket, _)) = server.accept() => {
|
||||
tokio::spawn(handle_connection(socket));
|
||||
}
|
||||
_ = &mut shutdown => {
|
||||
println!("Shutting down...");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
let ctrl_c = async {
|
||||
tokio::signal::ctrl_c().await.unwrap();
|
||||
};
|
||||
|
||||
serve_with_shutdown(ctrl_c).await;
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Backpressure Patterns
|
||||
|
||||
### Bounded Channels
|
||||
```rust
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
async fn with_backpressure() {
|
||||
// Buffer of 10 - producers will wait if full
|
||||
let (tx, mut rx) = mpsc::channel(10);
|
||||
|
||||
let producer = tokio::spawn(async move {
|
||||
for i in 0..1000 {
|
||||
// This will wait if channel is full
|
||||
tx.send(i).await.unwrap();
|
||||
}
|
||||
});
|
||||
|
||||
let consumer = tokio::spawn(async move {
|
||||
while let Some(item) = rx.recv().await {
|
||||
// Slow consumer
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
process(item);
|
||||
}
|
||||
});
|
||||
|
||||
let _ = tokio::join!(producer, consumer);
|
||||
}
|
||||
```
|
||||
|
||||
### Semaphore for Rate Limiting
|
||||
```rust
|
||||
use tokio::sync::Semaphore;
|
||||
use std::sync::Arc;
|
||||
|
||||
async fn rate_limited_requests(urls: Vec<String>) {
|
||||
let semaphore = Arc::new(Semaphore::new(10)); // max 10 concurrent
|
||||
|
||||
let handles: Vec<_> = urls
|
||||
.into_iter()
|
||||
.map(|url| {
|
||||
let sem = Arc::clone(&semaphore);
|
||||
tokio::spawn(async move {
|
||||
let _permit = sem.acquire().await.unwrap();
|
||||
fetch(&url).await
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
for handle in handles {
|
||||
handle.await.unwrap();
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Error Handling in Async
|
||||
|
||||
### Propagating Errors
|
||||
```rust
|
||||
async fn fetch_and_parse(url: &str) -> Result<Data, Error> {
|
||||
let response = fetch(url).await?;
|
||||
let data = parse(response).await?;
|
||||
Ok(data)
|
||||
}
|
||||
```
|
||||
|
||||
### Handling Task Panics
|
||||
```rust
|
||||
async fn robust_spawn() {
|
||||
let handle = tokio::spawn(async {
|
||||
risky_operation().await
|
||||
});
|
||||
|
||||
match handle.await {
|
||||
Ok(result) => println!("Success: {:?}", result),
|
||||
Err(e) if e.is_panic() => {
|
||||
println!("Task panicked: {:?}", e);
|
||||
}
|
||||
Err(e) => {
|
||||
println!("Task cancelled: {:?}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Try-Join for Multiple Results
|
||||
```rust
|
||||
use tokio::try_join;
|
||||
|
||||
async fn fetch_all() -> Result<(A, B, C), Error> {
|
||||
// All must succeed, or first error returned
|
||||
try_join!(
|
||||
fetch_a(),
|
||||
fetch_b(),
|
||||
fetch_c(),
|
||||
)
|
||||
}
|
||||
```
|
||||
331
skills/m07-concurrency/patterns/common-errors.md
Normal file
331
skills/m07-concurrency/patterns/common-errors.md
Normal file
@@ -0,0 +1,331 @@
|
||||
# Common Concurrency Errors & Fixes
|
||||
|
||||
## E0277: Cannot Send Between Threads
|
||||
|
||||
### Error Pattern
|
||||
```rust
|
||||
use std::rc::Rc;
|
||||
|
||||
let data = Rc::new(42);
|
||||
std::thread::spawn(move || {
|
||||
println!("{}", data); // ERROR: Rc<i32> cannot be sent between threads
|
||||
});
|
||||
```
|
||||
|
||||
### Fix Options
|
||||
|
||||
**Option 1: Use Arc instead**
|
||||
```rust
|
||||
use std::sync::Arc;
|
||||
|
||||
let data = Arc::new(42);
|
||||
let data_clone = Arc::clone(&data);
|
||||
std::thread::spawn(move || {
|
||||
println!("{}", data_clone); // OK: Arc is Send
|
||||
});
|
||||
```
|
||||
|
||||
**Option 2: Move owned data**
|
||||
```rust
|
||||
let data = 42; // i32 is Copy and Send
|
||||
std::thread::spawn(move || {
|
||||
println!("{}", data); // OK
|
||||
});
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## E0277: Cannot Share Between Threads (Not Sync)
|
||||
|
||||
### Error Pattern
|
||||
```rust
|
||||
use std::cell::RefCell;
|
||||
use std::sync::Arc;
|
||||
|
||||
let data = Arc::new(RefCell::new(42));
|
||||
// ERROR: RefCell is not Sync
|
||||
```
|
||||
|
||||
### Fix Options
|
||||
|
||||
**Option 1: Use Mutex for thread-safe interior mutability**
|
||||
```rust
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
let data = Arc::new(Mutex::new(42));
|
||||
let data_clone = Arc::clone(&data);
|
||||
std::thread::spawn(move || {
|
||||
let mut guard = data_clone.lock().unwrap();
|
||||
*guard += 1;
|
||||
});
|
||||
```
|
||||
|
||||
**Option 2: Use RwLock for read-heavy workloads**
|
||||
```rust
|
||||
use std::sync::{Arc, RwLock};
|
||||
|
||||
let data = Arc::new(RwLock::new(42));
|
||||
let data_clone = Arc::clone(&data);
|
||||
std::thread::spawn(move || {
|
||||
let guard = data_clone.read().unwrap();
|
||||
println!("{}", *guard);
|
||||
});
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Deadlock Patterns
|
||||
|
||||
### Pattern 1: Lock Ordering Deadlock
|
||||
```rust
|
||||
// DANGER: potential deadlock
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
let a = Arc::new(Mutex::new(1));
|
||||
let b = Arc::new(Mutex::new(2));
|
||||
|
||||
// Thread 1: locks a then b
|
||||
let a1 = Arc::clone(&a);
|
||||
let b1 = Arc::clone(&b);
|
||||
std::thread::spawn(move || {
|
||||
let _a = a1.lock().unwrap();
|
||||
let _b = b1.lock().unwrap(); // waits for b
|
||||
});
|
||||
|
||||
// Thread 2: locks b then a (opposite order!)
|
||||
let a2 = Arc::clone(&a);
|
||||
let b2 = Arc::clone(&b);
|
||||
std::thread::spawn(move || {
|
||||
let _b = b2.lock().unwrap();
|
||||
let _a = a2.lock().unwrap(); // waits for a - DEADLOCK
|
||||
});
|
||||
```
|
||||
|
||||
### Fix: Consistent Lock Ordering
|
||||
```rust
|
||||
// SAFE: always lock in same order (a before b)
|
||||
std::thread::spawn(move || {
|
||||
let _a = a1.lock().unwrap();
|
||||
let _b = b1.lock().unwrap();
|
||||
});
|
||||
|
||||
std::thread::spawn(move || {
|
||||
let _a = a2.lock().unwrap(); // same order
|
||||
let _b = b2.lock().unwrap();
|
||||
});
|
||||
```
|
||||
|
||||
### Pattern 2: Self-Deadlock
|
||||
```rust
|
||||
// DANGER: locking same mutex twice
|
||||
let m = Mutex::new(42);
|
||||
let _g1 = m.lock().unwrap();
|
||||
let _g2 = m.lock().unwrap(); // DEADLOCK on std::Mutex
|
||||
|
||||
// FIX: use parking_lot::ReentrantMutex if needed
|
||||
// or restructure code to avoid double locking
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Mutex Guard Across Await
|
||||
|
||||
### Error Pattern
|
||||
```rust
|
||||
use std::sync::Mutex;
|
||||
use tokio::time::sleep;
|
||||
|
||||
async fn bad_async() {
|
||||
let m = Mutex::new(42);
|
||||
let guard = m.lock().unwrap();
|
||||
sleep(Duration::from_secs(1)).await; // WARNING: guard held across await
|
||||
println!("{}", *guard);
|
||||
}
|
||||
```
|
||||
|
||||
### Fix Options
|
||||
|
||||
**Option 1: Scope the lock**
|
||||
```rust
|
||||
async fn good_async() {
|
||||
let m = Mutex::new(42);
|
||||
let value = {
|
||||
let guard = m.lock().unwrap();
|
||||
*guard // copy value
|
||||
}; // guard dropped here
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
println!("{}", value);
|
||||
}
|
||||
```
|
||||
|
||||
**Option 2: Use tokio::sync::Mutex**
|
||||
```rust
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
async fn good_async() {
|
||||
let m = Mutex::new(42);
|
||||
let guard = m.lock().await; // async lock
|
||||
sleep(Duration::from_secs(1)).await; // OK with tokio::Mutex
|
||||
println!("{}", *guard);
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Data Race Prevention
|
||||
|
||||
### Pattern: Missing Synchronization
|
||||
```rust
|
||||
// This WON'T compile - Rust prevents data races
|
||||
use std::sync::Arc;
|
||||
|
||||
let data = Arc::new(0);
|
||||
let d1 = Arc::clone(&data);
|
||||
let d2 = Arc::clone(&data);
|
||||
|
||||
std::thread::spawn(move || {
|
||||
// *d1 += 1; // ERROR: cannot mutate through Arc
|
||||
});
|
||||
|
||||
std::thread::spawn(move || {
|
||||
// *d2 += 1; // ERROR: cannot mutate through Arc
|
||||
});
|
||||
```
|
||||
|
||||
### Fix: Add Synchronization
|
||||
```rust
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::sync::atomic::{AtomicI32, Ordering};
|
||||
|
||||
// Option 1: Mutex
|
||||
let data = Arc::new(Mutex::new(0));
|
||||
let d1 = Arc::clone(&data);
|
||||
std::thread::spawn(move || {
|
||||
*d1.lock().unwrap() += 1;
|
||||
});
|
||||
|
||||
// Option 2: Atomic (for simple types)
|
||||
let data = Arc::new(AtomicI32::new(0));
|
||||
let d1 = Arc::clone(&data);
|
||||
std::thread::spawn(move || {
|
||||
d1.fetch_add(1, Ordering::SeqCst);
|
||||
});
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Channel Errors
|
||||
|
||||
### Disconnected Channel
|
||||
```rust
|
||||
use std::sync::mpsc;
|
||||
|
||||
let (tx, rx) = mpsc::channel();
|
||||
drop(tx); // sender dropped
|
||||
match rx.recv() {
|
||||
Ok(v) => println!("{}", v),
|
||||
Err(_) => println!("channel disconnected"), // this happens
|
||||
}
|
||||
```
|
||||
|
||||
### Fix: Handle Disconnection
|
||||
```rust
|
||||
// Use try_recv for non-blocking
|
||||
loop {
|
||||
match rx.try_recv() {
|
||||
Ok(msg) => handle(msg),
|
||||
Err(TryRecvError::Empty) => continue,
|
||||
Err(TryRecvError::Disconnected) => break,
|
||||
}
|
||||
}
|
||||
|
||||
// Or iterate (stops on disconnect)
|
||||
for msg in rx {
|
||||
handle(msg);
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Async Common Errors
|
||||
|
||||
### Forgetting to Spawn
|
||||
```rust
|
||||
// WRONG: future not polled
|
||||
async fn fetch_data() -> Result<Data, Error> { ... }
|
||||
|
||||
fn process() {
|
||||
fetch_data(); // does nothing! returns Future that's dropped
|
||||
}
|
||||
|
||||
// RIGHT: await or spawn
|
||||
async fn process() {
|
||||
let data = fetch_data().await; // awaited
|
||||
}
|
||||
|
||||
fn process_sync() {
|
||||
tokio::spawn(fetch_data()); // spawned
|
||||
}
|
||||
```
|
||||
|
||||
### Blocking in Async Context
|
||||
```rust
|
||||
// WRONG: blocks the executor
|
||||
async fn bad() {
|
||||
std::thread::sleep(Duration::from_secs(1)); // blocks!
|
||||
std::fs::read_to_string("file.txt").unwrap(); // blocks!
|
||||
}
|
||||
|
||||
// RIGHT: use async versions
|
||||
async fn good() {
|
||||
tokio::time::sleep(Duration::from_secs(1)).await;
|
||||
tokio::fs::read_to_string("file.txt").await.unwrap();
|
||||
}
|
||||
|
||||
// Or spawn_blocking for CPU-bound work
|
||||
async fn compute() {
|
||||
let result = tokio::task::spawn_blocking(|| {
|
||||
heavy_computation() // OK to block here
|
||||
}).await.unwrap();
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Thread Panic Handling
|
||||
|
||||
### Unhandled Panic
|
||||
```rust
|
||||
let handle = std::thread::spawn(|| {
|
||||
panic!("oops");
|
||||
});
|
||||
|
||||
// Main thread continues, might miss the error
|
||||
handle.join().unwrap(); // panics here
|
||||
```
|
||||
|
||||
### Proper Error Handling
|
||||
```rust
|
||||
let handle = std::thread::spawn(|| {
|
||||
panic!("oops");
|
||||
});
|
||||
|
||||
match handle.join() {
|
||||
Ok(result) => println!("Success: {:?}", result),
|
||||
Err(e) => println!("Thread panicked: {:?}", e),
|
||||
}
|
||||
|
||||
// For async: use catch_unwind
|
||||
use std::panic;
|
||||
|
||||
async fn safe_task() {
|
||||
let result = panic::catch_unwind(|| {
|
||||
risky_operation()
|
||||
});
|
||||
|
||||
match result {
|
||||
Ok(v) => use_value(v),
|
||||
Err(_) => log_error("task panicked"),
|
||||
}
|
||||
}
|
||||
```
|
||||
Reference in New Issue
Block a user