Async in Rust im Detail: select!, Timeouts, Streams und Pin

Auf der Seite Async/Await hast du die Grundlagen gelernt: Eine async fn gibt ein „faules“ Future zurück, .await führt es aus, #[tokio::main] startet die Runtime, und mit tokio::join! und tokio::spawn laufen mehrere Dinge gleichzeitig. Wenn dir davon noch etwas neu vorkommt, lies zuerst diese Seite. Hier bauen wir darauf auf.

Echte Async-Programme brauchen noch ein paar Werkzeuge mehr. Denk wieder an eine volle Restaurantküche. Manchmal wartest du auf jedes Gericht (join!). Manchmal zählt nur, was zuerst passiert: Das Essen ist fertig – oder der Gast geht (select!). Manchmal gibst du nach fünf Minuten auf (timeout). Und manchmal kommen den ganzen Abend lang Bestellungen herein, eine nach der anderen (Streams). Diese Seite behandelt all das. Dazu kommt Pin, das seltsame Wort, das dir früher oder später in Fehlermeldungen begegnet – in einfachen Worten erklärt.

Alle Beispiele verwenden tokio mit features = ["full"], genau wie auf der Grundlagenseite. Die „Selbst ausprobieren“-Buttons funktionieren, denn der Rust Playground hat tokio schon eingebaut.

Viele Dinge auf einmal ausführen

tokio::join! kennst du schon: Es wartet auf alle seine Futures und gibt ihre Ergebnisse als Tupel zurück. Sein naher Verwandter tokio::try_join! ist für Futures gedacht, die ein Result zurückgeben. Du bekommst Ok mit allen Werten oder den ersten Err. Sobald ein Future fehlschlägt, wartet try_join! nicht mehr auf die anderen:

use std::time::Duration;
use tokio::time::sleep;

async fn load_profile(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(20)).await;
    Ok(format!("Profil #{id}"))
}

async fn load_settings(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(10)).await;
    if id == 2 {
        return Err(format!("keine Einstellungen für Benutzer {id}"));
    }
    Ok(String::from("dunkles Design"))
}

#[tokio::main]
async fn main() {
    for id in [1, 2] {
        match tokio::try_join!(load_profile(id), load_settings(id)) {
            Ok((profile, settings)) => println!("Benutzer {id}: {profile}, {settings}"),
            Err(e) => println!("Benutzer {id} fehlgeschlagen: {e}"),
        }
    }
}

Ausgabe:

Benutzer 1: Profil #1, dunkles Design
Benutzer 2 fehlgeschlagen: keine Einstellungen für Benutzer 2

▶ Selbst ausprobieren

join! und try_join! führen die Futures innerhalb des aktuellen Tasks aus, abwechselnd. Das ist perfekt für eine Handvoll Futures, die du vorher kennst. Hast du dagegen eine Liste von Aufgaben, deren Länge erst zur Laufzeit feststeht, startest du sie als eigene Tasks. Du könntest jedes JoinHandle in einen Vec legen, wie auf der Grundlagenseite. Aber tokio hat dafür ein schöneres Werkzeug: JoinSet, einen Beutel voller Tasks, der dir jedes Ergebnis gibt, sobald es fertig ist:

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::sleep;

async fn check_site(site: &'static str, ms: u64) -> (&'static str, u64) {
    sleep(Duration::from_millis(ms)).await; // tu so, als wäre das ein Netzwerkaufruf
    (site, ms)
}

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    for (site, ms) in [("rust-lang.org", 30), ("docs.rs", 10), ("crates.io", 20)] {
        set.spawn(check_site(site, ms));
    }

    let mut results = Vec::new();
    while let Some(res) = set.join_next().await {
        results.push(res.unwrap()); // Err nur, wenn der Task in Panik geraten ist
    }

    results.sort(); // die Reihenfolge beim Fertigwerden kann sich von Lauf zu Lauf ändern
    for (site, ms) in results {
        println!("{site} hat in {ms}ms geantwortet");
    }
}

Ausgabe:

crates.io hat in 20ms geantwortet
docs.rs hat in 10ms geantwortet
rust-lang.org hat in 30ms geantwortet

▶ Selbst ausprobieren

