Les Streams
Lecture 21 min. âą
Table des matiĂšres
- La boucle for
- Le trait Iterator
- Le trait IntoIterator
- Le trait Future
- Les Runtimes
- Des futures de futures
- Les sucres syntaxiques async/.await
- Fonctions asynchrones
- Itérer sur des futures
- Itérateur asynchrone
- le trait Stream et lâĂ©cosystĂšme asynchrone
- API Haut niveau
- Conclusion
Bonjour Ă toutes et Ă tous đ
La boucle for
Lorsque jâai dĂ©couvert les boucles en Rust, je suis tombĂ© sur cet Ă©trange bout de code.
for i in 0..4 {
println!("{}", i)
} Lorsque lâexĂ©cute, on obtient
0
1
2
3 Le moi de lâĂ©poque nâa pensĂ© quâune chose : âMais câest quoi cette magie noire!! đ±â
Mais nous sommes en Rust et en Rust, presque tout est explicite. đŠ
Il suffit, de dĂ©rouler le fil. đ§¶
Voici le vrai code qui se déroule:
let x = ???;
loop {
if let Some(i) = x.next() {
println!("{}", i)
} else {
break
}
}Nous allons détailler ce que x représente dans un instant.
Pour lâinstant il faut comprendre que next retourne un Option<i32>.
Tant que lâon a pas atteint i = 3, on obtient un Some(i). Puis un None.
Ce que lâon peut redĂ©finir en
x.next() // Some(0)
x.next() // Some(1)
x.next() // Some(2)
x.next() // Some(3)
x.next() // NoneBon maintenant que nous avons le cahier des charges du comportement voulu par notre x, créons-le.
Nous savons quâil se doit de poossĂ©der un Ă©tat interne.
struct X {
state: i32
}Il possÚde également une méthode next() -> Option<i32>.
Implémentons-la
impl X {
fn next(& mut self) {
if self.state < 4 {
let i = self.state;
self.state +=1;
Some(i)
} else {
None
}
}
}A chaque fois que lâon appelle next(), on vĂ©rifie le state,
- si lâon est dans lâintervalle $[0; 3]$, on incrĂ©ment le
stateet on renvoie unSome(i) - si celui excĂšde $3$, on renvoie
None
On peut alors utiliser notre super code
let mut x = X { state: 0 };
loop {
if let Some(i) = x.next() {
println!("{}", i)
} else {
break;
}
}Qui nous affiche comme prévu
0
1
2
3Victory ! đ
Le trait Iterator
Du coup faisons une petite expĂ©rience đ§Ș
let mut x = X { state: 0 };
for i in x {
println!("{i}")
}Nop ! đ„
error[E0277]: `X` is not an iterator
|
| for i in x {
| ^ `X` is not an iterator
|
= help: the trait `Iterator` is not implemented for `X`Quand je vous disais que ça se passait en réalité comme ça je vous ai à demi menti.
Afin de normaliser les choses, Rust sâappuie de maniĂšre massive sur les traits.
Ici le compilateur nous signale que X nâest pas un itĂ©rateur. parce quâil nâimplĂ©mente pas le trait Iterator.
Mais du coup cela signifie que la boucle for peut se résumer à :
for i in #quelque chose qui implémente Iterator# {
}Implémentons Iterator pour X.
Voici la coquille vide de son implémentation
impl Iterator for X {
type Item = ();
fn next(&mut self) -> Option<Self::Item> {
todo!()
}
}Ici deux choses nous intéresse:
La ligne
type Item = ()on appelle ça un associated type en Rust, cela permet de dĂ©finir un type prĂ©sent dans toute lâimplĂ©mentation du trait.
On le retrouve dâailleurs dans
Option<Self::Item>La deuxiĂšme chose intĂ©ressante câest que
fn next(&mut self) -> Option<Self::Item> {}Ressemble trĂšs fortement Ă notre
fn next(&mut self) -> Option<i32> {}Par conséquent on peut remplir par mimétisme le
type Item = i32Et implémenter le reste:
impl Iterator for X {
type Item = i32;
fn next(&mut self) -> Option<Self::Item> {
if self.state < 4 {
let i = self.state;
self.state +=1;
Some(i)
} else {
None
}
}
}Ce qui nous permet de lâutiliser dans notre boucle for
let x = X { state: 0 };
for i in x {
println!("{i}")
}Cool đ
Mais que se passe-t-il si on tente de redemander un tour de boucle ?
let x = X { state: 0 };
for i in x {
println!("{i}")
}
for i in x {
println!("{i}")
}Réponse une erreur, enfin plusieurs:
error[E0382]: use of moved value: `x`
--> src\main.rs:39:14
|
| let x = X { state: 0 };
| - move occurs because `x` has type `X`, which does not implement the `Copy` trait
| for i in x {
| - `x` moved due to this implicit call to `.into_iter()`
...
| for i in x {
| ^ value used here after move
|
note: `into_iter` takes ownership of the receiver `self`, which moves `x`
|
| fn into_iter(self) -> Self::IntoIter;La premiĂšre est que x a Ă©tĂ© dĂ©placĂ©, et ce dĂ©placement a Ă©tĂ© provoquĂ© par for qui implicitement (ouais câest moche pour le coup đ«) Ă©crit ce code
for i in x.into_iter() {}Sauf que into_iter consomme ce que lâon lui donne Ă manger.
fn into_iter(self)Donc cela signifie quâavant mĂȘme que la boucle for ne dĂ©bute ses appels Ă x.next(). x nâexiste dĂ©jĂ plus ⊠đ
Ainsi quand on tente de rĂ©appliquer le x.into_iter(), nous nâavons plus de x pour bosser!
Ce comportement bien que perturbant est trĂšs intĂ©ressant car il empĂȘche dâutiliser un itĂ©rateur qui a dĂ©jĂ Ă©tĂ© utilisĂ© et donc consommĂ©.
La deuxiĂšme erreur câest que lâon ne possĂšde pas le trait Copy sur X, donc comme le compilo peut rien faire il dĂ©place x, mais sâil est Copy alors il pourra faire quelque chose.
Pour implémenter Copy, nous devons implémenter Clone, nous passons par les derives pour nous simplifier le travail.
#[derive(Copy, Clone)]
struct X {
state: i32,
}Ainsi, on peut maintenant rĂ©utiliser x autant de fois que lâon dĂ©sire.
let x = X { state: 0 };
for i in x {
println!("{i}")
}
for i in x {
println!("{i}")
}Comme le trait Copy est appelé pour chacune des boucles comme ceci:
let x = X { state: 0 };
let x_copy1 = x; // x_copy1 = X { state: 0 } -- copie de x en x_copy1
let iterator1 = x_copy1.into_iter(); // x_copy1 est consommé par into_iter mais pas x
for i in iterator1 {
println!("{i}")
}
let x_copy2 = x; // x_copy2 = X { state: 0 } -- copie de x en x_copy2
let iterator2 = x_copy2.into_iter(); // x_copy2 est consommé par into_iter mais pas x
for i in iterator2 {
println!("{i}")
}Le x nâest jamais modifiĂ© puisque copiĂ© en lâĂ©tat et donc state = 0 pour chaque copies en dĂ©but de boucle.
Ce qui donne:
0
1
2
3
0
1
2
3Le trait IntoIterator
Il sâest passĂ© un truc Ă©trange tout Ă lâheure, le compilo a rajoutĂ© un x.into_iter() et ça nâa pas bronchĂ© alors que ni X, ni le trait Iterator ne dĂ©finit de mĂ©thode into_iter.
Ceci câest encore un coup des traits !!
En fait tout Iterator implémente automatiquement IntoIterator au travers de
impl<I: Iterator> IntoIterator for I {
type Item = I::Item;
type IntoIter = I;
#[inline]
fn into_iter(self) -> I {
self
}
}Ceci est fourni dans le langage directement.
Du coup ici
x.into_iterator() <=> xCela ne change rien Ă notre vie mais câest intĂ©ressant pour autre chose. ^^
Il existe Ă©galement une implĂ©mentation pour Vec<T>, et lĂ on commence Ă causer sĂ©rieusement. đ
for i in vec![0, 1, 2, 3] {
}Ainsi le code réel deviens alors:
let iterator = (vec![0, 1, 2, 3]).into_iterator();
for i in iterator {
}Et on revient en terrain connu.
En parlant de terrain connu, on peut maintenant presque deviner le fonctionnement de
for i in 0..4 {
println!("{}", i)
} On peut supposer que en fait câest plus
let iterator = (0..4).into_iterator();
for i in iterator {
println!("{}", i)
} Plus quâĂ comprendre le 0..4.
En Rust, ceci se nomme une Range, la notation du dessus nâest quâun sucre syntaxique.
Donc 0..4 <=> Range { start: 0, end: 4 }.
Un rapide coup dâoeil dans la doc nous montre que Range implĂ©mente Iterator et donc par consĂ©quent IntoIterator.
La boucle est bouclĂ©e! đ€Ł
Je vous laisse implémenter X pour avoir une borne de début et de fin si vous voulez.
Le trait Future
Rappelez-vous lorsque vous aviez 10 ans et que vous attiendiez le livreur arriver avec votre nouveau jeu.
Vous alliez tout les 5 min Ă la fenĂštre pour vĂ©rifier si le camion nâĂ©tait pas au coin de la rue.
Deux solutions:
- il nâest pas encore lĂ et vous reviendrez dans 5 min vĂ©rifier pour la 20 Ăšme fois de la journĂ©e â
- le camion est lĂ !! vous vous prĂ©cipitez pour ouvrir la porte đ
En Rust on peut modéliser ces deux état via une énumération qui possÚde deux variantes.
Here: le camion est lĂ ! đNotYet: il nâest pas encore arrivĂ© âł
enum DeliveryState {
Here(i32)
NotYet
}Et pour simuler notre moi passĂ© qui va Ă la fenĂȘtre on faire quelque chose comme ceci.
On va lĂ aussi dĂ©tailler ce quâest x.
let x = ???;
loop {
if let DeliveryState::Here(i) = x.check_delivery() {
println!("The truck is here {i}");
break
}
// on attend 1 seconde avant de réessayer
println!("Truck not yet here");
sleep(Duration::from_secs(1))
}Ce quâil faut comprendre câest que lâon dĂ©sire que x possĂšde une fonction check_delivery qui doit renvoyer un DeliveryState.
fn delivery(&self) -> DeliveryState;Du coup on peut se créer cette structure qui correspond au cahier des charges du dessus.
struct X;
impl X {
fn check_delivery(&self) -> DeliveryState {
DeliveryState::NotYet
}
}Ce qui donne
let x = X;
loop {
if let DeliveryState::Here(i) = x.check_delivery() {
println!("The truck is here {i}");
break;
}
// on attend 1 seconde avant de réessayer
println!("Truck not yet here");
sleep(Duration::from_secs(1))
}Si on lance ce code, nous allons obtenir:
Truck not here yet
Truck not here yet
Truck not here yet
...
Truck not here yet
Truck not here yet
Truck not here yetIndĂ©fininiment, oui, normal, vu que lâon a pas dâĂ©tat, cela ne risque pas de bouger.
Créons nous un état un peu stupide, mais suffisant pour la démonstration.
Lorsque le compteur atteint $0$, le paquet est livré.
struct X { counter: i32, data: i32 }
impl X {
fn check_delivery(&mut self) -> DeliveryState {
self.counter -= 1;
if self.counter == 0 {
DeliveryState::Here(self.data)
} else {
DeliveryState::NotYet
}
}
}On retente notre chance avec la nouvelle implémentation
let mut x = X {
data: 42,
counter: 4,
};
loop {
if let DeliveryState::Here(i) = x.check_delivery() {
println!("The truck is here {i}");
break;
}
// on attend 1 seconde avant de réessayer
println!("Truck not here yet");
sleep(Duration::from_secs(1))
}Cette fois-ci câest mieux đ
Truck not here yet
Truck not here yet
Truck not here yet
The truck is here 42Bien-sĂ»r, on se doute que lâon ne va pas devoir rĂ©implĂ©menter cette logique Ă chaque fois.
Tout comme notre implĂ©mentation naĂŻve dâitĂ©rateur avait la version trait Iterator.
Il existe également un trait qui se nomme Future, il est fourni par la bibliothÚque standard.
En voici son squelette:
impl Future for X {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
todo!()
}
}Donc ici pas de grosse surprise
type OutputPermet de dĂ©finir ce que lâon veut renvoyer lorsque que ce sera le moment.
Le type de retour est
Poll<Self::Output>Voyons ce quâest Poll.
Il sâagit dâune Ă©numĂ©ration.
pub enum Poll<T> {
Ready(T),
Pending,
}Je ne vous fait pas lâaffront de vous expliquer en quoi câest un peu beacoup ressemblant Ă DeliveryState. đ
Seul petite amĂ©lioration, on peut mettre ce que lâon veut dans le Ready et pas seulement un i32.
Les paramĂštres de fonctions câest un peu plus compliquĂ©.
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>);Le self est un peu bizarre, nous sommes dâhabitude habituĂ© Ă voir &mut self et câest tout.
Ici le self est de type Pin.
Je ne vais pas expliquer ici en dĂ©tail, mais ce quâil faut en comprendre câest que lâon est assurĂ© de pouvoir modifier correctement self grĂące Ă ce type.
Le cx reprĂ©sente le contexte dâexĂ©cution, on revient Ă lui plus tard đ
Bon implémentons notre trait Future pour X.
let mut x = X {
data: 42,
counter: 4,
};
loop {
if let Poll::Ready(i) = x.poll() {
println!("The truck is here {i}");
break;
}
// on attend 1 seconde avant de réessayer
println!("Truck not here yet");
sleep(Duration::from_secs(1))
}Enfin, essayons, parce que pour le moment ce nâest pas brillantâŠ
error[E0599]: no method named `poll` found for struct `X` in the current scope
|
| struct X {
| -------- method `poll` not found for this struct
...
| if let Poll::Ready(i) = x.poll() {
| ^^^^ method not found in `X`
|
= help: items from traits can only be used if the trait is implemented and in scope
= note: the following trait defines an item `poll`, perhaps you need to implement it:
candidate #1: `Future`Mais, mais ⊠On a un trait Future pour X donc x.poll() devrait exister, non ?
Ah non! on a implémenté
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>);Pas
fn poll(&mut self, cx: &mut Context<'_>);Donc le vrai variable qui possĂšde poll() câest:
Pin::new(&mut x)Et non x tout seul.
On change:
let mut x = X {
data: 42,
counter: 4,
};
loop {
if let Poll::Ready(i) = Pin::new(&mut x).poll() {
println!("The truck is here {i}");
break;
}
// on attend 1 seconde avant de réessayer
println!("Truck not here yet");
sleep(Duration::from_secs(1))
}Mieux mais pas top âŠ
error[E0061]: this method takes 1 argument but 0 arguments were supplied
|
| if let Poll::Ready(i) = Pin::new(&mut x).poll() {
| ^^^^-- an argument of type `&mut Context<'_>` is missing
|
note: method defined here
--> C:\Users\Noa\.rustup\toolchains\nightly-x86_64-pc-windows-msvc\lib/rustlib/src/rust\library\core\src\future\future.rs:105:8
|
| fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
| ^^^^
help: provide the argument
|
| if let Poll::Ready(i) = Pin::new(&mut x).poll(/* &mut Context<'_> */) {
| Ah oui câest vrai que poll prend un deuxiĂšme paramĂštre de type Context.
Plus quâĂ le crĂ©er đ
Sauf quâil ne possĂšde pas de constructeur ce type, et il nĂ©cessite un champ waker de type Waker.
struct Context<'a> {
waker: &'a Waker,Bon, du coup créons un Waker, sauf que rebelote pas de constructeur non plus. la structure demande un RaWaker.
struct Waker {
waker: RawWaker,
}On continue Ă suivre les miettes de pains âŠ
Un constructeur mais RawWakerVTable, câest quoi ça encore đ«
impl RawWaker {
fn new(data: *const (), vtable: &'static RawWakerVTable) -> RawWaker {...}
}De toute façon on a descendu dans les entrailles donc pourquoi pas continuer ⊠đ„±
impl RawWakerVTable {
fn new(
clone: unsafe fn(*const ()) -> RawWaker,
wake: unsafe fn(*const ()),
wake_by_ref: unsafe fn(*const ()),
drop: unsafe fn(*const ()),
) -> Self {
Self { clone, wake, wake_by_ref, drop }
}
}Allez ! de lâunsafe, bon on ira pas plus loin đ€Ł
Du coup, nous sommes bloqué ?
Mais non, il y a toujours une porte de sortie.
La structure Waker possÚde une méthode noop qui a pour signature
fn Waker::noop() -> WakerElle est juste caché derriÚre un flag unstable. On peut débrayer ce comportement en rajoutant un attribut à notre fichier.
#![feature(noop_waker)]Ok, donc on a un Waker, plus quâĂ crĂ©er notre contexte et on est bon non ? ^^
Et comme Context possĂšde le bon goĂ»t dâavoir une mĂ©thode from_waker(&Waker).
Nous pouvons terminer de rajouter ce quâil nous manque.
let mut x = X {
data: 42,
counter: 4,
};
// création du waker
let waker = Waker::noop();
// création du contexte
let mut cx = Context::from_waker(&waker);
loop {
if let Poll::Ready(i) = Pin::new(&mut x).poll(&mut cx) {
println!("The truck is here {i}");
break;
}
// on attend 1 seconde avant de réessayer
println!("Truck not here yet");
sleep(Duration::from_secs(1))
}Victoire
Truck not here yet
Truck not here yet
Truck not here yet
The truck is here 42Mais en demi teinte, câest passablement plus compliquĂ© Ă Ă©crire, et je nâai pas du tout apprĂ©ciĂ© mon aventure âĂ la recheche du Contextâ perdu ⊠đ
Les Runtimes
Tout semble compliqué parce que nous ne sommes pas sensé faire tout cela.
En réalité, nous sommes sensé utiliser un Runtime qui va se charger de tout le travail pénible à notre place.
En Rust, il existe plein de runtimes asynchrones différents, et aucun fourni par la bibliothÚque standard.
On va donc devoir en rĂ©cupĂ©rer un. Jâai choisi de le faire avec la crate smol.
Mais ça aurait pu aussi ĂȘtre du tokio ou de lâasync-std.
Bref, installons smol.
cargo add smolLa crate smol fourni une fonction block_on qui prend une Future<Output=T>.
Et ça tombe bien, X implémente Future<Output=i32>.
Modifions notre code pour utiliser smol
fn main() {
let i = smol::block_on(X {
data: 42,
counter: 4,
});
println!("The truck is here {i}");
}Comme la boucle semble avoir disparu, modfions un peu X::poll pour faire apparaĂźtre le print du ânot yetâ.
impl Future for X {
type Output = i32;
fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
self.counter -= 1;
if self.counter == 0 {
Poll::Ready(self.data)
} else {
println!("Truck not here yet");
Poll::Pending
}
}
}Bon on essaie đ
Truck not here yetUne ligne, pas une de plus, le thread est bloquĂ©âŠ
Mais ça câest parce que notre hypothĂšse de travail est fausse.
Si on ouvre le smol::block_on, on peut le découper grosso modo en ce ceci
(je schĂ©matise câest plus compliquĂ© que cela, mais ça sera suffisant pour lâexplication)
fn block_on(future: Future<Output=T>) -> T {
// On créé un tuyaux qui lie unparker à parker
let (parker, unparker) = channel();
// On créé un waker
let waker = waker_fn(|| {
// On envoie un message Ă parker
unparker.unpark()
});
// on créé le contexte
let cx = &mut Context::from_waker(&waker);
loop {
if let Poll::Ready(i) = Pin::new(&mut future).poll(cx) {
return i;
}
// Met en pause le thread âž
// Attend que unparker lui envoie un signal
parker.park()
}
}La deuxiÚme piÚce du puzzle est la méthode waker_fn, elle créé un Waker qui possÚde une méthode wake_by_ref().
Donc si lâon fait
let waker = waker_fn(|| println!("dring Ⱐc'est l'heure de se réveiller"));
waker.wake_by_ref();Cela affiche dans la console
dring â° c'est l'heure de se rĂ©veillerOr il est possible au travers du Context dâatteindre le Waker et par consĂ©quent le wake_by_ref.
cx.waker().wake_by_ref();Comme le Context est passé à la fonction poll(&mut cx).
Alors cela signifie que nous avons accĂšs dans lâimplĂ©mentation de Future de X Ă ce wake_by_ref
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.counter -= 1;
if self.counter == 0 {
Poll::Ready(self.data)
} else {
println!("Truck not here yet");
// On averti le runtime que de la donnée est déjà disponible et
// qu'il peut relancer poll() immédiatement
cx.waker().wake_by_ref();
Poll::Pending
}
}Qui a la possibilitĂ© dâappeler le unparker.unpark()
let waker = waker_fn(|| {
// On envoie un message Ă parker
unparker.unpark()
});et ainsi débloquer le thread
loop {
if let Poll::Ready(i) = Pin::new(&mut future).poll(cx) {
return i;
}
// Un signal est reçu âĄ
parker.park()
// le thread n'est plus en pause
// la boucle peut reprendre â¶
}Si on graphe temporellement, cela donne quelque chose comme cela.
sequenceDiagram
activate Runtime
Runtime->>Future: poll
activate Future
Future -->>Runtime : Poll::Pending
Runtime -->> Runtime: park
Future --)Runtime : wake_by_ref
deactivate Future
Runtime -->> Runtime: unpark
deactivate Runtime
Note left of Runtime: Fin du premier poll
activate Runtime
Runtime->>Future: poll
activate Future
Future -->>Runtime : Poll::Pending
Runtime -->> Runtime: park
Future --)Runtime : wake_by_ref
deactivate Future
Runtime -->> Runtime: unpark
deactivate Runtime
Note left of Runtime: Fin du second poll
activate Runtime
Runtime->>Future: poll
activate Future
Future -->>Runtime : Poll::Ready
deactivate Future
deactivate Runtime
Note left of Runtime: Retour
Le Runtime va poll un premiĂšre fois la future, se park en attente dâun unpark.
La future va lui dĂ©clencher son unpark au travers de wake_by_ref. Et le Runtime re-poll et ainsi de suite, jusquâau Poll::Ready qui coupe la boucle.
Récapitualtif de notre code:
struct X {
counter: i32,
data: i32,
}
impl Future for X {
type Output = i32;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.counter -= 1;
if self.counter == 0 {
Poll::Ready(self.data)
} else {
println!("Truck not here yet");
cx.waker().wake_by_ref();
Poll::Pending
}
}
}
fn main() {
let i = smol::block_on(X {
data: 42,
counter: 4,
});
println!("The truck is here {i}");
}On lance et joie !
Truck not here yet
Truck not here yet
Truck not here yet
The truck is here 42Tout marche !! đ
Finalement, ce nâest pas si compliquĂ© ou fastidieux ^^
Des futures de futures
Enfin presque, il nous manque le fait que lâon attend 1 seconde entre chaque poll.
On va donc simuler ça.
Attention
Cette implĂ©mentation est trĂšs mauvaise, câest simplement pour se donner une idĂ©e.
Pour bien faire il faudrait utiliser un Reactor et des timers, mais ça sera pour un prochain article.
struct Timer {
target: Instant,
}
impl Timer {
fn after(delay: u64) -> Self {
Self {
// timestamp actuel + "delay" secondes
target: Instant::now().add(Duration::from_secs(delay)),
}
}
}
impl Future for Timer {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// timestamp actuel
let last = Instant::now();
// si le temps est écoulé
if last >= self.target {
return Poll::Ready(());
}
// sinon on réveille le Runtime
cx.waker().wake_by_ref();
Poll::Pending
}
}Câest une future, on peut donc lâutiliser dans notre Runtime et la poll.
fn main() {
let now = Instant::now();
smol::block_on(Timer::after(5));
// elasped() calcul le temps passé depuis la création de `now`
println!("Il s'est passé {:?}", now.elapsed());
}Ce qui affiche
Il s'est passé 5.0000061sNous avons bien une future qui se résout au bout de 5 secondes.
Du coup, on peut la poll dans un polling !!
impl Future for X {
type Output = i32;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.counter -= 1;
if self.counter == 0 {
Poll::Ready(self.data)
} else {
println!("Truck not here yet");
// on créé le timer
let mut timer = Timer::after(1);
// on le poll de maniĂšre horrible !!
// Nous ne sommes pas sensĂ© ĂȘtre ici
loop {
if let Poll::Ready(_) = Pin::new(&mut timer).poll(cx) {
break;
}
}
cx.waker().wake_by_ref();
Poll::Pending
}
}
}Attention
Je rĂ©pĂšte : NE FAITES PAS CA câest pour lâexemple.
Vous allez tuer vos performances !!!
le wake interne au timer nâest pas le bon et plein dâautres choses ne vont pas.
Excepté ce petit avertissement ^^
fn main() {
let now = Instant::now();
let i = smol::block_on(X {
counter: 4,
data: 42,
});
// elasped() calcul le temps passé depuis la création de `now`
println!("Il s'est passé {:?}", now.elapsed());
println!("The truck is here {i}");
}Câest quand mĂȘme ce quâon dĂ©sire ^^
Truck not here yet
Truck not here yet
Truck not here yet
Il s'est passé 3.0014149s
The truck is here 42Les sucres syntaxiques async/.await
Maintenant que lâon a globalement fait nâimporte quoi, essayons de rentrer dans le rang. đ
Petite expérience avant.
Voici une future qui finit au premier poll
struct AlwaysReady<T>(T);
impl<T: Clone> Future for AlwaysReady<T> {
type Output = T;
fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
Poll::Ready(self.0.clone())
}
}Elle sâutilise ainsi:
fn main() {
let i = smol::block_on(AlwaysReady(42));
println!("La valeur est {i}");
}Ce qui affiche
La valeur est 42Mais si je vous disais que câest totalement inutile ^^
Rust fourni un sucre syntaxique pour ce genre dâusage au travers du mot-clef async.
fn main() {
let future = async {
42
};
let i = smol::block_on(future);
println!("La valeur est {i}");
}Qui écrit dans la console
La valeur est 42Cela fonction car aussi bien async {42} que AlwaysReady(42) implémente Future<Output=i32>.
On peut mĂȘme simplifier en:
fn main() {
let i = smol::block_on(async { 42 });
println!("La valeur est {i}");
}Ok donc on peut faire implémenter le comportement de X.
fn main() {
let now = Instant::now();
let i = smol::block_on(async {
let mut counter = 0;
loop {
counter += 1;
if counter > 4 {
break 42;
}
Timer::after(1);
println!("Truck not here yet");
}
});
println!("Il s'est passé {:?}", now.elapsed());
println!("La valeur est {i}");
}Alors oui mais câest un peu rapide âŠ
Truck not here yet
Truck not here yet
Truck not here yet
Truck not here yet
Il s'est passé 475.2”s
La valeur est 42Pourtant le timer est bien lĂ . đ
Oui mais une future qui nâest pas poll ne fait rien.
Pollons-la !
fn main() {
let now = Instant::now();
let i = smol::block_on(async {
let mut counter = 0;
loop {
counter += 1;
if counter > 4 {
break 42;
}
let mut timer = Timer::after(1);
loop {
// on a pas de contexte Ă fournir Ă poll()
if Pin::new(&mut timer).poll(???).is_ready() {
break
}
}
println!("Truck not here yet");
}
});
println!("Il s'est passé {:?}", now.elapsed());
println!("La valeur est {i}");
}Oui mais avec quel contexte ?
Alors oui notre arnaque moldave marche âŠ
let waker = Waker::noop();
let mut cx = Context::from_waker(&waker);
let mut timer = Timer::after(1);
loop {
if Pin::new(&mut timer).poll(&mut cx).is_ready() {
break;
}
}Et affiche bien ce quâon dĂ©sire
Truck not here yet
Truck not here yet
Truck not here yet
Truck not here yet
Il s'est passé 4.0007765s
La valeur est 42Mais lĂ aussi on sent que ce nâest pas Ă©lĂ©gant.
Câest pour cela que Rust fourni un dual Ă async qui se nomme .await.
Câest un sucre syntaxique qui sâutilise ainsi.
fn main() {
let now = Instant::now();
let i = smol::block_on(async {
let mut counter = 0;
loop {
counter += 1;
if counter > 4 {
break 42;
}
// on poll la future via la syntaxe `.await`
Timer::after(1).await;
println!("Truck not here yet");
}
});
println!("Il s'est passé {:?}", now.elapsed());
println!("La valeur est {i}");
}Bon lĂ on parle !
Truck not here yet
Truck not here yet
Truck not here yet
Truck not here yet
Il s'est passé 4.0006858s
La valeur est 42le mot-clef .await sâutilise sur nâimporte quelle Future. Il a pour rĂ©sultat dâĂ©crire lâarnaque moldave mais en mieux đ
Pour les personnes curieuses
je ne ferai aucun commentaire, ce nâest que pour votre culture.
Câest pas du vrai Rust mais du pseudo-rust, ça compile pas.
<expr>.awaitdevient
match ::std::future::IntoFuture::into_future(<expr>) { mut __awaitee => loop { match unsafe { ::std::future::Future::poll( <::std::pin::Pin>::new_unchecked(&mut __awaitee), ::std::future::get_context(task_context), ) } { ::std::task::Poll::Ready(result) => break result, ::std::task::Poll::Pending => {} } task_context = yield (); } }Mais ça ressemble vaguement Ă ce quâon a Ă©crit. La rĂ©cupĂ©ration du Context en mieux.
Fonctions asynchrones
Mais on ne va tout de mĂȘme pas Ă©crire tout notre code dans un block? si ? Non biensĂ»r que non.
On peut découper en fonction.
fn iteration(counter: i32) -> bool {
if counter > 4 {
return true
}
Timer::after(1).await;
println!("Truck not here yet");
false
}
fn main() {
let now = Instant::now();
let i = smol::block_on(async {
let mut counter = 0;
loop {
counter += 1;
if iteration(counter) {
break 42;
}
}
});
println!("Il s'est passé {:?}", now.elapsed());
println!("La valeur est {i}");
}(Le dĂ©coupage est un peu nul, mais câest pour lâexemple ^^â)
Si on essaie dâexĂ©cuter
error[E0728]: `await` is only allowed inside `async` functions and blocks
|
| fn iteration(counter: i32) -> bool {
| --------- this is not `async`
...
| Timer::after(1).await;
| ^^^^^ only allowed inside `async` functions and blocksRust nous dit que nous nâavons pas le droit dâutiliser .await en dehors dâun block async ou dâune fonction async.
Tiens, tiens.
On peut donc rendre une fonction asynchrone. Essayons.
async fn iteration(counter: i32) -> bool {
if counter > 4 {
return true
}
Timer::after(1).await;
println!("Truck not here yet");
false
}
fn main() {
let now = Instant::now();
let i = smol::block_on(async {
let mut counter = 0;
loop {
counter += 1;
if iteration(counter) {
break 42;
}
}
});
println!("Il s'est passé {:?}", now.elapsed());
println!("La valeur est {i}");
}Il est bien gentil il nous mĂąche tout le boulot, mĂȘme plus besoin de moi pour expliquer đ
error[E0308]: mismatched types
|
| if iteration(counter) {
| ^^^^^^^^^^^^^^^^^^ expected `bool`, found future
|
note: calling an async function returns a future
--> src\main.rs:104:16
|
| if iteration(counter) {
| ^^^^^^^^^^^^^^^^^^
help: consider `await`ing on the `Future`
|
| if iteration(counter).await {
| ++++++Bien Grand-Chef !
async fn iteration(counter: i32) -> bool {
if counter > 4 {
return true
}
Timer::after(1).await;
println!("Truck not here yet");
false
}
fn main() {
let now = Instant::now();
let i = smol::block_on(async {
let mut counter = 0;
loop {
counter += 1;
// on poll la future
if iteration(counter).await {
break 42;
}
}
});
println!("Il s'est passé {:?}", now.elapsed());
println!("La valeur est {i}");
}Pas mal !
Truck not here yet
Truck not here yet
Truck not here yet
Truck not here yet
Il s'est passé 4.0006166s
La valeur est 42Du coup câest aussi du sucre mais pourquoi ?
Prenons du plus simple.
async fn f() -> i32 {
42
}Clairement câest une Future qui est renvoyĂ©e.
fn main() {
let i = smol::block_on(f());
println!("La valeur est {i}"); // affiche La valeur est 42
}Donc Ă©gal Ă
fn g() -> impl Future<Output = i32> {
async { 42 }
}
fn main() {
let i = smol::block_on(g());
println!("La valeur est {i}"); // affiche La valeur est 42
}DâoĂč
async fn iteration(counter: i32) -> bool {
if counter > 4 {
return true
}
Timer::after(1).await;
println!("Truck not here yet");
false
}
// Equivalent Ă
fn iteration(counter: i32) -> impl Future<Output=bool> {
async {
if counter > 4 {
return true
}
Timer::after(1).await;
println!("Truck not here yet");
false
}
}On retrouve notre async, on peut utiliser notre .await. đ
Itérer sur des futures
Tout ça commence Ă ĂȘtre trĂšs sympa, maintenant, nous allons exploiter la lazyness des futures Ă notre avantage.
Autrement dit, tant quâune future nâest pas poll, elle ne fait rien, ce nâest quâune boĂźte que lâon peut transporter et rĂ©fĂ©rencer.
On va donc sâen faire une collections et itĂ©rer dessus.
Et oui ! Tout le laĂŻus interminable sur les itĂ©rateurs dâil y a plus de 1000 lignes câĂ©tait pas pour faire du remplissage. đ
Donc premiÚre chose, nous allons créer une fonction delivery qui nous renverra une Future.
async fn delivery(delay: i32, value: i32) -> i32 {
let mut counter = 0;
let now = Instant::now();
loop {
counter += 1;
if iteration(counter, delay).await {
println!("Il s'est passé {:?}", now.elapsed());
break value;
}
}
}On modifie également la méthode itération
async fn iteration(counter: i32, target: i32) -> bool {
if counter > target {
return true;
}
Timer::after(1).await;
println!("Truck not here yet");
false
}Notez que tout est en async fn donc tout est Future.
On peut donc mettre notre Future dans un tableau et itĂ©rĂ© dessus. đ
fn main() {
// on créer un tableau de future
let deliveries = vec![
delivery(4, 42),
delivery(2, 12)
];
smol::block_on(async {
// on boucle dessus
for delivery_process in deliveries {
// on poll la future
println!("La valeur est {}", delivery_process.await);
println!()
}
});
}On affiche bien
Truck not here yet
Truck not here yet
Truck not here yet
Truck not here yet
Il s'est passé 4.0006349s
La valeur est 42
Truck not here yet
Truck not here yet
Il s'est passé 2.0002937s
La valeur est 12Le code complet
use std::future::Future; use std::ops::Add; use std::pin::Pin; use std::task::{Context, Poll}; use std::time::{Duration, Instant}; struct Timer { target: Instant, } impl Timer { fn after(delay: u64) -> Self { Self { target: Instant::now().add(Duration::from_secs(delay)), } } } impl Future for Timer { type Output = (); fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { let last = Instant::now(); if last >= self.target { return Poll::Ready(()); } cx.waker().wake_by_ref(); Poll::Pending } } async fn iteration(counter: i32, target: i32) -> bool { if counter > target { return true; } Timer::after(1).await; println!("Truck not here yet"); false } async fn delivery(delay: i32, value: i32) -> i32 { let mut counter = 0; let now = Instant::now(); loop { counter += 1; if iteration(counter, delay).await { println!("Il s'est passé {:?}", now.elapsed()); break value; } } } fn main() { // on créer un tableau de future let deliveries = vec![ delivery(4, 42), delivery(2, 12) ]; smol::block_on(async { // on boucle dessus for delivery_process in deliveries { // on poll la future println!("La valeur est {}", delivery_process.await); println!() } }); }
Donc on sait itĂ©rer mais il manque un dernier morceau, pour le moment nous avons une relation 1:1 entre le tableau et ce quâon itĂšre.
Nous allons voir comment créer une indirection.
Itérateur asynchrone
Cahier des charges
Voici le jeu de donnĂ©es que lâon possĂšde. On va le transformer en un tableau de futures pour simuler une latence rĂ©seau par exemple.
let data = vec![
("prefix_1_data1", "jeu"),
("prefix_1_data2", "de"),
("prefix_1_data3", "données 1"),
("prefix_2_data1", "le"),
("prefix_2_data2", "ciel"),
("prefix_2_data3", "est"),
("prefix_2_data4", "gris"),
("prefix_3_data1", "L'asynchrone"),
("prefix_3_data2", "ce n'est"),
("prefix_3_data3", "pas si"),
("prefix_3_data4", "compliqué"),
];Voici le rĂ©sultat que lâon dĂ©sire
let expected = [
"jeu de données 1",
"le ciel est gris",
"L'asynchrone ce n'est pas si compliqué"
];Attention ça va ĂȘtre un peu tordu, lâabstraction nĂ©cessite parfois de tordre la rĂ©alitĂ© ^^
Tout dâabord que voulons nous comme API haut-niveau ?
Dans lâidĂ©al ceci :
fn main() {
let data = vec![....]; // Un tableau de futures
smol::block_on(async {
let mut x = X::new(data);
for sentence in x.await {
// ...
}
});
}Mais nous ne pourrons pas faire ça, rappellez vous, si lâon desugar cela donnerait:
let iterator = (x).into_iterator(); // iterator = x
for sentence in iterator.await {}Or cela nous oblige à implémenter à la fois Future et Iterator.
Mais du coup, le .await nâest lĂ que pour sortir lâiterateur, pas poller de la nouvelle donnĂ©es.
Bref on se retrouve coincé assez vite dans notre tentative de for.
Bon et par un while. On implémente Future sur X et fin.
fn main() {
let data = vec![....]; // Un tableau de futures
smol::block_on(async {
let mut x = X::new(data);
while let Some(sentence) = x.await {
}
});
}Mais non đ
error[E0382]: use of moved value: `x`
|
| let mut x = X::new(data);
| ----- move occurs because `x` has type `X`, which does not implement the `Copy` trait
|
| while let Some(t) = x.await {}
| --------------------^------
| | |
| | `x` moved due to this await, in previous iteration of loop
| inside of this loop
|
note: `into_future` takes ownership of the receiver `self`, which moves `x`
|
| fn into_future(self) -> Self::IntoFuture;Car derriĂšre le rideau, .await applle into_future qui consomme x,
Nous nâallons pas non plus pouvoir copier car les data seront allouĂ©es et donc non-copiable.
Nous devons donc partir sur une double indirection.
fn main() {
let data = vec![....];
smol::block_on(async {
let mut x = X::new(data);
while let Some(sentence) = x.next().await {
}
});
}Quâil faut lire comme
loop {
let next_future = x.next();
if let Some(sentence) = next_future.await {
} else {
break
}
}On va donc appelĂ© x.next() Ă chaque tour de boucle, on sait aussi que next_future est une impl Future<Output=Option<String>> vu que lâon veut des Option<String> en bout de chaĂźne.
On sait Ă©galement que next() ne peut pas consommer x sinon on retombe dans le mĂȘme travers que into_future.
Donc X possÚde une fonction next qui retourne quelque chose Y qui implémente Future<Output=Option<String>>.
Pour plus de lisibilité, nous allons appeler notre Y une NextFuture.
Il faudra Ă©galement que NextFuture connaisse le X quâil lâa créé.
Par conséquent la signature de X::next est :
fn next(&mut self) -> NextFuture;Nous pouvons en déterminer également le
type Output = Option<String>Et donc implémenter NextFuture.
Alors câest parti pour implĂ©menter tout ce beau monde.
Avec une future fixe
Dâabord X.
On se créé un type custom.
type FutureInput = AlwaysReady<(String, String)>;Puis X
struct X {
data: Vec<FutureInput>,
current: Option<FutureInput>,
}Puis on implémente le constructeur et next
impl X {
fn new(data: Vec<(&str, &str)>) -> Self {
let data = data
.into_iter()
// on créé un tableau de futures de nos données
.map(|(x, y)| AlwaysReady((x.to_string(), y.to_string())))
// on inverse parce que l'on va pop par la fin, or on veut le début
.rev()
.collect();
Self {
data,
current: None,
}
}
fn next(&mut self) -> NextFuture<'_> {
NextFuture { x: self }
}
}Le <'_> aprĂšs le NextFuture câest pour pas sâennuyer avec la lifetime, on laisse Rust bosser comme un grand.
Comme promis, NextFuture prend une référence de X mutable.
struct NextFuture<'a> {
x: &'a mut X,
}LâimplĂ©mentation de la Future appelle le poll de la rĂ©fĂ©rence de X.
impl<'a> Future for NextFuture<'a> {
type Output = Option<String>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
Pin::new(&mut self.x).poll(cx)
}
}Du coup, on implémente également Future pour X.
impl Future for X {
// peut-ĂȘtre un nouvel Ă©lĂ©ment
type Output = Option<String>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// on initialise le poll Ă venir
if self.current.is_none() {
self.current = self.data.pop();
}
match &mut self.current {
// s'il n'y a plus rien a poll on envoie None
None => Poll::Ready(None),
// si on peut poll, on poll
Some(ref mut current_future) => match Pin::new(current_future).poll(cx) {
// si les données son disponible
Poll::Ready(data) => {
// on réinitialise le current poll
self.current = None;
// on renvoie les données
Poll::Ready(Some(data.0))
}
// sinon on demande gentiement de revenir plus tard
Poll::Pending => {
// on réveille le runtime
cx.waker().wake_by_ref();
// on lui explique que ce n'est pas pour tout de suite les données
Poll::Pending
},
},
}
}
}Et en exécutant,
fn main() {
let data = vec![
("prefix_1_data1", "jeu"),
("prefix_1_data2", "de"),
("prefix_1_data3", "données 1"),
("prefix_2_data1", "le"),
("prefix_2_data2", "ciel"),
("prefix_2_data3", "est"),
("prefix_2_data4", "gris"),
("prefix_3_data1", "L'asynchrone"),
("prefix_3_data2", "ce n'est"),
("prefix_3_data3", "pas si"),
("prefix_3_data4", "compliqué"),
];
smol::block_on(async move {
let mut x = X::new(data);
while let Some(key) = x.next().await {
println!("la clef vaut {key}");
}
})
}on obtient
la clef vaut prefix_1_data1
la clef vaut prefix_1_data2
la clef vaut prefix_1_data3
la clef vaut prefix_2_data1
la clef vaut prefix_2_data2
la clef vaut prefix_2_data3
la clef vaut prefix_2_data4
la clef vaut prefix_3_data1
la clef vaut prefix_3_data2
la clef vaut prefix_3_data3
la clef vaut prefix_3_data4Cool mais mais câest pas trĂšs souple, nous avons seulement une future qui se rĂ©sout tout de seul.
Nous, nous voulons du délai pour simuler le réseau.
On va procéder un remaniement ministériel.
Avec une future quelconque
PremiÚrement, nous nons créons une fonction asynchrone qui sera notre future avec délai.
use async_io::Timer;
async fn delivery<T>(delay: u64, value: T) -> T {
let now = Instant::now();
Timer::after(Duration::from_millis(delay)).await;
println!("Il s'est passé {:?}", now.elapsed());
value
}Pour lâoccasion, je remplace mon Timer foireux par un vrai, celui fourni par async_io au travers de smol.
Je créé deux types pour me faciliter la vie
type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
type FutureInput = (String, String);Le premier permet de ranger une future quelconque dans une boĂźte et de pouvoir la transbahuter sans y penser.
Le second est lâentrĂ© des donnĂ©es: (clef, valeur)
Je modifie X pour refléter se nouveau type.
struct X {
data: Vec<BoxFuture<FutureInput>>,
current: Option<BoxFuture<FutureInput>>,
}Je créé une fonction boxed qui va transformer une Future en BoxFuture, le type est compliquĂ© mais globalement, ça veut juste dire âcâest open bar, fait ce que tu veux, jâai calĂ© ta future en mĂ©moire, je te file de quoi y accĂ©der et je te promet quâelle sera lĂ quand tu en aura besoinâ.
fn boxed<'a, U, T: Future<Output = U>>(value: T) -> Pin<Box<dyn Future<Output = U> + Send + 'a>>
where
T: Sized + Send + 'a,
{
Box::pin(value)
}Je modifie la fonction new en consĂ©quence, et en rajoutant un peu dâalĂ©atoire.
cargo add randimpl X {
fn new(data: Vec<(&str, &str)>) -> Self {
let mut rng = rand::thread_rng();
let data = data
.into_iter()
.map(|(x, y)| {
// on génÚre le délai
let delay = rng.gen_range(100..2000);
// on créé la future
let fut = delivery(delay, (x.to_string(), y.to_string()));
// on box la future
boxed(fut)
})
.rev()
.collect();
Self {
data,
current: None,
}
}
}Le reste est identique
Code complet
use async_io::Timer; use rand::Rng; use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; use std::time::{Duration, Instant}; // dĂ©claration des types type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>; type FutureInput = (String, String); // future avec dĂ©lai async fn delivery<T>(delay: u64, value: T) -> T { let now = Instant::now(); Timer::after(Duration::from_millis(delay)).await; println!("Il s'est passĂ© {:?}", now.elapsed()); value } // boxing de future fn boxed<'a, U, T: Future<Output = U>>(value: T) -> Pin<Box<dyn Future<Output = U> + Send + 'a>> where T: Sized + Send + 'a, { Box::pin(value) } /// ReprĂ©sente le retour de `X::next()` struct NextFuture<'a> { x: &'a mut X, } impl<'a> Future for NextFuture<'a> { type Output = Option<String>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { Pin::new(&mut self.x).poll(cx) } } /// ItĂ©rateur asynchrone struct X { /// Futures Ă itĂ©rer data: Vec<BoxFuture<FutureInput>>, /// Futur actuellement itĂ©rĂ©e current: Option<BoxFuture<FutureInput>>, } impl X { fn new(data: Vec<(&str, &str)>) -> Self { let mut rng = rand::thread_rng(); let data = data .into_iter() .map(|(x, y)| { // on gĂ©nĂšre le dĂ©lai let delay = rng.gen_range(100..2000); // on créé la future let fut = delivery(delay, (x.to_string(), y.to_string())); // on box la future boxed(fut) }) .rev() .collect(); Self { data, current: None, } } /// Retourne une instance de `NextFuture` fn next(&mut self) -> NextFuture<'_> { NextFuture { x: self } } } impl Future for X { type Output = Option<String>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { // on dĂ©finit la future Ă poller s'il n'y en a pas if self.current.is_none() { // en popant le dernier Ă©lĂ©ment de notre // tableau de futures self.current = self.data.pop(); } // on vĂ©rifie s'il a toujours quelque chose Ă poller match &mut self.current { // si non on indique que c'Ă©tait le dernier Ă©lĂ©ment None => Poll::Ready(None), // si oui, on poll la futur Some(ref mut current_future) => match Pin::new(current_future).poll(cx) { // si la futur est complĂ©tĂ©e Poll::Ready(data) => { // on rĂ©initialise la future Ă poller self.current = None; // on renvoie les donnĂ©es Poll::Ready(Some(data.0)) } // sinon Poll::Pending => { // on rĂ©veille le runtime cx.waker().wake_by_ref(); // on lui dit que ce n'est pas encore prĂȘt Poll::Pending } }, } } } fn main() { let data = vec![ ("prefix_1_data1", "jeu"), ("prefix_1_data2", "de"), ("prefix_1_data3", "donnĂ©es 1"), ("prefix_2_data1", "le"), ("prefix_2_data2", "ciel"), ("prefix_2_data3", "est"), ("prefix_2_data4", "gris"), ("prefix_3_data1", "L'asynchrone"), ("prefix_3_data2", "ce n'est"), ("prefix_3_data3", "pas si"), ("prefix_3_data4", "compliquĂ©"), ]; smol::block_on(async move { let mut x = X::new(data); while let Some(key) = x.next().await { println!("la clef vaut {key}"); println!() } }) }
Il s'est passé 185.0034ms
la clef vaut prefix_1_data1
Il s'est passé 1.7600028s
la clef vaut prefix_1_data2
Il s'est passé 1.2960022s
la clef vaut prefix_1_data3
Il s'est passé 1.2860019s
la clef vaut prefix_2_data1
Il s'est passé 331.0027ms
la clef vaut prefix_2_data2
Il s'est passé 406.0027ms
la clef vaut prefix_2_data3
Il s'est passé 1.7650035s
la clef vaut prefix_2_data4
Il s'est passé 688.0042ms
la clef vaut prefix_3_data1
Il s'est passé 1.5060066s
la clef vaut prefix_3_data2
Il s'est passé 1.8030035s
la clef vaut prefix_3_data3
Il s'est passé 124.0034ms
la clef vaut prefix_3_data4Yeah ! Du dĂ©lai ! đ€©
On simule du réseau pas stable ou de la DB lente ^^
DĂ©coupler le flux dâentrĂ© de la sortie
Bon câest cool mais ce nâest pas ce quâon dĂ©sire.
Pour rappel on en entré
let data = vec![
("prefix_1_data1", "jeu"),
("prefix_1_data2", "de"),
("prefix_1_data3", "données 1"),
("prefix_2_data1", "le"),
("prefix_2_data2", "ciel"),
("prefix_2_data3", "est"),
("prefix_2_data4", "gris"),
("prefix_3_data1", "L'asynchrone"),
("prefix_3_data2", "ce n'est"),
("prefix_3_data3", "pas si"),
("prefix_3_data4", "compliqué"),
];Et voici le rĂ©sultat que lâon dĂ©sire
let expected = [
"jeu de données 1",
"le ciel est gris",
"L'asynchrone ce n'est pas si compliqué"
];Or pour le moment, nous avons:
let result = vec![
"prefix_1_data1",
"prefix_1_data2",
"prefix_1_data3",
"prefix_2_data1",
"prefix_2_data2",
"prefix_2_data3",
"prefix_2_data4",
"prefix_3_data1",
"prefix_3_data2",
"prefix_3_data3",
"prefix_3_data4",
];Nous allons donc retarder la sortie de la données et accumuler assez de résultat.
Pour accumuler nous allons utiliser le fait que les clefs sont toutes sous le modĂšle
prefix_<id>_data<n>Accumuler sur ce prefix_<id> et lorsque lâon passe au suivant relĂącher la concatĂ©nation de ce quâon a accumulĂ©.
Pour le coup câest vraiment de lâalgoritmique, le seul point de vigilence est de bien renvoyer le dernier morceau dâaccumulateur
Code complet annoté
use async_io::Timer; use rand::Rng; use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; use std::time::{Duration, Instant}; // dĂ©claration des types type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>; type FutureInput = (String, String); // future avec dĂ©lai async fn delivery(delay: u64, value: FutureInput) -> FutureInput { let now = Instant::now(); Timer::after(Duration::from_millis(delay)).await; println!("La clef {} est arrivĂ©e en {:?}", value.0, now.elapsed()); value } // boxing de future fn boxed<'a, U, T: Future<Output = U>>(value: T) -> Pin<Box<dyn Future<Output = U> + Send + 'a>> where T: Sized + Send + 'a, { Box::pin(value) } /// ReprĂ©sente le retour de `X::next()` struct NextFuture<'a> { x: &'a mut X, } impl<'a> Future for NextFuture<'a> { type Output = Option<String>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { Pin::new(&mut self.x).poll(cx) } } /// ItĂ©rateur asynchrone struct X { /// Futures Ă itĂ©rer data: Vec<BoxFuture<FutureInput>>, /// Futur actuellement itĂ©rĂ©e current_futur: Option<BoxFuture<FutureInput>>, /// Prefix courant prefix: Option<String>, /// Accumulateur accumulator: Vec<String>, } impl X { fn new(data: Vec<(&str, &str)>) -> Self { let mut rng = rand::thread_rng(); let data = data .into_iter() .map(|(x, y)| { // on gĂ©nĂšre le dĂ©lai let delay = rng.gen_range(100..2000); // on créé la future let fut = delivery(delay, (x.to_string(), y.to_string())); // on box la future boxed(fut) }) .rev() .collect(); Self { data, current_futur: None, prefix: None, accumulator: vec![], } } /// Retourne une instance de `NextFuture` fn next(&mut self) -> NextFuture<'_> { NextFuture { x: self } } } impl Future for X { type Output = Option<String>; fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { // on dĂ©finit la future Ă poller s'il n'y en a pas if self.current_futur.is_none() { // en popant le dernier Ă©lĂ©ment de notre // tableau de futures self.current_futur = self.data.pop(); } // on vĂ©rifie s'il a toujours quelque chose Ă poller match &mut self.current_futur { // si non on indique que c'Ă©tait le dernier Ă©lĂ©ment None => { // on vĂ©rifie si l'accumulateur n'est pas vide if !self.accumulator.is_empty() { // on draine l'accumulateur let data = self.accumulator.drain(..).collect::<Vec<String>>(); // on renvoie les donnĂ©es return Poll::Ready(Some(data.join(" "))); } // au prochain poll l'itĂ©ration sera terminĂ©e Poll::Ready(None) } // si oui, on poll la futur Some(ref mut current_future) => match Pin::new(current_future).poll(cx) { // si la futur est complĂ©tĂ©e Poll::Ready((key, value)) => { // on split sur la clef let key = key.split('_').skip(1).take(1).collect::<String>(); // on vĂ©rifie si un prĂ©fix existe dĂ©jĂ if self.prefix.is_none() { // sinon on le créé self.prefix = Some(key.clone()) } // on dĂ©fini un tableau pour un potentiel renvoie de donnĂ©es let mut ready_data = vec![]; if let Some(ref actual_prefix) = self.prefix { // Le prefix a changĂ© nous pouvons relĂącher une valeur if &key != actual_prefix { // on vide l'accumulateur ready_data = self.accumulator.drain(..).collect(); // on initialise au prochain prefix self.prefix = Some(key); } } // on accumule le rĂ©sultat self.accumulator.push(value); // on rĂ©initialise la future Ă poller self.current_futur = None; // on vĂ©rifie si des donnĂ©es sont prĂȘte Ă ĂȘtre release if ready_data.is_empty() { // on rĂ©veille le runtime cx.waker().wake_by_ref(); // on lui dit que ce n'est pas encore prĂȘt Poll::Pending } else { // on fusionne le tableau let data = ready_data.join(" "); // on renvoie les donnĂ©es Poll::Ready(Some(data)) } } // sinon Poll::Pending => { // on rĂ©veille le runtime cx.waker().wake_by_ref(); // on lui dit que ce n'est pas encore prĂȘt Poll::Pending } }, } } } fn main() { let data = vec![ ("prefix_1_data1", "jeu"), ("prefix_1_data2", "de"), ("prefix_1_data3", "donnĂ©es 1"), ("prefix_2_data1", "le"), ("prefix_2_data2", "ciel"), ("prefix_2_data3", "est"), ("prefix_2_data4", "gris"), ("prefix_3_data1", "L'asynchrone"), ("prefix_3_data2", "ce n'est"), ("prefix_3_data3", "pas si"), ("prefix_3_data4", "compliquĂ©"), ]; smol::block_on(async move { let mut x = X::new(data); let mut now = Instant::now(); while let Some(message) = x.next().await { println!("le message \"{message}\" est arrivĂ©e en {:?}", now.elapsed()); now = Instant::now(); println!() } }) }
Ce qui nous donne
La clef prefix_1_data1 est arrivée en 1.1680078s
La clef prefix_1_data2 est arrivée en 1.6620032s
La clef prefix_1_data3 est arrivée en 152.0027ms
La clef prefix_2_data1 est arrivée en 1.3760026s
le message "jeu de données 1" est arrivée en 4.3585695s
La clef prefix_2_data2 est arrivée en 1.3740032s
La clef prefix_2_data3 est arrivée en 1.4890043s
La clef prefix_2_data4 est arrivée en 1.9170056s
La clef prefix_3_data1 est arrivée en 1.9830028s
le message "le ciel est gris" est arrivée en 6.7637874s
La clef prefix_3_data2 est arrivée en 310.0058ms
La clef prefix_3_data3 est arrivée en 587.0023ms
La clef prefix_3_data4 est arrivée en 527.0066ms
le message "L'asynchrone ce n'est pas si compliquĂ©" est arrivĂ©e en 1.4246736sEt lĂ câest vraiment le Champagne đŸ đ€©
Nous venons de crĂ©er un systĂšme qui est capable de retarder la crĂ©ation dâun flux avec des donnĂ©es qui elle mĂȘme sont retardĂ©e de maniĂšre alĂ©atoire.
Bref en un mot comme en 100, nous avons un Itérateur Asynchrone !!!
Sâil reste des futures Ă poller cela donne quelque chose comme cela.
sequenceDiagram
participant Future
X ->> NextFuture : next()
activate NextFuture
Note left of X: création du NextFuture
Runtime ->> NextFuture : poll()
Note right of Runtime: on poll NextFuture
NextFuture ->> X : poll()
Note left of Future: On poll la premiĂšre future
activate Future
X ->> Future : poll()
Future -->> X : Poll::Pending
Note left of X: La future n'est pas prĂȘte
X -->> NextFuture : Poll::Pending
NextFuture -->> Runtime : Poll::Pending
Runtime ->> NextFuture : poll()
Note right of Runtime: on poll NextFuture
NextFuture ->> X : poll()
X ->> Future : poll()
Future -->> X : Poll::Ready(data)
Note left of X: La future est prĂȘte
X -->> NextFuture : Poll::Pending
deactivate Future
Note left of X: Mais X n'a pas assez accumulé
NextFuture -->> Runtime : Poll::Pending
Runtime ->> NextFuture : poll()
Note right of Runtime: on poll NextFuture
NextFuture ->> X : poll()
Note left of Future: On passe Ă la future suivante
activate Future
X ->> Future : poll()
Future -->> X : Poll::Ready(data)
X -->> NextFuture : Poll::Ready(Some(data))
deactivate Future
Note left of X: X a assez accumulé
NextFuture -->> Runtime : Poll::Ready(Some(data))
Note left of Runtime: on relùche une donnée
deactivate NextFuture
Lorsque lâon arrive au bout de lâitĂ©ration cela donne
sequenceDiagram
X ->> NextFuture : next()
activate NextFuture
Note left of X: création du NextFuture
Runtime ->> NextFuture : poll()
Note right of Runtime: on poll NextFuture
NextFuture ->> X : poll()
X -->> NextFuture : Poll::Ready(Some(data))
NextFuture -->> Runtime : Poll::Ready(Some(data))
Note left of X: Avec data
deactivate NextFuture
X ->> NextFuture : next()
activate NextFuture
Note left of X: nouvelle itération
Runtime ->> NextFuture : poll()
Note right of Runtime: on poll NextFuture
NextFuture ->> X : poll()
X -->> NextFuture : Poll::Ready(None)
NextFuture -->> Runtime : Poll::Ready(None)
Note left of X: Plus de data
deactivate NextFuture
le trait Stream et lâĂ©cosystĂšme asynchrone
Bon, on a bien rigolĂ© Ă rĂ©inventer la roue, mais vous vous douter que tout ça existe dĂ©jĂ et nâest pas Ă rĂ©plĂ©menter Ă chaque fois.
La premiĂšre chose Ă comprendre câest que mĂȘme si la librairie ne fournie pas les outils, les choses sont dĂ©jĂ plus ou moins normailisĂ©es et nâattende que dâĂȘtre officialisĂ©e.
Prenez par exemple la crate future_util, véritable couteau suisse des Futures.
La méthode boxed que nous avons implémenté existe déjà .
fn boxed<'a, U, T: Future<Output = U>>(value: T) -> Pin<Box<dyn Future<Output = U> + Send + 'a>>
where
T: Sized + Send + 'a,
{
Box::pin(value)
}Seulement elle est directement rattaché au trait FutureExt.
Trait qui est automatiquement implémenté pour toute future.
impl<F: Future + ?Sized> FutureExt for F {}Ainsi on peut faire ceci
cargo add futures-liteOn utilise le type Boxed de futures-utils
pub type Boxed<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;dans notre X
// Attention a importer le trait, sinon .boxed() n'existe pas !
use futures_lite::FutureExt;
struct X {
/// Futures à itérer
data: Vec<Boxed<FutureInput>>,
/// Futur actuellement itérée
current_futur: Option<Boxed<FutureInput>>,
/// Prefix courant
prefix: Option<String>,
/// Accumulateur
accumulator: Vec<String>,
}
impl X {
fn new(data: Vec<(&str, &str)>) -> Self {
let mut rng = rand::thread_rng();
let data = data
.into_iter()
.map(|(x, y)| {
// on génÚre le délai
let delay = rng.gen_range(100..2000);
// on créé la future
let fut = delivery(delay, (x.to_string(), y.to_string()));
// on box la future avec le trait FutureExt::boxed
fut.boxed()
})
.rev()
.collect();
Self {
data,
current_futur: None,
prefix: None,
accumulator: vec![],
}
}
}Et du coup, nous pouvons nous séparer de notre fonction boxed maison, et de notre BoxFuture.
type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;Bon et maintenant si je vous disais que le NextFuture existe déjà ici.
Utilisons-le Ă la place
use futures_lite::stream::NextFuture;
impl X {
/// Retourne une instance de `NextFuture`
fn next(&mut self) -> NextFuture<'_, Self> {
NextFuture { stream: self }
}
}Oui mais on ne peut pas le crĂ©er, le champ est privĂ© et il nây a pas de constructeur.
error[E0451]: field `stream` of struct `NextFuture` is private
|
| NextFuture { stream: self }
| ^^^^^^^^^^^^ private field
Mais ça câest parce que encore une fois, on ne travail pas comme il faut.
La vraie maniĂšre est dâimplĂ©menter le trait Stream.
Elle demande de dĂ©finir une mĂ©thode poll_next qui a lâexacte signature par repport Ă poll.
impl Future for X {
type Output = Option<String>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {}
}
impl Stream for X {
type Item = String;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {}
}Allons-y.
Implémentation de Stream pour X
impl Stream for X { type Item = String; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { // on dĂ©finit la future Ă poller s'il n'y en a pas if self.current_futur.is_none() { // en popant le dernier Ă©lĂ©ment de notre // tableau de futures self.current_futur = self.data.pop(); } // on vĂ©rifie s'il a toujours quelque chose Ă poller match &mut self.current_futur { // si non on indique que c'Ă©tait le dernier Ă©lĂ©ment None => { // on vĂ©rifie si l'accumulateur n'est pas vide if !self.accumulator.is_empty() { // on draine l'accumulateur let data = self.accumulator.drain(..).collect::<Vec<String>>(); // on renvoie les donnĂ©es return Poll::Ready(Some(data.join(" "))); } // au prochain poll l'itĂ©ration sera terminĂ©e Poll::Ready(None) } // si oui, on poll la futur Some(ref mut current_future) => match Pin::new(current_future).poll(cx) { // si la futur est complĂ©tĂ©e Poll::Ready((key, value)) => { // on split sur la clef let key = key.split('_').skip(1).take(1).collect::<String>(); // on vĂ©rifie si un prĂ©fix existe dĂ©jĂ if self.prefix.is_none() { // sinon on le créé self.prefix = Some(key.clone()) } // on dĂ©fini un tableau pour un potentiel renvoie de donnĂ©es let mut ready_data = vec![]; if let Some(ref actual_prefix) = self.prefix { // Le prefix a changĂ© nous pouvons relĂącher une valeur if &key != actual_prefix { // on vide l'accumulateur ready_data = self.accumulator.drain(..).collect(); // on initialise au prochain prefix self.prefix = Some(key); } } // on accumule le rĂ©sultat self.accumulator.push(value); // on rĂ©initialise la future Ă poller self.current_futur = None; // on vĂ©rifie si des donnĂ©es sont prĂȘte Ă ĂȘtre release if ready_data.is_empty() { // on rĂ©veille le runtime cx.waker().wake_by_ref(); // on lui dit que ce n'est pas encore prĂȘt Poll::Pending } else { // on fusionne le tableau let data = ready_data.join(" "); // on renvoie les donnĂ©es Poll::Ready(Some(data)) } } // sinon Poll::Pending => { // on rĂ©veille le runtime cx.waker().wake_by_ref(); // on lui dit que ce n'est pas encore prĂȘt Poll::Pending } }, } } }
On implémente
poll_nextet paspollcar leNextFuturede la lib appellepoll_next.
Ce qui nous donne finalement ce code
Code complet
use async_io::Timer; use futures_lite::future::Boxed; use futures_lite::stream::StreamExt; use futures_lite::{FutureExt, Stream}; use rand::Rng; use std::future::Future; use std::pin::Pin; use std::task::{Context, Poll}; use std::time::{Duration, Instant}; // dĂ©claration des types type FutureInput = (String, String); // future avec dĂ©lai async fn delivery(delay: u64, value: FutureInput) -> FutureInput { let now = Instant::now(); Timer::after(Duration::from_millis(delay)).await; println!("La clef {} est arrivĂ©e en {:?}", value.0, now.elapsed()); value } /// ItĂ©rateur asynchrone struct X { /// Futures Ă itĂ©rer data: Vec<Boxed<FutureInput>>, /// Futur actuellement itĂ©rĂ©e current_futur: Option<Boxed<FutureInput>>, /// Prefix courant prefix: Option<String>, /// Accumulateur accumulator: Vec<String>, } impl X { fn new(data: Vec<(&str, &str)>) -> Self { let mut rng = rand::thread_rng(); let data = data .into_iter() .map(|(x, y)| { // on gĂ©nĂšre le dĂ©lai let delay = rng.gen_range(100..2000); // on créé la future let fut = delivery(delay, (x.to_string(), y.to_string())); // on box la future fut.boxed() }) .rev() .collect(); Self { data, current_futur: None, prefix: None, accumulator: vec![], } } } // on implĂ©mente Stream impl Stream for X { // pas besoin de l'option elle est dans le type de retour de poll_next type Item = String; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { // on dĂ©finit la future Ă poller s'il n'y en a pas if self.current_futur.is_none() { // en popant le dernier Ă©lĂ©ment de notre // tableau de futures self.current_futur = self.data.pop(); } // on vĂ©rifie s'il a toujours quelque chose Ă poller match &mut self.current_futur { // si non on indique que c'Ă©tait le dernier Ă©lĂ©ment None => { // on vĂ©rifie si l'accumulateur n'est pas vide if !self.accumulator.is_empty() { // on draine l'accumulateur let data = self.accumulator.drain(..).collect::<Vec<String>>(); // on renvoie les donnĂ©es return Poll::Ready(Some(data.join(" "))); } // au prochain poll l'itĂ©ration sera terminĂ©e Poll::Ready(None) } // si oui, on poll la futur Some(ref mut current_future) => match Pin::new(current_future).poll(cx) { // si la futur est complĂ©tĂ©e Poll::Ready((key, value)) => { // on split sur la clef let key = key.split('_').skip(1).take(1).collect::<String>(); // on vĂ©rifie si un prĂ©fix existe dĂ©jĂ if self.prefix.is_none() { // sinon on le créé self.prefix = Some(key.clone()) } // on dĂ©fini un tableau pour un potentiel renvoie de donnĂ©es let mut ready_data = vec![]; if let Some(ref actual_prefix) = self.prefix { // Le prefix a changĂ© nous pouvons relĂącher une valeur if &key != actual_prefix { // on vide l'accumulateur ready_data = self.accumulator.drain(..).collect(); // on initialise au prochain prefix self.prefix = Some(key); } } // on accumule le rĂ©sultat self.accumulator.push(value); // on rĂ©initialise la future Ă poller self.current_futur = None; // on vĂ©rifie si des donnĂ©es sont prĂȘte Ă ĂȘtre release if ready_data.is_empty() { // on rĂ©veille le runtime cx.waker().wake_by_ref(); // on lui dit que ce n'est pas encore prĂȘt Poll::Pending } else { // on fusionne le tableau let data = ready_data.join(" "); // on renvoie les donnĂ©es Poll::Ready(Some(data)) } } // sinon Poll::Pending => { // on rĂ©veille le runtime cx.waker().wake_by_ref(); // on lui dit que ce n'est pas encore prĂȘt Poll::Pending } }, } } } fn main() { let data = vec![ ("prefix_1_data1", "jeu"), ("prefix_1_data2", "de"), ("prefix_1_data3", "donnĂ©es 1"), ("prefix_2_data1", "le"), ("prefix_2_data2", "ciel"), ("prefix_2_data3", "est"), ("prefix_2_data4", "gris"), ("prefix_3_data1", "L'asynchrone"), ("prefix_3_data2", "ce n'est"), ("prefix_3_data3", "pas si"), ("prefix_3_data4", "compliquĂ©"), ]; smol::block_on(async move { let mut x = X::new(data); let mut now = Instant::now(); while let Some(message) = x.next().await { println!("le message \"{message}\" est arrivĂ©e en {:?}", now.elapsed()); now = Instant::now(); println!() } }) }
Qui a pour affichage
La clef prefix_1_data1 est arrivée en 308.0041ms
La clef prefix_1_data2 est arrivée en 1.1420077s
La clef prefix_1_data3 est arrivée en 1.251002s
La clef prefix_2_data1 est arrivée en 1.1130028s
le message "jeu de données 1" est arrivée en 3.8145345s
La clef prefix_2_data2 est arrivée en 354.0151ms
La clef prefix_2_data3 est arrivée en 669.003ms
La clef prefix_2_data4 est arrivée en 439.0069ms
La clef prefix_3_data1 est arrivée en 682.0036ms
le message "le ciel est gris" est arrivée en 2.1447636s
La clef prefix_3_data2 est arrivée en 239.0026ms
La clef prefix_3_data3 est arrivée en 1.1480031s
La clef prefix_3_data4 est arrivée en 1.6000024s
le message "L'asynchrone ce n'est pas si compliquĂ©" est arrivĂ©e en 2.9876462sOn a rien cassĂ© đ
API Haut niveau
Continuons Ă ne pas Ă©crire du code, câest assez reposant.
Nous allons nous créer un stream de nos futures.
unfold
Pour cela nous utilisons une fonction appelée unfold.
Celle-ci a pour signature
unfold(state, |mut state| async move {
// on fait des trucs
// on renvoit soit le prochain état avec la valeur généré
Some(data_courrante, next_state)
// soit on finit l'itération
})Nous, nous allons mettre ce que nous avion dans le
use futures_lite::stream::{unfold, StreamExt};
fn main()
// on utilise VecDeque pour pouvoir récupérer le premier élément
let data = VecDeque::from([...]);
let rng = rand::thread_rng();
// on donne le data et le rng comme état
let stream = unfold((data, rng), |(mut data, mut rng)| async move {
// on récupÚre le premier élément
data.pop_front()
.map(|(key, value)| {
let delay = rng.gen_range(100..2000);
// on créé la future
let fut = delivery(delay, (key.to_string(), value.to_string()));
// on box la future
let boxed_fut = fut.boxed();
(boxed_fut, (data, rng))
})
// ou on renvoit none
.or(None)
});
}Pour lâutiliser, cela donne
use std::pin::{pin, Pin};
fn main() {
// créé un Box::pin de notre stream
let mut stream = pin!(stream);
smol::block_on(async move {
while let Some(future) = stream.next().await {
let result = future.await;
println!("{:?}", result)
}
})
}Cool !
La clef prefix_1_data1 est arrivé en 1.9214232s
("prefix_1_data1", "jeu")
La clef prefix_1_data2 est arrivé en 251.7301ms
("prefix_1_data2", "de")
La clef prefix_1_data3 est arrivé en 1.9595308s
("prefix_1_data3", "données 1")
La clef prefix_2_data1 est arrivé en 1.9699186s
("prefix_2_data1", "le")
La clef prefix_2_data2 est arrivé en 1.6447015s
("prefix_2_data2", "ciel")
La clef prefix_2_data3 est arrivé en 202.1345ms
("prefix_2_data3", "est")
La clef prefix_2_data4 est arrivé en 897.8317ms
("prefix_2_data4", "gris")
La clef prefix_3_data1 est arrivé en 616.9607ms
("prefix_3_data1", "L'asynchrone")
La clef prefix_3_data2 est arrivé en 1.5978526s
("prefix_3_data2", "ce n'est")
La clef prefix_3_data3 est arrivé en 1.1858525s
("prefix_3_data3", "pas si")
La clef prefix_3_data4 est arrivé en 1.4951382s
("prefix_3_data4", "compliquĂ©")Mais ce nâest pas ce quâon veut, du coup on continue dans lâexploration de lâecosystĂšme
async-stream
Pour finir, je veux vous montrer une lib trĂšs pratique async-stream
cargo add async-streamElle fourni une macro stream! qui permet de faire ce que lâon dĂ©sire.
Le fonctionnement est simple.
Tout ce quâon yield est rajoutĂ© dans le stream.
async fn f() {
let stream = async_stream::stream! {
for i in 0..3 {
yield i;
}
};
let mut stream = pin!(stream); // needed for iteration
while let Some(value) = stream.next().await {
println!("got {}", value);
}
}Affiche
0
1
2Comment cela marche ?
Et bien en fait, si on démonte la macro, ça donne ceci
// on créé un tuyau
let (rx, tx) = channel();
// une partie est dans `AsyncStream`
let stream = AsyncStream::new(rx, async move {
for i in 0..3 {
// L'autre sert à envoyer des données
tx.send(Some(i)).await;
}
tx.send(None)
});A chaque fois que lâon yield on envoie dans un tuyau qui dĂ©bouche sur AsyncStream.
Et lui Ă lâautre bout dans son poll_next.
Il attend alors rx pour la prochaine valeur, jusquâau None.
Ainsi nous pouvons lâutiliser pour ce que lâon dĂ©sire
use async_io::Timer;
use futures_lite::stream::{unfold, StreamExt};
use futures_lite::FutureExt;
use rand::Rng;
use std::collections::VecDeque;
use std::pin::pin;
use std::time::{Duration, Instant};
// déclaration des types
type FutureInput = (String, String);
// future avec délai
async fn delivery(delay: u64, value: FutureInput) -> FutureInput {
let now = Instant::now();
Timer::after(Duration::from_millis(delay)).await;
println!("La clef {} est arrivé en {:?}", value.0, now.elapsed());
value
}
fn main() {
let data = VecDeque::from([
("prefix_1_data1", "jeu"),
("prefix_1_data2", "de"),
("prefix_1_data3", "données 1"),
("prefix_2_data1", "le"),
("prefix_2_data2", "ciel"),
("prefix_2_data3", "est"),
("prefix_2_data4", "gris"),
("prefix_3_data1", "L'asynchrone"),
("prefix_3_data2", "ce n'est"),
("prefix_3_data3", "pas si"),
("prefix_3_data4", "compliqué"),
]);
let rng = rand::thread_rng();
let stream = unfold((data, rng), |(mut data, mut rng)| async move {
data.pop_front()
.map(|(key, value)| {
let delay = rng.gen_range(100..2000);
// on créé la future
let fut = delivery(delay, (key.to_string(), value.to_string()));
// on box la future
let boxed_fut = fut.boxed();
//on retourne la valeur suivante et notre futur dans sa boĂźte
(boxed_fut, (data, rng))
})
.or(None)
});
let mut stream = pin!(stream);
let stream = async_stream::stream! {
let mut accumulator = vec![];
let mut prefix = None;
while let Some(message_futur) = stream.next().await {
let (key, value) = message_futur.await;
// on split sur la clef
let key = key.split('_').skip(1).take(1).collect::<String>();
// on vĂ©rifie si un prĂ©fix existe dĂ©jĂ
if prefix.is_none() {
// sinon on le créé
prefix = Some(key.clone())
}
// on défini un tableau pour un potentiel renvoie de données
let mut ready_data = vec![];
if let Some(ref actual_prefix) = prefix {
// Le prefix a changé nous pouvons relùcher une valeur
if &key != actual_prefix {
// on vide l'accumulateur
ready_data = accumulator.drain(..).collect();
// on initialise au prochain prefix
prefix = Some(key);
}
}
// on accumule le résultat
accumulator.push(value);
if !ready_data.is_empty() {
yield ready_data.join(" ")
}
}
// on oublie pas la derniÚre donnée
if !accumulator.is_empty() {
yield accumulator.join(" ")
}
};
let mut stream = pin!(stream);
smol::block_on(async move {
let mut now = Instant::now();
while let Some(message) = stream.next().await {
println!(
"le message \"{message}\" est arrivée en {:?}",
now.elapsed()
);
now = Instant::now();
println!()
}
})
}Adieu les implémentations, vous faites du Rust asynchrone !
Conclusion
Jâen ai pas vraiment, article beaucoup trop long, mais ça on a lâhabitude ^^
JâespĂšre quâil vous a plu.
A la prochaine â€ïž
