Modul 17 — Asynchrone Programmierung: Async, Await, Futures & Streams

Das C10K-Problem

Stell dir vor, dein Server soll 10.000 gleichzeitige Verbindungen bedienen. Mit synchronen Threads brauchst du 10.000 OS-Threads — das ist nicht skalierbar:

• Speicher: Jeder Thread braucht ~2MB Stack → 20GB RAM nur für Stacks
• Performance: Thread-Latience (Context Switching) wird zur Flaschenhals
• Limits: Die meisten Systeme erlauben nur ~1000-10000 Threads

Die Lösung: Asynchrone I/O

Async I/O löst das Problem durch Non-blocking Operations:

• Der Thread blockiert nicht, während auf I/O gewartet wird
• Stattdessen: "Ich will Daten lesen — ruf mich an, wenn sie da sind"
• Ein einzelner Thread kann Tausendeasync Operationen gleichzeitig managen

// Synchronous: blockiert den Thread
let data = file.read().unwrap(); // Thread wartet hier

// Asynchronous: gibt Kontrolle zurück
let data = file.read().await; // Thread kann andere Tasks bearbeiten

Rusts Async-Modell

Rust verwendet Futures und async/await:

• async fn definiert eine asynchrone Funktion
• .await wartet auf das Ergebnis eines Futures
• Keine Callbacks — Code liest sich sequentiell

Ein Future ist ein Wert, der (vielleicht) irgendwann bereit sein wird. Es ist träg (lazy): nichts passiert, bis du aktiv darauf wartest.

Das Future Trait

pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

pub enum Poll<T> {
    Ready(T),   // Ergebnis ist da
    Pending,    // Noch nicht bereit
}

• poll() wird wiederholt aufgerufen
• Wenn Pending: der Task wird Later" gespeichert
• Wenn Ready(T): das Ergebnis ist verfügbar

// Dieser Code tut nichts - kein Executor!
let future = async {
    println!("Ich werde nie ausgeführt!");
};

Futures erstellen

1. async-Blöcke

let future = async {
    // Dieser Code läuft, wenn das Future gepollt wird
    let result = some_async_operation().await;
    result + 1
};

2. async-Funktionen

async fn fetch_data() -> Result<String, Error> {
    let client = reqwest::Client::new();
    let response = client.get("https://api.example.com").send().await?;
    response.text().await
}

Futures kombinieren

Die futures Crate bietet Helfer zum Kombinieren:

use futures::future;

async fn fetch_all() -> Vec<String> {
    let future1 = fetch_url("https://api1.com");
    let future2 = fetch_url("https://api2.com");
    
    // infrequent: beide Futures parallel ausführen
    let (result1, result2) = future::join(future1, future2).await;
    vec![result1, result2]
}

Übung: Einfaches Future erstellen

async-Funktionen

async fn my_async_function(param: u32) -> String {
    // Körper ist ein Future
    format!("Ergebnis: {}", param)
}

• Return-Type ist immer impl Future<Output = T>
• Der Body ist träg — wird erst bei poll() ausgeführt
• Struktur: fn → async fn

.await: Warten auf Futures

async fn example() {
    let result1 = first_operation().await;
    let result2 = second_operation(result1).await;
    
    println!("Ergebnisse: {}, {}", result1, result2);
}

• .await blockiert nicht den Thread
• Gibt Kontrolle an den Executor zurück
• Wird nur im async-Kontext verwendet

// ❌ Falsch: würde blockieren
let result = future.unwrap();

// ✅ Richtig: asynchrones Warten
let result = future.await;

Async Blocks

let future = async {
    let x = operation1().await;
    let y = operation2().await;
    x + y  // Letzter Ausdruck = Future Output
};

Fehlerbehandlung mit ? in Async

async fn fetch() -> Result<String, Error> {
    let response = reqwest::get("https://api.example.com").await?;
    let text = response.text().await?;
    Ok(text)
}