join_next() gibt None zurück, wenn das Set leer ist. Die while let-Schleife endet also von selbst. Die Ergebnisse kommen in der Reihenfolge an, in der die Tasks fertig werden, nicht in der Reihenfolge, in der du sie gestartet hast. Brauchen die Tasks ähnlich lange, kann sich diese Reihenfolge zwischen zwei Läufen ändern. Deshalb sortieren wir vor der Ausgabe. Noch etwas Schönes: Wenn du ein JoinSet verwirfst (Drop), wird jeder Task darin abgebrochen. Nichts läuft aus Versehen weiter.

Futures um die Wette laufen lassen mit select!

join! wartet auf alles. tokio::select! wartet auf das erste Future. Jeder Zweig besteht aus einem Pattern, einem Future und einem Codeblock: pattern = future => code. Welches Future zuerst fertig ist, dessen Code läuft. Die anderen Zweige werden abgebrochen.

Was heißt „abgebrochen“ in Rust? Das unterlegene Future wird einfach verworfen (gedroppt), wie jeder andere Wert, der seinen Gültigkeitsbereich verlässt. Es läuft nie wieder. Auch seine lokalen Variablen werden verworfen, also läuft ihr Aufräumcode. Dieses Beispiel macht das mit einem kleinen Drop-Typ sichtbar:

use std::time::Duration;
use tokio::time::sleep;

struct Download(&'static str);

impl Drop for Download {
    fn drop(&mut self) {
        println!("räume den Download von {} auf", self.0);
    }
}

async fn download(mirror: &'static str, ms: u64) -> &'static str {
    let _download = Download(mirror);
    sleep(Duration::from_millis(ms)).await;
    println!("Download von {mirror} fertig");
    mirror
}

#[tokio::main]
async fn main() {
    tokio::select! {
        winner = download("Europa", 50) => println!("Gewinner: {winner}"),
        winner = download("Asien", 1000) => println!("Gewinner: {winner}"),
    }
    println!("main läuft weiter");
}

Ausgabe:

Download von Europa fertig
räume den Download von Europa auf
räume den Download von Asien auf
Gewinner: Europa
main läuft weiter

▶ Selbst ausprobieren

Schau, was fehlt: „Download von Asien fertig“ wird nie ausgegeben. Das Asien-Future wurde an seinem .await angehalten und verworfen, und sein Download-Wert wurde sofort aufgeräumt. So funktioniert Abbrechen überall in Async-Rust: Hör auf, ein Future zu pollen, und verwirf es. Einen speziellen „Abbrechen“-Aufruf brauchst du nicht.

Ein klassischer Einsatz von select! ist ein Timeout: Du lässt deine Arbeit gegen einen Timer antreten.

tokio::select! {
    report = build_report() => println!("{report}"),
    _ = tokio::time::sleep(Duration::from_secs(5)) => println!("zu langsam, ich gebe auf"),
}

Das funktioniert gut, aber es gibt dafür eine Abkürzung: tokio::time::timeout. Die lernst du im nächsten Abschnitt kennen.

Arbeiten bis zum Signal zum Herunterfahren

Richtig glänzt select! in einer loop. Hier holt sich ein Worker Aufgaben aus einem Channel, hört aber gleichzeitig auf einem zweiten Channel auf die Nachricht „bitte aufhören“. Ein oneshot-Channel transportiert genau eine Nachricht. Damit ist er das perfekte Signal zum Herunterfahren:

use std::time::Duration;
use tokio::sync::{mpsc, oneshot};
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                Some(job) = job_rx.recv() => {
                    println!("arbeite an: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("Signal zum Herunterfahren empfangen");
                    break;
                }
            }
        }
        done
    });

    for job in ["Foto verkleinern", "E-Mail senden", "Suchindex aktualisieren"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    sleep(Duration::from_millis(100)).await; // dem Worker Zeit zum Fertigwerden geben
    stop_tx.send(()).unwrap();

    let done = worker.await.unwrap();
    println!("der Worker hat {done} Aufgaben erledigt");
}

Ausgabe:

arbeite an: Foto verkleinern
arbeite an: E-Mail senden
arbeite an: Suchindex aktualisieren
Signal zum Herunterfahren empfangen
der Worker hat 3 Aufgaben erledigt

▶ Selbst ausprobieren

Ein paar Details, die du kennen solltest:

