1. Il problema del buffer: producer piu veloce del consumer
Quando scrivi per la prima volta AsyncStream, sembra quasi magia: fai “yield” dei valori e, da qualche altra parte nel codice, arrivano in modo elegante nel for await. Ma non appena introduci un timing reale (rete, operazioni su disco, stampa in console), emerge un fatto spiacevole: producer e consumer vivono a velocita diverse. Il producer puo “sfornare” eventi rapidamente, mentre il consumer puo elaborarli lentamente.
Immagina che il producer sia un corriere che porta notifiche, e che il consumer sia tu che le leggi. Se il corriere porta una notifica ogni millisecondo e tu ne leggi una al secondo, da qualche parte si dovra formare una pila di fogli non letti. Quella pila e il buffer. In AsyncStream il buffer conserva i valori gia pubblicati ma non ancora consumati dall’iterazione: “bufferizzazione solo per i valori che non sono ancora stati consumati” e una parte importante del modello.
Se non limiti il buffer, puoi ottenere due effetti contemporaneamente: l’app “sembra funzionare”, ma la memoria cresce come la lista delle cose da fare di uno studente prima degli esami. Per questo AsyncStream ha una politica di bufferizzazione: quanti elementi si possono tenere e cosa fare in caso di overflow.
Per vedere il problema in modo intuitivo, costruiamo una situazione volutamente “cattiva”: producer molto veloce, consumer molto lento (per esempio, stampa in console con ritardo).
import Foundation
func fastInts() -> AsyncStream<Int> {
AsyncStream(bufferingPolicy: .unbounded) { c in
Task {
for i in 1...10_000 { c.yield(i) }
c.finish()
}
}
}
Il codice e corretto, ma se il consumer esegue un’elaborazione pesante per ogni elemento, il flusso puo accumulare molti elementi. E non e un bug di AsyncStream: e il prezzo onesto del “ho ricevuto l’evento, ma non l’ho ancora elaborato”.
2. Le politiche di bufferizzazione di AsyncStream
Ora che abbiamo una “pila di fogli”, vogliamo metterci d’accordo: quanti fogli siamo disposti a tenere sul tavolo, e quali buttare se il tavolo non basta piu. Questo e cio che descrive AsyncStream.Continuation.BufferingPolicy. Questa politica ha tre varianti principali: .unbounded, .bufferingOldest(n), .bufferingNewest(n). In sostanza, sono modalita di overflow del buffer.
.unbounded significa “buffer senza limiti”. E semplice, comodo per gli esempi didattici e per eventi rari, ma pericoloso per eventi frequenti: se il consumer resta indietro, la memoria cresce.
.bufferingOldest(n) significa “quando il buffer e pieno, i nuovi elementi vengono scartati”. In altre parole, garantiamo di conservare cio che e gia stato accumulato (cio che e piu vecchio nella coda) fino alla dimensione n. E utile quando conta la “storia” e preferisci perdere gli eventi nuovi piuttosto che buttare via cio che hai gia raccolto.
.bufferingNewest(n) significa “quando il buffer e pieno, scartiamo l’elemento piu vecchio dal buffer e inseriamo quello nuovo”. In altre parole, garantiamo che nel buffer restino gli ultimi n eventi piu recenti. Questo e spesso perfetto per “stato” e “progresso”, quando ti interessa di piu la situazione attuale che ogni micro-passaggio.
Riassumiamo tutto in una piccola tabella, cosi il cervello fa meno fatica:
| Politica | Cosa conserva il buffer in caso di overflow | Cosa perdiamo | Significato tipico |
|---|---|---|---|
|
Tutto (senza limiti) | Niente, ma rischiamo la memoria | Eventi rari, esempi didattici |
|
Gli elementi vecchi (fino a n) | Gli eventi nuovi | “La storia conta piu della freschezza” |
|
Gli elementi piu recenti (fino a n) | Gli eventi vecchi | “Conta lo stato attuale” |
Un altro dettaglio: se specifichi 0, il flusso diventa di fatto “dropper”: il valore verra scartato se in quel momento nessuno aspetta next() (in breve: “nessuna domanda, nessuna consegna”). Come strumento tecnico, a volte puo essere utile.
3. bufferingNewest e bufferingOldest: come scegliere
In pratica, la scelta della politica non e “quale e piu veloce”, ma “quale e semanticamente piu corretta”. E come decidere cosa mostrare all’utente: tutte le percentuali intermedie di caricamento oppure solo l’ultimo stato. Aiuta una semplice domanda: un evento e un “fatto” o un “istantanea dello stato”?
Se un evento e un “fatto”, perderlo cambia il significato. Per esempio: “un libro e stato aggiunto alla libreria”, “l’operazione di eliminazione e stata completata”, “e stata ricevuta una risposta del server con i dati”. Di solito questi eventi non si possono perdere, altrimenti una parte della logica verrebbe saltata. Per questi casi serve un buffer sufficientemente grande, oppure un design del flusso del tutto diverso (in questa lezione non entriamo nelle strategie avanzate di coordinamento delle velocita: ci limitiamo a fissare che, per i fatti, la perdita e dolorosa). In questi casi .bufferingOldest puo sembrare “piu sicuro”, perche non scarta cio che e gia stato accumulato, ma scartera i nuovi fatti, e anche questo puo essere inaccettabile. Quindi, piu spesso, per i fatti si aumenta il limite, si rende il consumer piu veloce oppure si riduce la frequenza degli eventi.
Se un evento e un “istantanea dello stato”, perdere parte delle istantanee di solito non e un problema. Progresso 3/100, poi 4/100, poi 5/100 — questo e un tipico “snapshot”. Se l’utente vede subito 50/100 e i passi intermedi spariscono, il significato non si rompe: stiamo comunque andando avanti. E proprio per questo il progresso spesso si abbina a .bufferingNewest(1) o .bufferingNewest(2).
Per fissare la differenza, immaginiamo il buffer come una piccola coda di tre posti:
flowchart LR
P[producer yield] --> B[(buffer da 3 elementi)]
B --> C[consumer for await]
subgraph Overflow["overflow"]
O1["bufferingOldest(3):
scartiamo il nuovo evento"]
O2["bufferingNewest(3):
spingiamo fuori il vecchio evento"]
end
In bufferingNewest e come se dicessi: “non mi serve tutta la coda dei messaggi, mi serve l’ultima informazione aggiornata”. In bufferingOldest dici: “dato che ho gia iniziato a raccogliere la cronologia, non toccatela, meglio non accettare nuovi eventi”.
4. yield(...) e YieldResult: cosa e successo all’evento
Un errore comune dei principianti nell’asincronia e pensare che yield “garantisca la consegna”. In realta yield garantisce qualcos’altro: “ho provato a consegnare un valore al flusso, ecco il risultato del tentativo”. Per questo yield restituisce YieldResult, che puo segnalare che il valore e finito nel buffer (o ha svegliato subito un next() in attesa), oppure che e stato scartato perche il buffer era pieno, oppure che il flusso e gia in stato terminale.
AsyncStream.Continuation.YieldResult ha tre casi:
- .enqueued(remaining: Int) — il valore e stato accettato, e remaining mostra quanti slot nel buffer erano rimasti al momento del yield (utile per la diagnostica; non conviene costruirci sopra una logica complessa, se non sei sicuro dell’esclusivita delle chiamate).
- .dropped(Element) — il valore non e entrato nel buffer a causa dell’overflow, e ti viene restituito “chi” e stato scartato (e informativo sia con .bufferingOldest sia con .bufferingNewest).
- .terminated — il flusso e gia terminato (tramite finish() o a causa della cancellazione), e accettare nuovi valori non ha piu senso.
Mini-esempio che stampa “cosa e successo” all’evento:
import Foundation
func debugYield(_ value: Int, into c: AsyncStream<Int>.Continuation) {
switch c.yield(value) {
case .enqueued(let remaining):
print("enqueued, remaining:", remaining) // ad esempio: enqueued, remaining: 7
case .dropped(let dropped):
print("dropped:", dropped) // dropped: 42
case .terminated:
print("terminated") // terminated
}
}
Importante: yield restituisce subito e non “aspetta” che il consumer elabori l’elemento. E parte del contratto di AsyncStream: adatta sistemi senza backpressure (per esempio, callback), quindi il “fornitore” non deve sincronizzarsi in modo sincrono con il “consumatore”. Il buffer e la politica di scarto sono un compromesso tra i mondi “mi hanno segnalato un evento” e “sono riuscito a elaborare l’evento”.
5. Pratica: il progresso come stream e bufferingNewest(1)
Torniamo ora alla nostra applicazione CLI didattica LibraryCLI. Il progresso e il luogo classico in cui il producer “spara” spesso, mentre il consumer (la stampa in console) lavora lentamente, perche stdout non e infinitamente veloce.
Definiamo un evento di progresso, il piu semplice possibile e tipizzato:
enum ProgressEvent {
case started(total: Int)
case advanced(done: Int, total: Int)
case completed
}
Ora costruiamo uno stream di progresso che pubblica eventi spesso, per esempio a ogni passo. E qui scegliamo consapevolmente .bufferingNewest(1): ci interessa mostrare “l’ultimo progresso”, non stampare ogni millisecondo 1/100, 2/100, 3/100… mentre l’utente e gia stanco di vivere.
import Foundation
func makeProgressStream(total: Int) -> AsyncStream<ProgressEvent> {
AsyncStream(bufferingPolicy: .bufferingNewest(1)) { c in
Task {
c.yield(.started(total: total))
for done in 1...total { c.yield(.advanced(done: done, total: total)) }
c.yield(.completed)
c.finish()
}
}
}
Il senso di .bufferingNewest(1) qui e soprattutto UX: se il consumer (la stampa) non riesce a tenere il passo, non vogliamo accumulare storia, vogliamo avere sempre il valore piu recente.
Consumer che “renderizza” l’evento in una stringa:
func render(_ e: ProgressEvent) -> String {
switch e {
case .started(let total): return "Start: 0/\(total)"
case .advanced(let done, let total): return "Progress: \(done)/\(total)"
case .completed: return "Done"
}
}
E il consumo tramite for await:
import Foundation
func showProgress() async {
for await event in makeProgressStream(total: 5) {
print(render(event))
}
print("Report: completed") // Report: completed
}
Questi frammenti sono volutamente piccoli: l’obiettivo e percepire il meccanismo del buffer, non affogare nell’infrastruttura della CLI.
6. Dettagli utili: unbounded e diagnostica delle perdite
Perche .unbounded e pericoloso per eventi frequenti
E molto facile iniziare con .unbounded, perche “lasciamo pure che tutto si accumuli, la memoria e grande”. Poi per errore crei uno stream di eventi che pubblica piu rapidamente di quanto tu riesca a stampare o elaborare, e l’app comincia a comportarsi in modo strano: l’output resta indietro di secondi, poi di minuti, e infine stampi il progresso 1% quando l’operazione e gia terminata. Nel frattempo, la memoria puo crescere gradualmente.
AsyncStream con questa impostazione bufferizza davvero gli elementi finche il consumer non li prende — ed e prevedibile per molte API adattabili (notifiche, scritture su DB e cosi via). Ma “prevedibile” non significa “sempre buono”. Per il progresso e per gli indicatori simili all’UI non serve conservare tutto.
Se vuoi vedere l’effetto in miniatura, puoi rallentare artificialmente il consumer:
import Foundation
func slowConsumer() async {
for await e in makeProgressStream(total: 50) {
print(render(e))
try? await Task.sleep(nanoseconds: 50_000_000) // 0.05s
}
}
Se il producer ha “sfornato” tutti i 50 eventi all’istante, il consumer li stampera con ritardo. Con .bufferingNewest(1) stamperai valori piu aggiornati (e ne perderai una parte), mentre con .unbounded stamperai tutto in modo onesto — ma con un enorme ritardo.
Logghiamo .dropped, ma senza panico
Quando abiliti bufferingNewest o bufferingOldest, devi accettare in anticipo che alcuni eventi andranno persi. Non e una tragedia se gli eventi, per significato, sono “perdibili” (progresso, aggiornamento di stato). Ma a volte e utile diagnosticare: “stiamo davvero perdendo eventi oppure il consumer riesce a star dietro?”.
Qui torna utile YieldResult. Aggiungiamo una versione del producer che conta quante volte qualcosa e stato scartato.
import Foundation
func makeDebugProgress(total: Int) -> AsyncStream<ProgressEvent> {
AsyncStream(bufferingPolicy: .bufferingNewest(1)) { c in
Task {
var droppedCount = 0
for done in 1...total {
if case .dropped = c.yield(.advanced(done: done, total: total)) {
droppedCount += 1
}
}
print("dropped:", droppedCount) // dropped: ...
c.finish()
}
}
}
E un altro dettaglio: la continuation e thread-safe, e le chiamate a yield/finish sono serializzate, ma se chiami yield da piu punti concorrenti, l’ordine di consegna puo diventare inatteso. In questa lezione non costruiamo producer concorrenti complessi, ma come “odore di problema” e bene ricordarlo: un solo producer-Task e piu semplice da capire e da debuggare.
7. Errori tipici
Errore n. 1: scegliere la politica “a caso”, senza definire il significato dell’evento.
Quando uno sviluppatore imposta .bufferingNewest(1) “perche e piu corto”, puo perdere per sbaglio elementi davvero importanti (per esempio, eventi della logica di dominio). La politica del buffer non e un’ottimizzazione, ma una parte del significato dello stream: per il progresso e per lo stato va bene “teniamo l’ultimo”, mentre per i fatti spesso serve un altro approccio, dove la perdita e inammissibile.
Errore n. 2: aspettarsi che yield garantisca la consegna di ogni elemento.
yield puo restituire .dropped(...) o .terminated, e questa e una parte normale dell’API: AsyncStream ha un limite di buffer e uno stato terminale. Se scrivi il codice come se yield “consegnasse” sempre, sotto carico il tuo consumer iniziera a vedere qualcosa di diverso da cio che ti aspettavi, e sembrera tutto “random”.
Errore n. 3: lasciare .unbounded su uno stream di eventi frequenti e stupirsi per memoria e lag.
.unbounded e comodo finche gli eventi sono pochi, ma appena il producer diventa piu veloce del consumer, ottieni accumulo. Questo puo manifestarsi non solo nella memoria, ma anche nell’UX: l’output inizia a “inseguire il passato” e a essere in ritardo rispetto alla realta. Per il progresso e lo stato, di solito e piu ragionevole limitare il buffer.
Errore n. 4: confondere bufferingOldest e bufferingNewest nel significato.
Il nome si puo confondere facilmente, soprattutto alla fine di una giornata pesante. Qui aiuta l’autoverifica: .bufferingNewest(n) significa “nel buffer restano i piu recenti n”, quindi i vecchi vengono spinti fuori; .bufferingOldest(n) significa “nel buffer restano i piu vecchi n”, quindi i nuovi vengono scartati in caso di overflow.
Errore n. 5: continuare a fare yield dopo finish() e aspettarsi che il consumer lo veda.
Dopo finish() il flusso e in stato terminale, e i successivi yield non hanno piu un significato normale: potresti vedere .terminated. In una buona architettura decidi in anticipo chi chiude lo stream e, dopo la chiusura, non “pubblichi” piu eventi al suo interno.
GO TO FULL VERSION