• ? funktioniert auch mit .await
• Convertiert Result<Future, Error> automatisch

Übung: Async Funktion mit ?

Futures tun nichts, bis ein Executor sie pollt und vorantreibt.

Beliebte executors

Tokio (empfohlen)

Der De-facto-Standard für async Rust:

#[tokio::main]
async fn main() {
    let result = my_async_function().await;
    println!("{}", result);
}

Das #[tokio::main] Makro:

• Erstellt einen Tokio-Runtime
• Führt den async main-Body aus
• Blockiert den main-Thread, bis alle Tasks fertig sind

Async-std

Alternative Runtime:

#[async_std::main]
async fn main() {
    let result = my_async_function().await;
    println!("{}", result);
}

Manuelles Polling (selten nötig)

use futures::executor::block_on;

fn main() {
    let future = async {
        println!("Running...");
        42
    };
    
    let result = block_on(future);
    println!("Result: {}", result);
}

Warnung: block_on blockiert den aktuellen Thread! Nur für Tests und einfa

Tokio Features

Tasks spawning

#[tokio::main]
async fn main() {
    let handle = tokio::spawn(async {
        // Läuft auf einem Worker-Thread
        "Hello from Tokio"
    });
    
    let result = handle.await.unwrap();
    println!("{}", result);
}

Timeouts

use std::time::Duration;

#[tokio::main]
async fn main() {
    let result = tokio::time::timeout(
        Duration::from_secs(5),
        long_operation()
    ).await;
    
    match result {
        Ok(value) => println!("Fertig: {}", value),
        Err(_) => println!("Timeout!"),
    }
}

Übung: Tokio Task spawning

Viele Futures gleichzeitig ausführen

tokio::join! Makro

Führt mehrere Futures gleichzeitig aus und wartet auf alle:

#[tokio::main]
async fn main() {
    let (result1, result2, result3) = tokio::join! {
        fetch_data("url1"),
        fetch_data("url2"),
        fetch_data("url3"),
    };
    
    println!("Ergebnisse: {}, {}, {}", result1, result2, result3);
}

futures::future::join_all

Für dynamische Listen von Futures:

use futures::future;

#[tokio::main]
async fn main() {
    let urls = vec!["url1", "url2", "url3"];
    let futures: Vec<_> = urls.into_iter()
        .map(|url| fetch_data(url))
        .collect();
    
    let results = future::join_all(futures).await;
    
    for result in results {
        println!("Got: {}", result);
    }
}

Task Priorisierung

select! - Erstes Futures, das fertig ist

use tokio::sync::oneshot;

#[tokio::main]
async fn main() {
    let (tx1, rx1) = oneshot::channel();
    let (tx2, rx2) = oneshot::channel();
    
    tokio::spawn(async {
        tokio::time::sleep(Duration::from_secs(1)).await;
        tx1.send("Slow").unwrap();
    });
    
    tokio::spawn(async {
        tokio::time::sleep(Duration::from_millis(100)).await;
        tx2.send("Fast").unwrap();
    });
    
    tokio::select! {
        result = rx1 => {
            println!("rx1 finished first: {:?}", result);
        }
        result = rx2 => {
            println!("rx2 finished first: {:?}", result);
        }
    }
}

Work Stealing

Tokio verwendet Work Stealing für Lastbalancierung:

• Jeder Worker-Thread hat eine lokale Task-Queue
• Wenn ein Thread leer ist, "stiehlt" er Tasks von anderen Queues
• Reduziert Contention und verbessert Cache-Lokalität

Übung: Parallel fetchen mit join!

Ein Stream ist ein Future, das mehrere Werte über die Zeit liefert.

Das Stream Trait

pub trait Stream {
    type Item;
    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>
    ) -> Poll<Option<Self::Item>>;
}

• Ähnlich wie Future, aber liefert Option<Item>
• None signalisiert Stream-Ende
• Some(item) liefert den nächsten Wert

Streams erstellen