• Wir schreiben &mut stop_rx, nicht stop_rx. Bei jedem Schleifendurchlauf braucht select! denselben Empfänger wieder. Deshalb leihen wir ihn aus, statt ihn in die erste Runde zu verschieben (Move).
• Bei jedem Durchlauf erzeugt job_rx.recv() ein neues Future, und das unterlegene wird verworfen. Das ist in Ordnung, denn recv() ist cancel-safe (sicher beim Abbrechen): Wenn es verworfen wird, geht nie eine Nachricht verloren.
• Passt das Pattern nicht (hier: recv() gibt None zurück, weil alle Sender weg sind), schaltet select! diesen Zweig ab und wartet weiter auf die anderen.
• Sind mehrere Zweige im selben Moment bereit, wählt select! zufällig einen aus. So kann kein Zweig die Schleife für sich allein beanspruchen. Schreib biased; als erste Zeile in select!, damit die Zweige stattdessen von oben nach unten geprüft werden.

Cancel-Safety in einem Absatz. Weil der unterlegene Zweig verworfen wird, ist alles weg, was er schon halb erledigt hatte. Bei recv() auf einem Channel oder bei sleep() ist das harmlos: Es wurde ja noch nichts entnommen. Aber stell dir ein Future vor, das 100 Bytes von einem Netzwerk-Socket liest und schon 60 gelesen hat, als es verworfen wird. Diese 60 Bytes sind verloren. So ein Future ist nicht cancel-safe. Die tokio-Dokumentation sagt bei jeder Methode, ob sie cancel-safe ist. Wenn du select! in einer Schleife verwendest, prüf die Methoden, die du aufrufst. Oder erzeuge das Future einmal außerhalb der Schleife und polle immer wieder dasselbe Future (wie das geht, siehst du im Abschnitt über Pin).

Timeouts und Timer

tokio::time::timeout(duration, future) umhüllt ein beliebiges Future. Ist das Future rechtzeitig fertig, bekommst du Ok(value). Läuft die Zeit vorher ab, wird das Future verworfen (abgebrochen, genau wie ein unterlegener select!-Zweig), und du bekommst Err(Elapsed):

use std::time::Duration;
use tokio::time::{sleep, timeout};

async fn fetch_price(server_delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(server_delay_ms)).await;
    42
}

#[tokio::main]
async fn main() {
    let limit = Duration::from_millis(200);

    for (server, delay) in [("schneller Server", 20), ("langsamer Server", 2000)] {
        match timeout(limit, fetch_price(delay)).await {
            Ok(price) => println!("{server}: Preis ist {price}"),
            Err(e) => println!("{server}: {e}"),
        }
    }
}

Ausgabe:

schneller Server: Preis ist 42
langsamer Server: deadline has elapsed

▶ Selbst ausprobieren

„deadline has elapsed“ ist die englische Meldung, die tokio selbst für einen Elapsed-Fehler ausgibt. Sie heißt so viel wie „die Frist ist abgelaufen“.

Setz einen Timeout auf alles, was mit der Außenwelt redet: HTTP-Aufrufe, Datenbankabfragen, Lesen von einem Socket. Ein Server, der nie antwortet, sollte dein Programm nicht für immer einfrieren.

sleep oder interval? sleep wartet einmal. interval gibt dir einen gleichmäßigen Takt, wie ein Metronom. Du rufst .tick().await in einer Schleife auf, und jeder Tick kommt pünktlich nach Plan. Der erste Tick kommt sofort:

use std::time::Duration;
use tokio::time::interval;

#[tokio::main]
async fn main() {
    let mut heartbeat = interval(Duration::from_millis(50));

    for beat in 1..=3 {
        heartbeat.tick().await; // der erste Tick ist sofort fertig
        println!("Herzschlag {beat}");
    }
}

Ausgabe:

Herzschlag 1
Herzschlag 2
Herzschlag 3

▶ Selbst ausprobieren

Warum nicht einfach 50ms sleep in einer Schleife? Weil dann jede Runde „deine Arbeit + 50ms“ dauert und der Zeitplan langsam wegdriftet. Ein interval misst ab dem geplanten Start jedes Ticks. Dauert deine Arbeit 10ms, wartet es nur noch die restlichen 40ms. Nimm interval für Heartbeats, regelmäßiges Abfragen (Polling) und Aufgaben wie „jede Minute speichern“. Nimm sleep für eine einzelne Pause.

Streams: asynchrone Iteratoren

Ein Iterator gibt dir mit next() Werte nacheinander. Ein Stream ist die Async-Version davon: next().await gibt dir den nächsten Wert, sobald er bereit ist, und None, wenn keine mehr kommen. Denk an einen Börsenticker mit Aktienkursen, an Zeilen, die aus einer Datenbank eintreffen, oder an Nachrichten von einem Chat-Server.

