huiali/rust-skills/rust-async

use tokio_stream::{self as stream, StreamExt};
 
async fn process_stream(stream: impl Stream<Item = Data>) {
    stream
        .chunks(100)           // Batch processing
        .for_each(|batch| async {
            process_batch(batch).await;
        })
        .await;
}

use tokio::sync::Semaphore;
use std::sync::Arc;
 
let semaphore = Arc::new(Semaphore::new(10));  // Max 10 concurrent
 
let stream = tokio_stream::iter(0..1000)
    .map(|i| {
        let permit = semaphore.clone().acquire_owned();
        async move {
            let _permit = permit.await?;
            process(i).await
        }
    })
    .buffer_unordered(100);  // Max 100 buffered futures

use tokio::select;
use tokio::time::{sleep, Duration};
 
async fn multiplex() {
    loop {
        select! {
            msg = receiver.recv() => {
                if let Some(msg) = msg {
                    handle(msg).await;
                } else {
                    break;  // Channel closed
                }
            }
            _ = sleep(Duration::from_secs(5)) => {
                // Timeout handling
                check_health().await;
            }
            else => break,  // All branches complete
        }
    }
}

use tokio::time::timeout;
use std::time::Duration;
 
async fn with_timeout() -> Result<Value, TimeoutError> {
    timeout(Duration::from_secs(5), long_operation()).await
        .map_err(|_| TimeoutError)?
}
 
// Cooperative cancellation
let mut task = tokio::spawn(async move {
    loop {
        // Check cancellation
        tokio::task::yield_now().await;  // Yield point
 
        // Do work
        if let Err(_) = work().await {
            return;
        }
    }
});
 
// Cancel task
task.abort();
let _ = task.await;  // Will return JoinError::Cancelled

Sync data source?
  → Use Iterator (more efficient)
 
Async data source (network, DB)?
  → Use Stream
 
Need backpressure?
  → Definitely Stream

Sequential processing?
  → for_each / fold
 
Limited concurrency?
  → buffer_unordered(N) + Semaphore
 
Unlimited (dangerous)?
  → Use with extreme caution

Long-running task?
  → Add timeout wrapper
 
User-initiated?
  → Implement abort signal
 
Resource cleanup?
  → Use Drop or explicit cleanup

use tokio::join;
 
// All operations run concurrently, wait for all to complete
let (a, b, c) = join!(
    fetch_user(),
    fetch_posts(),
    fetch_comments()
);
// All values available, even if some operations failed

use tokio::try_join;
 
// Stop on first error
let (a, b) = try_join!(
    async_op_a(),
    async_op_b()
)?;
// Both succeeded, or error from first failure

async fn fetch_dashboard() -> Result<Dashboard, Error> {
    let (user, posts, comments) = try_join!(
        fetch_user(),
        fetch_posts(),
        fetch_comments()
    )?;
 
    Ok(Dashboard { user, posts, comments })
}

let sem = Arc::new(Semaphore::new(10));
 
stream
    .map(|item| {
        let sem = sem.clone();
        async move {
            let _permit = sem.acquire().await?;
            process(item).await
        }
    })
    .buffer_unordered(10)

stream
    .chunks(100)
    .for_each_concurrent(5, |batch| async move {
        process_batch(batch).await
    })
    .await

let (tx, mut rx) = mpsc::channel(100);  // Buffer size = backpressure
 
// Producer respects backpressure
tx.send(item).await?;
 
// Consumer pulls at own pace
while let Some(item) = rx.recv().await {
    process(item).await;
}

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
 
struct Delay {
    when: Instant,
}
 
impl Future for Delay {
    type Output = ();
 
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if Instant::now() >= self.when {
            Poll::Ready(())
        } else {
            // Wake me later
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

# Check async code compilation
cargo check
 
# Run async tests
cargo test
 
# Check for common async mistakes
cargo clippy -- -W clippy::await_holding_lock
 
# Test with tokio-console for debugging
RUSTFLAGS="--cfg tokio_unstable" cargo run
 
# Profile async runtime
cargo flamegraph --bin your-app

// ❌ Bad: future not awaited
async fn bad() {
    fetch_data();  // Returns Future, never runs!
}
 
// ✅ Good
async fn good() {
    fetch_data().await;  // Actually runs
}

// ❌ Bad: all operations run concurrently
let futures: Vec<_> = urls.iter()
    .map(|url| fetch(url))
    .collect();
let results = join_all(futures).await;
 
// ✅ Good: limited concurrency
use futures::stream::{self, StreamExt};
 
let results = stream::iter(urls)
    .map(|url| fetch(url))
    .buffer_unordered(10)  // Max 10 concurrent
    .collect::<Vec<_>>()
    .await;

// ❌ Bad: lock held during await
let guard = mutex.lock().await;
some_async_op().await;  // DANGER
drop(guard);
 
// ✅ Good: drop lock before await
let value = {
    let guard = mutex.lock().await;
    guard.clone()
};  // lock dropped
some_async_op().await;

// ❌ Bad: async operation in Drop
impl Drop for Resource {
    fn drop(&mut self) {
        // Cannot await here!
        self.cleanup().await;  // Won't compile
    }
}
 
// ✅ Good: explicit async cleanup
impl Resource {
    async fn cleanup(self) {
        // Async cleanup logic
    }
}
 
// Or spawn cleanup task
impl Drop for Resource {
    fn drop(&mut self) {
        let handle = self.handle.take();
        tokio::spawn(async move {
            if let Some(h) = handle {
                h.cleanup().await;
            }
        });
    }
}