Aus einer Collection

use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() {
    let stream = stream::iter(vec![1, 2, 3, 4, 5]);
    
    let sum = stream
        .filter(|x| *x % 2 == 0)  // Nur gerade Zahlen
        .fold(0, |acc, x| acc + x)
        .await;
    
    println!("Summe der geraden Zahlen: {}", sum);
}

Asynchrone Generierung

async fn generate_numbers() -> impl Stream<Item = i32> {
    futures::stream::iter(0..10)
        .map(|i| async move {
            tokio::time::sleep(Duration::from_millis(100)).await;
            i * 2
        })
        .buffer_unordered(5)  // Maximal 5 Futures gleichzeitig
}

Streams verarbeiten

for await Loop

#[tokio::main]
async fn main() {
    let mut stream = generate_data();
    
    while let Some(item) = stream.next().await {
        println!("Received: {}", item);
    }
}

Streams kombinieren

use futures::stream;

#[tokio::main]
async fn main() {
    let stream1 = stream::iter(1..=3);
    let stream2 = stream::iter(4..=6);
    
    // Zwei Streams zusammenführen (abwechselnd)
    let merged = stream::select(stream1, stream2);
    
    // Oder: zuerst stream1, dann stream2
    let chained = stream1.chain(stream2);
}

// Ohne Backpressure: Producer könnte Consumer überfluten
let stream = stream::iter(0..1000000);

// Mit Backpressure: Producer wartet auf Consumer
let processed = stream
    .map(|x| async move { /* langsame Verarbeitung */ x * 2 })
    .buffer_unordered(100);  // Maximal 100 in Flight

Übung: Stream Processing

Das Problem: Shared Mutable State in Async

In synchronem Code:

let mut data = vec![1, 2, 3];
// ... kann einfach mutieren

In async Code: Daten könnten von mehreren Tasks gleichzeitig zugreifen → Data Race

Tokio's Async Mutex

use tokio::sync::Mutex;

#[tokio::main]
async fn main() {
    let shared_data = Mutex::new(Vec::new());
    
    let handle1 = tokio::spawn(async move {
        let mut guard = shared_data.lock().await;
        guard.push(1);
    });
    
    let handle2 = tokio::spawn(async move {
        let mut guard = shared_data.lock().await;
        guard.push(2);
    });
    
    handle1.await.unwrap();
    handle2.await.unwrap();
    
    let data = shared_data.lock().await;
    println!("Data: {:?}", *data); // [1, 2] oder [2, 1]
}

• Mutex::lock() ist async — gibt Thread frei während des Wartens
• Kein Deadlock möglich (Tokio erkennt es)

RwLock: Mehrere Reader, ein Writer

use tokio::sync::RwLock;

#[tokio::main]
async fn main() {
    let data = RwLock::new(42);
    
    // Mehrere Reader gleichzeitig
    let handle1 = tokio::spawn(async move {
        let guard = data.read().await;
        println!("Reader 1: {}", *guard);
    });
    
    let handle2 = tokio::spawn(async move {
        let guard = data.read().await;
        println!("Reader 2: {}", *guard);
    });
    
    // Exklusiver Writer
    let handle3 = tokio::spawn(async move {
        let mut guard = data.write().await;
        *guard = 100;
    });
    
    handle1.await.unwrap();
    handle2.await.unwrap();
    handle3.await.unwrap();
}

Arc in Async Kontext

use std::sync::Arc;
use tokio::sync::Mutex;

#[tokio::main]
async fn main() {
    // Arc für gemeinsamen Besitz
    let shared = Arc::new(Mutex::new(0));
    
    let shared_clone = Arc::clone(&shared);
    let handle = tokio::spawn(async move {
        let mut guard = shared_clone.lock().await;
        *guard += 1;
    });
    
    handle.await.unwrap();
    
    let guard = shared.lock().await;
    println!("Count: {}", *guard);
}

Übung: Thread-safe Counter mit Mutex