Etwas Stream-Ähnliches hast du sogar schon benutzt: den Empfänger eines tokio-Channels. rx.recv().await wartet auf die nächste Nachricht und gibt None zurück, sobald alle Sender weg sind:

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);

    tokio::spawn(async move {
        for temp in [21, 22, 24, 23] {
            tx.send(temp).await.unwrap();
            sleep(Duration::from_millis(10)).await; // alle 10ms ein Messwert
        }
        // tx wird hier verworfen, und das schließt den Channel
    });

    while let Some(temp) = rx.recv().await {
        println!("Temperatur: {temp}°C");
    }
    println!("Sensor gestoppt");
}

Ausgabe:

Temperatur: 21°C
Temperatur: 22°C
Temperatur: 24°C
Temperatur: 23°C
Sensor gestoppt

▶ Selbst ausprobieren

Die Standardbibliothek hat noch keinen Stream-Trait (er wird noch entworfen). Das Ökosystem verwendet den Trait aus dem Crate futures, und tokios Hilfs-Crate tokio-stream bringt einen StreamExt-Trait mit den Methoden, die du von Iteratoren kennst: next(), map(), filter(), take() und weitere. Füg ihn zu deinem Projekt hinzu:

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-stream = "0.1"

use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    let prices = stream::iter(vec![100, 102, 98, 105, 110]);

    let mut doubled = prices.map(|p| p * 2).take(3);

    while let Some(p) = doubled.next().await {
        println!("{p}");
    }
}

Das gibt 200, 204 und 196 aus. Es liest sich genau wie Iterator-Code, bis auf das .await nach next(). Um den Empfänger eines tokio-Channels in einen echten Stream zu verwandeln, umhüllst du ihn: tokio_stream::wrappers::ReceiverStream::new(rx). (Das Crate futures hat ein eigenes StreamExt mit derselben Idee. Nimm pro Datei nur eines davon, denn beide Traits haben eine next()-Methode, und die kommen sich in die Quere.)

Pin, einfach erklärt

Früher oder später siehst du Pin in einem Typ oder einer Fehlermeldung. Hier ist die Idee – ohne die tiefe Theorie.

Wenn Rust eine async fn kompiliert, macht es daraus einen versteckten Struct, eine Zustandsmaschine (State Machine). Jede lokale Variable, die nach einem .await noch gebraucht wird, wird zu einem Feld dieses Structs. Stell dir jetzt vor, eine lokale Variable leiht sich eine andere aus, etwa let name = String::from("Ana"); let r = &name; mit einem .await dazwischen. Dann enthält der Struct einen Zeiger auf sein eigenes Feld. Würdest du den Struct an eine andere Stelle im Speicher verschieben, zeigte dieser Zeiger auf die alte, leere Stelle.

Pin ist die Lösung. Ein gepinnter Wert ist ein Versprechen: „Dieser Wert wird im Speicher nicht mehr verschoben.“ Sobald ein Future gepinnt ist, kann man es sicher pollen. Meistens merkst du davon gar nichts, denn .await pinnt das Future für dich.

Pin begegnet dir in drei Situationen:

1. Dasselbe Future mehrmals pollen, zum Beispiel eine Deadline in einer select!-Schleife.
2. Futures speichern, in einem Vec oder einem Struct-Feld, meistens als Pin<Box<dyn Future<Output = T>>>.
3. Rekursion in einer async fn.

Zum ersten Fall: Das passiert, wenn du ein sleep in einer Schleife wiederverwenden willst, ohne es zu pinnen:

let deadline = sleep(Duration::from_millis(300));

loop {
    tokio::select! {
        _ = &mut deadline => break,
        Some(n) = rx.recv() => println!("{n}"),
    }
}

error[E0277]: `PhantomPinned` cannot be unpinned
   |
   = note: consider using the `pin!` macro
           consider using `Box::pin` if you need to access the pinned value outside of the current scope
   = note: required for `Sleep` to implement `Unpin`

Die Meldung bedeutet: PhantomPinned (ein Teil von Sleep) kann nicht „entpinnt“ werden, also ist Sleep nicht Unpin. Der Compiler verrät dir sogar die Lösung. std::pin::pin! (oder tokio::pin!) pinnt das Future direkt an Ort und Stelle auf dem Stack. Jetzt teilen sich alle Runden der Schleife dieselbe Deadline:

