Asynchronie

Principe de base

En programmation, dès qu’on touche à du réseau, à l’utilisation de ressources externes et que l’on veut pouvoir servir plusieurs utilisateurs en même temps : faire les choses de façon asynchrone devient une bonne idée. C’est une alternative performante au multi-threading, au multi-processus ou au mutli-machine.

Aujourd’hui, l’asynchrone sur un seul fil d’exécution est la façon par défaut du JavaScript et du NodeJS de gerer plusieurs tâches en simultané. L’asynchrone nous permet de lancer une tâche, et tant que son résultat n’est pas là, le code qui en dépend est mis en attente, au profit d’autres tâches qui elles viennent de finir.

• Typiquement, mettons on reçoit deux requêtes en même temps sur une API.
• La première commence à être traitée et déclenche un appel asynchrone à la base de données.
• Le temps que la BDD réponde on commence à traiter la seconde requête, jusqu’à ce qu’elle déclenche aussi un appel asynchrone à la BDD.
• Maintenant le programme n’a plus rien à faire mais sais qu’il a deux tâches en attente qui dépendent chacune d’une réponse de la BDD.
• Le 1er appel à la BDD renvoie, on fini le traitement de la requête 1.
• Le second appel à la BDD renvoie, on fini le traitement de la requête 2.

Si dans cet exemple nous avions fait les choses de façon synchrone, la requête 2 n’aurais pu commencer à être traité seulement après que nous ayons complètement terminé avec la requête 1.

(cf. cette très bonne vidéo pour comprendre comment l’asynchrone fonctionne en javascript : Lydia Hallie • *Javascript Visualized - Event Loop, Web APIs (Micro)task Queue)

C’est le même principe que le projet clavardage avec poll. Mais avec moins d’efforts à fournir de votre côté.

Appliquée au rust

Le Rust intègre un mécanisme de programmation asynchrone. Certains crates fournissent des fonctions async sur lesquelles il faut .await pour obtenir un résultat. Pour pouvoir .await dans une fonction à nous, elle doit être marquée async. Dans un contexte normal, le main n’est pas async et ne peux donc pas exécuter de fonctions qui le sont.

Prennons pour exemple :

async fn my_add(a: i32, b: i32) -> i32 {
	println!("my_add called");
	a + b
}

Cette fonction est marquée async. Cela signifie qu’elle ne renvoie pas directement un i32 mais qu’elle renvoie un Future<Output = i32>.

Si on appelle la fonction ainsi :

let add_res = my_add(1, 2);

add_res ne contiendrait pas le résultat de l’opération et “my_add called” ne s’afficherait pas sur la sortie standard.

Pour que add_res contienne la valeur de retour et pour que le message s’affiche, il faut utiliser .await.

let add_res = my_add(1, 2).await;

CEPENDANT, pour avoir le droit de se servir d’await il nous manque un truc. Un runtime pour l’asynchrone. Le plus populaire c’est tokio.

cargo add tokio --features=full

#[tokio::main]
async fn main() {
	let add_res = my_add(1, 2).await;
	println!("{add_res}");
}

fn main() {
	// Create the runtime
	let rt = tokio::runtime::Runtime::new().unwrap();
	
	// Execute the future, blocking the current thread until completion
	rt.block_on(async {
		let add_res = my_add(1, 2).await;
		println!("{add_res}");
	});
}

tokio nous permet de :

• décorer un main avec #[tokio::main] pour :
  • créer automatiquement un runtime asynchrone, et
  • avoir le droit de le marquer async et d’exécuter du code qui doit se .await ; ou
• créer un tokio::runtime::Runtime pour sur l’exécution de code asynchrone dans un contexte qui ne l’est pas (exemple rt.block_on).

Des trucs rigolo avec tokio

Spawn

spawn permet d’ajouter une tâche à la liste des tâches en attente d’un runtime tokio existant. Elle ne bloque pas l’exécution et n’est même async elle-même.

La tâche, elle, est async et nous permet de bénéficier de tous les avantages de l’asynchrone.

Ce programme d’exemple fait s’afficher de façon perpétuelle :

• “tick!” puis “tock!” toutes les secondes ; ET en même temps
• “fliiiiiip” puis “flooooop!” toutes les 5 secondes.

#[tokio::main]
async fn main() {
    interval_flip_flop(1, "tick!", "tock!");
    interval_flip_flop(5, "fliiiiiip", "flooooop!");
}

fn interval_flip_flop(secs: u64, msg_1: &'static str, msg_2: &'static str) {
    use tokio::time::{interval, Duration};
    tokio::spawn(async move {
        let mut interval = interval(Duration::from_secs(secs));
        loop {
            println!("{msg_1}");
            interval.tick().await;
            println!("{msg_2}");
            interval.tick().await;
        }
    });
}

Keep pushing

En parallèle de l’exemple précédent, on pourrait lancer un serveur TCP qui echo les messages de clients connectés (c’est du jamais vu).

#[tokio::main]
async fn main() {
    interval_flip_flop(1, "tick!", "tock!");
    interval_flip_flop(5, "fliiiiiip", "flooooop!");
    let tcp_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let localaddr = tcp_listener.local_addr().unwrap();
    println!("{}", localaddr);
    server(tcp_listener).await;
}

fn interval_flip_flop(secs: u64, msg_1: &'static str, msg_2: &'static str) {
    use tokio::time::{interval, Duration};
    tokio::spawn(async move {
        let mut interval = interval(Duration::from_secs(secs));
        loop {
            println!("{msg_1}");
            interval.tick().await;
            println!("{msg_2}");
            interval.tick().await;
        }
    });
}

async fn server(tcp_listener: tokio::net::TcpListener) {
    let mut last_client_id = 0;
    loop {
        if let Ok((stream, _)) = tcp_listener.accept().await {
            // Spawn a new task that echoes what the client says.
            last_client_id += 1;
            let this_client_id = last_client_id;
            tokio::spawn(async move {
                echo_this_client(stream, this_client_id).await;
            });
        }
    }
}

async fn echo_this_client(mut socket: tokio::net::TcpStream, client_id: u32) {
    let mut buf = [0; 1024];
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    println!("client: {client_id} joined.");
    let msg_welcome = format!("hello: {client_id}\n");
    if let Err(err) = socket.write(msg_welcome.as_bytes()).await {
        eprintln!("failed to write {err} to {client_id}");
        return;
    }

    while let Ok(size_read) = socket.read(&mut buf).await {
        if let Err(err) = socket.write(&buf[..size_read]).await {
            eprintln!("failed to write {err} to {client_id}");
        }
        if size_read == 0 {
            break;
        }
    }
    println!("client: {client_id} left.");
}