use std::pin::pin;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);
    tokio::spawn(async move {
        for order in 1..=3 {
            tx.send(order).await.unwrap();
            sleep(Duration::from_millis(10)).await;
        }
        sleep(Duration::from_secs(5)).await; // danach wird es still im Laden
    });

    let mut deadline = pin!(sleep(Duration::from_millis(300)));

    loop {
        tokio::select! {
            _ = &mut deadline => {
                println!("Feierabend");
                break;
            }
            Some(order) = rx.recv() => println!("Bestellung #{order} erhalten"),
        }
    }
}

Ausgabe:

Bestellung #1 erhalten
Bestellung #2 erhalten
Bestellung #3 erhalten
Feierabend

▶ Selbst ausprobieren

Würden wir sleep(...) innerhalb von select! schreiben, würde jede neue Bestellung einen frischen 300ms-Timer starten – und ein gut besuchter Laden würde nie schließen. Eine einzige gepinnte Deadline außerhalb der Schleife löst das Problem.

Bei Rekursion muss der Compiler wissen, wie groß das Future ist. Ein Future, das sich selbst enthält, wäre aber unendlich groß. Ruft async fn countdown in sich selbst countdown(n - 1).await auf, bekommst du:

error[E0733]: recursion in an async fn requires boxing
  |
  = note: a recursive `async fn` call must introduce indirection such as `Box::pin` to avoid an infinitely sized future

Der Compiler sagt: Rekursion in einer async fn braucht eine Box, also eine Indirektion wie Box::pin, damit das Future nicht unendlich groß wird. Box::pin verschiebt das innere Future auf den Heap und pinnt es dort. Das äußere Future enthält jetzt nur noch einen Zeiger, und der hat eine feste Größe:

async fn countdown(n: u32) {
    if n == 0 {
        println!("Abheben!");
        return;
    }
    println!("{n}...");
    Box::pin(countdown(n - 1)).await;
}

#[tokio::main]
async fn main() {
    countdown(3).await;
}

Ausgabe:

3...
2...
1...
Abheben!

▶ Selbst ausprobieren

Ehrlich gesagt ist das alles, was die meisten brauchen: pin! für ein Future, das du mehrmals pollst, und Box::pin zum Speichern von Futures oder für Rekursion. Wenn du die ganze Geschichte wissen willst: Das Kapitel aus dem Rust-Buch in der Liste unter „Weiterlesen“ erklärt sie Schritt für Schritt.

async fn in Traits und asynchrone Closures

Seit Rust 1.75 kannst du async fn direkt in einen Trait schreiben. Seit Rust 1.85 gibt es außerdem asynchrone Closures, geschrieben als async |x| ..., mit den passenden Traits AsyncFn, AsyncFnMut und AsyncFnOnce (Async-Versionen der Closure-Traits). Hier beides in einem Beispiel:

use std::time::Duration;
use tokio::time::sleep;

trait Notifier {
    async fn notify(&self, user: &str) -> String;
}

struct Email;

impl Notifier for Email {
    async fn notify(&self, user: &str) -> String {
        sleep(Duration::from_millis(10)).await; // tu so, als würden wir mit einem Mailserver reden
        format!("E-Mail an {user} gesendet")
    }
}

async fn for_each_user(users: &[&str], job: impl AsyncFn(&str) -> String) {
    for user in users {
        println!("{}", job(user).await);
    }
}

#[tokio::main]
async fn main() {
    let email = Email;
    let users = ["ana", "ben"];

    for_each_user(&users, async |user: &str| email.notify(user).await).await;
}

Ausgabe:

E-Mail an ana gesendet
E-Mail an ben gesendet

▶ Selbst ausprobieren

Bei async fn in Traits gibt es einen Haken. Der Trait verspricht nicht, dass das zurückgegebene Future Send ist, und tokio::spawn braucht Send-Futures (sie können zwischen Threads wandern). Deshalb scheitert es, einen Aufruf über ein generisches N: Notifier zu spawnen:

fn send_in_background<N: Notifier + Send + Sync + 'static>(notifier: N) {
    tokio::spawn(async move {
        let msg = notifier.notify("ana").await;
        println!("{msg}");
    });
}

error: future cannot be sent between threads safely
  |
7 |         let msg = notifier.notify("ana").await;
  |                   ^^^^^^^^^^^^^^^^^^^^^^ await occurs here on type `impl Future<Output = String>`, which is not `Send`
  |
help: `Send` can be made part of the associated future's guarantees for all implementations of `Notifier::notify`
  |
2 -     async fn notify(&self, user: &str) -> String;
2 +     fn notify(&self, user: &str) -> impl std::future::Future<Output = String> + Send;

Die Fehlermeldung heißt: Das Future kann nicht sicher zwischen Threads verschickt werden, weil der Typ impl Future<Output = String> nicht Send ist. Der Vorschlag des Compilers ist die übliche Lösung: Schreib die Trait-Methode als fn ... -> impl Future<Output = String> + Send. Wer den Trait implementiert, kann trotzdem eine ganz normale async fn schreiben. Beachte außerdem: Traits mit async fn lassen sich noch nicht als dyn Notifier-Trait-Objekte verwenden. Wenn du das brauchst, packt das beliebte Crate async-trait die Futures für dich in eine Box.

Häufige Fehler

Die Runtime blockieren. std::thread::sleep, aufwendige Berechnungen oder große std::fs-Lesevorgänge in einem Async-Task frieren den ganzen Worker-Thread ein, und jeder andere Task darauf muss warten. Die Grundlagenseite hat die sleep-Variante gezeigt. Rechenintensive oder blockierende Arbeit gibst du an tokio::task::spawn_blocking ab. Das führt sie auf einem eigenen Pool von Threads aus, der genau dafür gedacht ist:

fn count_primes(limit: u64) -> usize {
    (2..limit)
        .filter(|&n| (2..).take_while(|d| d * d <= n).all(|d| n % d != 0))
        .count()
}

#[tokio::main]
async fn main() {
    let handle = tokio::task::spawn_blocking(|| count_primes(50_000));

    println!("wir beantworten weiter Anfragen, während wir zählen ...");

    let primes = handle.await.unwrap();
    println!("es gibt {primes} Primzahlen unter 50000");
}

Ausgabe:

wir beantworten weiter Anfragen, während wir zählen ...
es gibt 5133 Primzahlen unter 50000

▶ Selbst ausprobieren

Einen Guard von std::sync::Mutex über ein .await hinweg halten. Das sieht harmlos aus:

let handle = tokio::spawn(async move {
    let mut count = visits.lock().unwrap();
    *count += 1;
    save_to_disk(*count).await;
});

error: future cannot be sent between threads safely
   |
   = help: within `{async block@src/main.rs:13:31: 13:41}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, u32>`
note: future is not `Send` as this value is used across an await
   |
14 |         let mut count = visits.lock().unwrap();
   |             --------- has type `std::sync::MutexGuard<'_, u32>` which is not `Send`
15 |         *count += 1;
16 |         save_to_disk(*count).await;
   |                              ^^^^^ await occurs here, with `mut count` maybe used later

Der Compiler sagt: Das Future ist nicht Send, weil der MutexGuard über ein .await hinweg benutzt wird. Und damit schützt er dich. Während dieser Task am .await schläft, würde er die Sperre (Lock) immer noch halten. Jeder andere Task, der die Sperre haben will, würde seinen ganzen Thread blockieren. Die Lösung: Gib den Guard vor dem .await frei, indem du das Sperren in einen eigenen Block packst:

let count = {
    let mut count = visits.lock().unwrap();
    *count += 1;
    *count
}; // der Guard wird hier verworfen
save_to_disk(count).await;

Wenn du eine Sperre wirklich über ein .await hinweg halten musst, nimm tokio::sync::Mutex. Dessen lock().await wartet, ohne den Thread zu blockieren. (Mehr über Sperren erfährst du unter Geteilter Zustand.)

.await vergessen. Futures sind „faul“: save_file(); ohne .await tut also überhaupt nichts. Der Compiler warnt dich mit „futures do nothing unless you .await or poll them“ (Futures tun nichts, solange du sie nicht mit .await erwartest oder pollst). Die ganze Geschichte findest du unter Async/Await – und ignorier diese Warnung nicht.

Einen Task starten und nie auf ihn warten. Wenn main zurückkehrt, fährt die Runtime herunter, und jeder unfertige Task wird verworfen:

use std::time::Duration;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        println!("Bericht gespeichert"); // wird nie ausgegeben!
    });

    println!("main ist fertig");
}

Ausgabe:

main ist fertig

▶ Selbst ausprobieren

Der Bericht wird nie gespeichert. Wenn die Arbeit wichtig ist, behalte das JoinHandle und warte mit .await darauf (oder leg die Tasks in ein JoinSet und leere es), bevor main endet.

Übungen

1. Preisvergleich mit Frist

Vier Läden beantworten eine Preisanfrage nach 20, 40, 300 und 60 Millisekunden. Frag alle vier gleichzeitig mit einem JoinSet an, aber gib jedem Laden nur 150ms Zeit. Gib die Preise aus, die angekommen sind (sortiert nach dem Namen des Ladens), und danach, wie viele Läden zu langsam waren. Tipp: Spawne timeout(limit, ask_shop(name, delay)) in das Set. Jeder fertige Task liefert dir dann ein Ok oder ein Err(Elapsed). Gib also den Namen des Ladens zusammen mit diesem Ergebnis zurück.

Lösung anzeigen

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{sleep, timeout};

async fn ask_shop(delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(delay_ms)).await;
    100 + delay_ms as u32 // ein ausgedachter Preis
}

#[tokio::main]
async fn main() {
    let shops = [("Tante-Emma-Laden", 20), ("MegaMart", 40), ("SlowCo", 300), ("Krimskrams", 60)];
    let limit = Duration::from_millis(150);

    let mut set = JoinSet::new();
    for (name, delay) in shops {
        set.spawn(async move { (name, timeout(limit, ask_shop(delay)).await) });
    }

    let mut prices = Vec::new();
    let mut timed_out = 0;
    while let Some(res) = set.join_next().await {
        match res.unwrap() {
            (name, Ok(price)) => prices.push((name, price)),
            (_, Err(_)) => timed_out += 1,
        }
    }

    prices.sort();
    for (name, price) in prices {
        println!("{name}: {price}");
    }
    println!("{timed_out} Laden/Läden zu langsam");
}

Ausgabe:

Krimskrams: 160
MegaMart: 140
Tante-Emma-Laden: 120
1 Laden/Läden zu langsam

▶ Selbst ausprobieren

Alle Läden warten gleichzeitig. Die ganze Abfrage dauert also etwa 150ms, egal wie langsam SlowCo ist. Wenn dessen Timeout zuschlägt, wird sein Future verworfen. Es kann also nicht im Hintergrund weiterlaufen.

2. Ordnergrößen mit asynchroner Rekursion

Ein Folder hat einen Namen, eine Liste von Dateigrößen in KB und eine Liste von Unterordnern. Schreib async fn total_size(folder: &Folder) -> u64, die die eigenen Dateien plus die Größe jedes Unterordners zusammenzählt und sich dafür für jeden Unterordner selbst aufruft. Tu so, als bräuchte jeder Ordner vorher einen kurzen asynchronen „Festplattenzugriff“, mit tokio::task::yield_now().await. Tipp: Um den rekursiven Aufruf gehört ein Box::pin(...), sonst bekommst du den Fehler E0733.

Lösung anzeigen

struct Folder {
    name: &'static str,
    files_kb: Vec<u64>,
    subfolders: Vec<Folder>,
}

async fn total_size(folder: &Folder) -> u64 {
    tokio::task::yield_now().await; // tu so, als würden wir von der Festplatte lesen

    let mut total: u64 = folder.files_kb.iter().sum();
    for sub in &folder.subfolders {
        total += Box::pin(total_size(sub)).await;
    }
    println!("{}: {} KB", folder.name, total);
    total
}

#[tokio::main]
async fn main() {
    let home = Folder {
        name: "home",
        files_kb: vec![4],
        subfolders: vec![
            Folder { name: "fotos", files_kb: vec![2048, 1024], subfolders: vec![] },
            Folder {
                name: "code",
                files_kb: vec![12, 8],
                subfolders: vec![Folder { name: "target", files_kb: vec![500], subfolders: vec![] }],
            },
        ],
    };

    let total = total_size(&home).await;
    println!("gesamt: {total} KB");
}

Ausgabe:

fotos: 3072 KB
target: 500 KB
code: 520 KB
home: 3596 KB
gesamt: 3596 KB

▶ Selbst ausprobieren

Jeder Ordner wird erst nach all seinen Unterordnern ausgegeben, denn er muss zuerst auf deren Größen warten. Box::pin legt das Future jedes verschachtelten Aufrufs auf den Heap, damit das äußere Future eine feste Größe behält.

3. Erst die Warteschlange leeren, dann aufhören

Ändere das Beispiel zum Herunterfahren so, dass es das sleep(100ms) vor dem Stoppsignal nicht mehr braucht. Schick die drei Aufgaben und das Stoppsignal sofort, aber sorg dafür, dass der Worker trotzdem alle wartenden Aufgaben erledigt, bevor er aufhört. Tipp: Ohne Hilfe wählt select! einen zufälligen Zweig, wenn beide bereit sind. Füg biased; hinzu und setz den Zweig für die Aufgaben an die erste Stelle.

Lösung anzeigen

use tokio::sync::{mpsc, oneshot};

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    for job in ["Foto verkleinern", "E-Mail senden", "Suchindex aktualisieren"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    stop_tx.send(()).unwrap();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                biased;
                Some(job) = job_rx.recv() => {
                    println!("arbeite an: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("Warteschlange ist leer, ich höre auf");
                    break;
                }
            }
        }
        done
    });

    let done = worker.await.unwrap();
    println!("der Worker hat {done} Aufgaben erledigt");
}

Ausgabe:

arbeite an: Foto verkleinern
arbeite an: E-Mail senden
arbeite an: Suchindex aktualisieren
Warteschlange ist leer, ich höre auf
der Worker hat 3 Aufgaben erledigt

▶ Selbst ausprobieren

Mit biased; probiert select! immer zuerst den Zweig für die Aufgaben. Das Stoppsignal schaut es sich erst an, wenn recv() nichts mehr bereit hat. Ohne biased; könnte der Worker schon nach ein oder zwei Aufgaben aufhören – ein Bug, der nur ab und zu auftaucht.

Zusammenfassung

• join! wartet auf alle Futures, try_join! hört beim ersten Err auf, und JoinSet führt viele gespawnte Tasks aus und gibt dir die Ergebnisse, sobald sie fertig sind (sortiere sie, wenn die Reihenfolge wichtig ist).
• select! wartet darauf, dass das erste Future fertig wird. Die anderen werden verworfen – so funktioniert Abbrechen in Async-Rust. Nimm in einer Schleife cancel-safe Futures und biased;, wenn die Reihenfolge zählt.
• timeout gibt ein langsames Future auf. interval tickt in einem festen Takt, sleep wartet einmal.
• Ein Stream ist ein asynchroner Iterator. Der Empfänger eines Channels funktioniert wie einer, und tokio-stream bringt map, take und Co. mit.
• Pin verspricht, dass ein Future im Speicher nicht verschoben wird. Nimm pin!, um ein Future mehrmals zu pollen, und Box::pin für gespeicherte Futures und asynchrone Rekursion.
• Blockiere die Runtime nicht (spawn_blocking), halte keine std-Sperre über ein .await hinweg, und warte auf die Tasks, die du startest.

Kurzer Check

Kurzer Check

1. Was passiert bei tokio::select! mit den Zweigen, die nicht als Erste fertig werden?

Sie laufen im Hintergrund weiter Sie werden verworfen, hören also auf und räumen auf Sie laufen in der nächsten Schleifenrunde noch einmal

2. Was gibt tokio::time::timeout zurück, wenn die Zeit abläuft?

Es löst eine Panic aus Ok mit einem Standardwert Err(Elapsed), und das innere Future wird verworfen

3. Warum brauchst du Box::pin, wenn eine async fn sich selbst aufruft?

Ein Future, das sich selbst enthält, wäre unendlich groß Rekursion ist in Async-Code grundsätzlich verboten Damit die Funktion auf einem anderen Thread läuft

4. Du musst in einem tokio-Programm 2 Sekunden lang rechenintensive Arbeit erledigen. Was ist das beste Werkzeug?

std::thread::sleep drumherum tokio::select! Ein JoinSet mit tokio::spawn tokio::task::spawn_blocking

Punkte: 0 / 4

Weiterlesen

• Select – das Tokio-Tutorial (auf Englisch): select! im Detail – wie die Zweige funktionieren, was Abbrechen bedeutet und wie du es in einer Schleife einsetzt.
• Streams – das Tokio-Tutorial (auf Englisch): wie du Streams mit tokio-stream benutzt, anpasst und selbst baust.
• Ein genauerer Blick auf die Traits für Async – Das Rust-Buch: eine behutsame Schritt-für-Schritt-Erklärung von Future, Pin, Unpin und Stream.
• std::pin – die Rust-Standardbibliothek (auf Englisch): die offizielle Erklärung zu Pinning, Pin, Unpin und dem Makro pin!.
• tokio::time auf docs.rs (auf Englisch): sleep, timeout, interval und wie verpasste Ticks behandelt werden.