Tutorial Rust: Introducción a Tokio

Setup de Tokio

Tokio es el framework asíncrono más utilizado en el ecosistema de Rust, proporcionando las herramientas necesarias para construir aplicaciones concurrentes de alto rendimiento. Para comenzar a utilizar Tokio en nuestros proyectos, necesitamos configurar correctamente nuestro entorno de desarrollo.

Añadiendo Tokio a nuestro proyecto

Lo primero que debemos hacer es incluir Tokio como dependencia en nuestro archivo Cargo.toml. Tokio está diseñado de forma modular, lo que nos permite incluir solo las características que necesitamos para nuestro proyecto:

[dependencies]
tokio = { version = "1.36", features = ["full"] }

En este ejemplo, estamos utilizando la característica "full" que incluye todas las funcionalidades de Tokio. Sin embargo, en proyectos reales es recomendable seleccionar solo las características específicas que necesitamos para reducir el tamaño de la compilación:

[dependencies]
tokio = { version = "1.36", features = ["rt", "rt-multi-thread", "macros", "time", "net", "fs"] }

Las características más comunes incluyen:

• rt: El runtime básico de Tokio
• rt-multi-thread: Soporte para ejecución en múltiples hilos
• macros: Incluye el macro #[tokio::main] y otros útiles
• time: Funcionalidades relacionadas con tiempo y temporizadores
• net: Operaciones de red asíncronas (TCP, UDP)
• fs: Operaciones de sistema de archivos asíncronas
• io-util: Utilidades de I/O asíncronas

Configurando el punto de entrada

Una vez añadida la dependencia, necesitamos configurar el punto de entrada de nuestra aplicación para utilizar el runtime de Tokio. La forma más sencilla es mediante el macro #[tokio::main]:

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("¡Hola desde Tokio!");
    
    // Código asíncrono aquí
    
    Ok(())
}

Este macro transforma nuestra función main asíncrona en una función sincrónica que inicializa el runtime de Tokio y ejecuta nuestro código asíncrono dentro de él. Internamente, el macro expande a algo similar a:

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let runtime = tokio::runtime::Runtime::new()?;
    runtime.block_on(async {
        println!("¡Hola desde Tokio!");
        
        // Código asíncrono aquí
        
        Ok(())
    })
}

Configuración manual del runtime

En algunos casos, podemos necesitar más control sobre la configuración del runtime. Podemos crear manualmente una instancia del runtime utilizando el builder pattern:

use tokio::runtime::Runtime;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Crear un runtime con configuración personalizada
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(4)         // Número de worker threads
        .enable_io()               // Habilitar I/O asíncrono
        .enable_time()             // Habilitar temporizadores
        .thread_name("mi-servicio") // Nombre para los threads
        .build()?;
    
    // Ejecutar código asíncrono en el runtime
    runtime.block_on(async {
        println!("Runtime personalizado iniciado");
        
        // Código asíncrono aquí
        
        Ok(())
    })
}

Esta aproximación nos permite ajustar parámetros como:

• El número de worker threads en el pool
• Habilitar o deshabilitar características específicas
• Configurar nombres personalizados para los threads
• Establecer hooks para eventos del ciclo de vida de los threads

Integrando Tokio en bibliotecas

Cuando desarrollamos bibliotecas que utilizan Tokio, es importante no asumir un runtime específico. En lugar de usar #[tokio::main], debemos exponer funciones asíncronas que el usuario pueda ejecutar en su propio runtime:

// En nuestra biblioteca
pub async fn procesar_datos(datos: &[u8]) -> Result<Vec<u8>, std::io::Error> {
    // Procesamiento asíncrono
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
    Ok(datos.to_vec())
}

// No incluir esto en una biblioteca:
// #[tokio::main]
// pub async fn ejecutar() { ... }

Para las dependencias en bibliotecas, es recomendable hacer opcionales las características de Tokio:

[dependencies]
tokio = { version = "1.36", features = ["rt"], optional = true }

[features]
tokio-runtime = ["tokio"]

Estructura básica de un proyecto con Tokio

Un proyecto típico con Tokio suele seguir esta estructura:

mi-proyecto/
├── Cargo.toml
├── src/
│   ├── main.rs       # Punto de entrada con #[tokio::main]
│   ├── error.rs      # Tipos de error personalizados
│   ├── config.rs     # Configuración de la aplicación
│   └── handlers/     # Manejadores de eventos asíncronos

Veamos un ejemplo básico de un servidor TCP echo que utiliza Tokio:

use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Crear un listener TCP en localhost:8080
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Servidor escuchando en 127.0.0.1:8080");

    // Aceptar conexiones en bucle
    loop {
        let (socket, addr) = listener.accept().await?;
        
        // Spawn de una nueva tarea para manejar la conexión
        tokio::spawn(async move {
            println!("Nueva conexión: {}", addr);
            if let Err(e) = procesar_conexion(socket).await {
                println!("Error procesando conexión: {}", e);
            }
        });
    }
}

async fn procesar_conexion(mut socket: TcpStream) -> Result<(), Box<dyn std::error::Error>> {
    let mut buffer = vec![0; 1024];
    
    loop {
        let n = socket.read(&mut buffer).await?;
        
        // Si leemos 0 bytes, el cliente cerró la conexión
        if n == 0 {
            return Ok(());
        }
        
        // Escribir los datos de vuelta (echo)
        socket.write_all(&buffer[0..n]).await?;
    }
}

Herramientas de diagnóstico

Al configurar Tokio, es útil conocer algunas herramientas de diagnóstico:

• tokio-console: Una herramienta de diagnóstico para inspeccionar y depurar aplicaciones Tokio en tiempo de ejecución.

[dependencies]
tokio = { version = "1.36", features = ["full", "tracing"] }
console-subscriber = "0.2"

#[tokio::main]
async fn main() {
    console_subscriber::init();
    
    // Resto del código
}

• tracing: Para instrumentar el código y obtener información detallada sobre la ejecución.

[dependencies]
tokio = { version = "1.36", features = ["full"] }
tracing = "0.1"
tracing-subscriber = "0.3"

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt::init();
    
    tracing::info!("Aplicación iniciada");
    // Resto del código
}

Con esta configuración básica, estamos listos para comenzar a desarrollar aplicaciones asíncronas con Tokio, aprovechando su eficiente modelo de concurrencia y sus potentes abstracciones para I/O asíncrono.

Multi-threaded runtime

El runtime multi-hilo de Tokio constituye uno de los pilares fundamentales que hacen de este framework una herramienta tan eficiente para la programación asíncrona en Rust. A diferencia de los runtimes de un solo hilo, Tokio aprovecha múltiples núcleos de CPU para ejecutar tareas asíncronas en paralelo.

Arquitectura del runtime multi-hilo

El runtime multi-hilo de Tokio está compuesto por varios componentes clave:

• Un thread pool (grupo de hilos) de trabajadores
• Un programador cooperativo (scheduler)
• Colas de tareas para distribuir el trabajo
• Mecanismos de robo de trabajo (work-stealing)

Cuando iniciamos el runtime multi-hilo de Tokio, este crea un conjunto de hilos de sistema operativo que forman el thread pool. Por defecto, Tokio configura el número de hilos basándose en la cantidad de núcleos lógicos disponibles en el sistema:

// El runtime multi-hilo se crea automáticamente con #[tokio::main]
#[tokio::main]
async fn main() {
    // En este punto, ya tenemos un runtime multi-hilo funcionando
    println!("Ejecutando en el runtime multi-hilo de Tokio");
}

Funcionamiento interno del scheduler

El scheduler (programador) de Tokio es responsable de distribuir las tareas asíncronas entre los hilos disponibles. Utiliza un enfoque de programación cooperativa, donde las tareas voluntariamente ceden el control cuando esperan por operaciones de I/O u otros eventos.

Cuando una tarea llega a un punto .await, el scheduler puede suspenderla y cambiar a otra tarea lista para ejecutarse. Este modelo permite un alto grado de concurrencia sin los costos asociados al cambio de contexto de hilos del sistema operativo:

async fn tarea_ejemplo() {
    // Esta tarea se ejecuta hasta que llega a un punto de espera
    println!("Iniciando tarea");
    
    // Al llegar a .await, la tarea puede ser suspendida
    // permitiendo que el hilo ejecute otras tareas
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
    
    // Cuando la operación de espera completa, la tarea continúa
    println!("Tarea completada");
}

Creación y distribución de tareas

Para ejecutar código asíncrono en el runtime multi-hilo, utilizamos la función tokio::spawn(). Esta función crea una nueva tarea asíncrona y la programa para su ejecución en cualquiera de los hilos del pool:

#[tokio::main]
async fn main() {
    // Creamos 5 tareas que se ejecutarán concurrentemente
    for i in 0..5 {
        tokio::spawn(async move {
            // Esta tarea podría ejecutarse en cualquier hilo del pool
            println!("Tarea {} ejecutándose", i);
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            println!("Tarea {} completada", i);
        });
    }
    
    // Esperamos un poco para que las tareas tengan tiempo de ejecutarse
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
}

Cada tarea creada con spawn es una unidad independiente de trabajo que puede ejecutarse en cualquier hilo del pool. El scheduler se encarga de balancear la carga entre los hilos disponibles.

Mecanismo de work-stealing

Una característica clave del runtime multi-hilo de Tokio es su algoritmo de robo de trabajo (work-stealing). Cuando un hilo del pool termina de procesar todas las tareas en su cola local, puede "robar" tareas de las colas de otros hilos que están más ocupados.

Este mecanismo garantiza una distribución eficiente del trabajo y evita situaciones donde algunos hilos están ociosos mientras otros están sobrecargados:

Thread 1: [Tarea A, Tarea B, Tarea C, Tarea D]  ← Thread 3 (sin tareas) roba Tarea D
Thread 2: [Tarea E, Tarea F]
Thread 3: []  → Roba Tarea D de Thread 1

Localidad y afinidad de tareas

Aunque las tareas pueden ejecutarse en cualquier hilo del pool, Tokio intenta mantener cierta localidad para mejorar la eficiencia. Las subtareas creadas dentro de una tarea tienen mayor probabilidad de ejecutarse en el mismo hilo, lo que puede mejorar el rendimiento al aprovechar la localidad de caché:

#[tokio::main]
async fn main() {
    tokio::spawn(async {
        println!("Tarea principal");
        
        // Esta subtarea tiene mayor probabilidad de ejecutarse
        // en el mismo hilo que la tarea principal
        let subtarea = tokio::spawn(async {
            println!("Subtarea ejecutándose");
        });
        
        // Esperamos a que la subtarea termine
        subtarea.await.unwrap();
    });
    
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
}

Bloqueo y worker threads dedicados

Un aspecto importante a considerar es que las operaciones bloqueantes pueden afectar negativamente al rendimiento del runtime multi-hilo. Si una tarea realiza una operación que bloquea el hilo (como una llamada a sistema sincrónica o un cálculo intensivo), ese hilo no podrá procesar otras tareas hasta que la operación termine.

Para manejar operaciones bloqueantes, Tokio proporciona tokio::task::spawn_blocking:

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Simulamos una operación bloqueante (como leer un archivo grande de forma sincrónica)
    let resultado = tokio::task::spawn_blocking(|| {
        // Esta operación se ejecuta en un hilo dedicado para operaciones bloqueantes
        println!("Realizando operación bloqueante...");
        
        // Simulamos un cálculo intensivo
        let mut suma = 0;
        for i in 0..1_000_000 {
            suma += i;
        }
        
        suma
    }).await?;
    
    println!("Resultado de la operación bloqueante: {}", resultado);
    Ok(())
}

spawn_blocking ejecuta la operación en un pool separado de hilos dedicados a tareas bloqueantes, evitando así interferir con el pool principal de worker threads.

Monitoreo y ajuste del runtime

Para aplicaciones con requisitos específicos, podemos monitorear y ajustar el comportamiento del runtime multi-hilo:

use tokio::runtime::Builder;
use std::time::Duration;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Creamos un runtime personalizado
    let runtime = Builder::new_multi_thread()
        .worker_threads(8)                      // Fijamos 8 worker threads
        .thread_keep_alive(Duration::from_secs(60))  // Tiempo de vida de threads inactivos
        .thread_stack_size(3 * 1024 * 1024)    // 3MB de stack por thread
        .build()?;
    
    // Ejecutamos nuestro código asíncrono en el runtime personalizado
    runtime.block_on(async {
        println!("Ejecutando en runtime personalizado");
        
        // Creamos múltiples tareas para aprovechar los 8 worker threads
        let mut handles = vec![];
        for i in 0..20 {
            let handle = tokio::spawn(async move {
                tokio::time::sleep(tokio::time::Duration::from_millis(100 * i)).await;
                println!("Tarea {} completada", i);
                i
            });
            handles.push(handle);
        }
        
        // Esperamos a que todas las tareas terminen
        for handle in handles {
            let resultado = handle.await?;
            println!("Resultado: {}", resultado);
        }
        
        Ok::<_, Box<dyn std::error::Error>>(())
    })
}

Casos de uso del runtime multi-hilo

El runtime multi-hilo de Tokio es especialmente adecuado para ciertos escenarios:

• Servidores web que manejan miles de conexiones concurrentes
• Procesamiento paralelo de datos con operaciones de I/O frecuentes
• Aplicaciones de microservicios con múltiples puntos de integración
• Sistemas de procesamiento de eventos que requieren baja latencia

Un ejemplo práctico sería un servidor que procesa múltiples solicitudes en paralelo:

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Servidor escuchando en 127.0.0.1:8080");
    
    // Contador de solicitudes procesadas por cada worker thread
    let mut contador_por_thread = vec![0; num_cpus::get()];
    
    loop {
        let (mut socket, addr) = listener.accept().await?;
        println!("Nueva conexión: {}", addr);
        
        // Cada conexión se maneja en una tarea separada
        tokio::spawn(async move {
            let thread_id = std::thread::current().id();
            println!("Manejando conexión de {} en thread {:?}", addr, thread_id);
            
            let mut buffer = vec![0; 1024];
            
            // Leemos datos del cliente
            match socket.read(&mut buffer).await {
                Ok(n) if n > 0 => {
                    // Procesamos los datos (ejemplo: convertir a mayúsculas)
                    let respuesta = String::from_utf8_lossy(&buffer[0..n])
                        .to_uppercase()
                        .into_bytes();
                    
                    // Enviamos respuesta
                    if let Err(e) = socket.write_all(&respuesta).await {
                        println!("Error al escribir en socket: {}", e);
                    }
                }
                Ok(_) => println!("Conexión cerrada por el cliente"),
                Err(e) => println!("Error al leer del socket: {}", e),
            }
        });
    }
}

En este ejemplo, el runtime multi-hilo distribuye automáticamente las conexiones entrantes entre los worker threads disponibles, permitiendo que el servidor maneje múltiples clientes simultáneamente sin bloquear.

Comparación con el runtime de un solo hilo

Tokio también ofrece un runtime de un solo hilo que puede ser más eficiente para ciertos casos de uso:

// Creación de un runtime de un solo hilo
let runtime = tokio::runtime::Builder::new_current_thread()
    .enable_all()
    .build()?;

La elección entre multi-hilo y un solo hilo depende de varios factores:

• Multi-hilo: Mejor para aplicaciones con muchas operaciones en paralelo y múltiples núcleos disponibles.
• Un solo hilo: Más eficiente para aplicaciones con pocas tareas concurrentes o entornos con recursos limitados.

El runtime multi-hilo introduce una pequeña sobrecarga por la coordinación entre hilos, pero ofrece un mejor aprovechamiento de los recursos del sistema en aplicaciones con alta concurrencia.

I/O asíncrono

El I/O asíncrono es una de las principales fortalezas de Tokio, permitiendo realizar operaciones de entrada/salida sin bloquear los hilos de ejecución. Esto resulta fundamental para desarrollar aplicaciones de alto rendimiento que puedan manejar miles de conexiones concurrentes con recursos limitados.

Fundamentos del I/O asíncrono en Tokio

Tokio proporciona abstracciones asíncronas para las operaciones de I/O más comunes, reemplazando las llamadas bloqueantes de la biblioteca estándar con alternativas que pueden suspenderse y reanudarse sin bloquear el hilo subyacente:

// I/O bloqueante (biblioteca estándar)
let mut file = std::fs::File::open("archivo.txt")?;
let mut contenido = String::new();
file.read_to_string(&mut contenido)?;

// I/O asíncrono (Tokio)
let mut file = tokio::fs::File::open("archivo.txt").await?;
let mut contenido = String::new();
file.read_to_string(&mut contenido).await?;

Internamente, Tokio utiliza diferentes estrategias de I/O asíncrono según el sistema operativo:

• En Linux: epoll
• En macOS/BSD: kqueue
• En Windows: IOCP (I/O Completion Ports)

Estas APIs del sistema operativo permiten monitorear múltiples descriptores de archivo sin bloquear hilos.

Operaciones de red asíncronas

Las operaciones de red son probablemente el caso de uso más común para I/O asíncrono. Tokio proporciona implementaciones asíncronas para los protocolos TCP y UDP:

TCP asíncrono

Para crear un servidor TCP asíncrono:

use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};

async fn ejecutar_servidor() -> Result<(), Box<dyn std::error::Error>> {
    // Crear un listener TCP
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    
    loop {
        // Aceptar conexiones (operación asíncrona)
        let (socket, addr) = listener.accept().await?;
        
        // Manejar cada conexión en una tarea separada
        tokio::spawn(async move {
            manejar_conexion(socket, addr).await
        });
    }
}

async fn manejar_conexion(mut socket: TcpStream, addr: std::net::SocketAddr) -> Result<(), Box<dyn std::error::Error>> {
    println!("Cliente conectado: {}", addr);
    
    let mut buffer = vec![0; 1024];
    
    // Leer datos del cliente (operación asíncrona)
    let n = socket.read(&mut buffer).await?;
    
    // Procesar datos...
    
    // Escribir respuesta (operación asíncrona)
    socket.write_all(&buffer[0..n]).await?;
    
    Ok(())
}

Para crear un cliente TCP asíncrono:

use tokio::net::TcpStream;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

async fn conectar_a_servidor() -> Result<(), Box<dyn std::error::Error>> {
    // Conectar al servidor (operación asíncrona)
    let mut stream = TcpStream::connect("127.0.0.1:8080").await?;
    
    // Enviar datos (operación asíncrona)
    stream.write_all(b"Hola servidor").await?;
    
    // Leer respuesta (operación asíncrona)
    let mut buffer = vec![0; 1024];
    let n = stream.read(&mut buffer).await?;
    
    println!("Respuesta: {}", String::from_utf8_lossy(&buffer[0..n]));
    
    Ok(())
}

UDP asíncrono

Tokio también proporciona soporte para comunicación UDP asíncrona:

use tokio::net::UdpSocket;

async fn ejemplo_udp() -> Result<(), Box<dyn std::error::Error>> {
    // Crear socket UDP
    let socket = UdpSocket::bind("127.0.0.1:8081").await?;
    
    // Enviar datos (operación asíncrona)
    socket.send_to(b"Mensaje UDP", "127.0.0.1:8082").await?;
    
    // Recibir datos (operación asíncrona)
    let mut buffer = vec![0; 1024];
    let (n, origen) = socket.recv_from(&mut buffer).await?;
    
    println!("Recibido desde {}: {}", origen, String::from_utf8_lossy(&buffer[0..n]));
    
    Ok(())
}

Operaciones de archivo asíncronas

Tokio proporciona el módulo tokio::fs para realizar operaciones de archivo de forma asíncrona:

use tokio::fs::File;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

async fn operaciones_archivo() -> Result<(), Box<dyn std::error::Error>> {
    // Abrir archivo para lectura (operación asíncrona)
    let mut archivo = File::open("entrada.txt").await?;
    
    // Leer contenido (operación asíncrona)
    let mut contenido = String::new();
    archivo.read_to_string(&mut contenido).await?;
    
    // Procesar contenido
    let contenido_procesado = contenido.to_uppercase();
    
    // Crear archivo para escritura (operación asíncrona)
    let mut salida = File::create("salida.txt").await?;
    
    // Escribir contenido (operación asíncrona)
    salida.write_all(contenido_procesado.as_bytes()).await?;
    
    Ok(())
}

Otras operaciones asíncronas de sistema de archivos incluyen:

// Verificar si un archivo existe
let existe = tokio::fs::try_exists("archivo.txt").await?;

// Obtener metadatos de un archivo
let metadata = tokio::fs::metadata("archivo.txt").await?;

// Crear un directorio
tokio::fs::create_dir("nuevo_directorio").await?;

// Leer entradas de un directorio
let mut entradas = tokio::fs::read_dir(".").await?;
while let Some(entrada) = entradas.next_entry().await? {
    println!("Entrada: {}", entrada.file_name().to_string_lossy());
}

Temporizadores y tiempo asíncrono

El módulo tokio::time proporciona utilidades para trabajar con tiempo de forma asíncrona:

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

async fn ejemplo_temporizadores() -> Result<(), Box<dyn std::error::Error>> {
    // Esperar de forma asíncrona
    println!("Esperando 1 segundo...");
    sleep(Duration::from_secs(1)).await;
    println!("¡Espera completada!");
    
    // Establecer un timeout para una operación asíncrona
    match timeout(Duration::from_millis(500), operacion_lenta()).await {
        Ok(resultado) => println!("Operación completada: {:?}", resultado),
        Err(_) => println!("¡La operación excedió el timeout!"),
    }
    
    // Crear un intervalo para ejecutar código periódicamente
    let mut intervalo = tokio::time::interval(Duration::from_secs(2));
    
    // Primera llamada a .tick().await consume el tick inicial inmediatamente
    intervalo.tick().await;
    
    // Las siguientes llamadas esperarán el intervalo completo
    for _ in 0..3 {
        intervalo.tick().await;
        println!("Tick del intervalo");
    }
    
    Ok(())
}

async fn operacion_lenta() -> String {
    sleep(Duration::from_secs(2)).await;
    "Resultado de operación lenta".to_string()
}

Streams asíncronos

Los streams son el equivalente asíncrono a los iteradores. Representan una secuencia de valores que se producen de forma asíncrona:

use tokio_stream::StreamExt;
use tokio::net::TcpListener;

async fn ejemplo_stream() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    
    // Convertir el listener en un stream de conexiones
    let mut connection_stream = tokio_stream::wrappers::TcpListenerStream::new(listener);
    
    // Procesar cada conexión a medida que llega
    while let Some(socket_result) = connection_stream.next().await {
        let socket = socket_result?;
        println!("Nueva conexión desde: {}", socket.peer_addr()?);
        
        // Manejar la conexión...
    }
    
    Ok(())
}

Para usar streams, necesitamos añadir la dependencia tokio-stream:

[dependencies]
tokio-stream = "0.1"

Canales asíncronos

Los canales son una herramienta fundamental para la comunicación entre tareas asíncronas. Tokio proporciona varios tipos de canales:

use tokio::sync::mpsc;
use tokio::sync::oneshot;

async fn ejemplo_canales() {
    // Canal mpsc (multi-productor, single-consumidor)
    let (tx, mut rx) = mpsc::channel(32); // Buffer de 32 mensajes
    
    // Enviar mensajes desde múltiples tareas
    for i in 0..3 {
        let tx_clone = tx.clone();
        tokio::spawn(async move {
            for j in 0..5 {
                if tx_clone.send(format!("Tarea {} - Mensaje {}", i, j)).await.is_err() {
                    println!("Receptor cerrado");
                    return;
                }
            }
        });
    }
    
    // Descartar el transmisor original para que el canal pueda cerrarse
    drop(tx);
    
    // Recibir mensajes
    while let Some(mensaje) = rx.recv().await {
        println!("Recibido: {}", mensaje);
    }
    
    // Canal oneshot (para una única respuesta)
    let (tx_one, rx_one) = oneshot::channel();
    
    tokio::spawn(async move {
        // Simular algún procesamiento
        sleep(Duration::from_millis(500)).await;
        
        // Enviar el resultado (consume el transmisor)
        let _ = tx_one.send("Resultado de la operación");
    });
    
    // Esperar el resultado
    match rx_one.await {
        Ok(resultado) => println!("Resultado recibido: {}", resultado),
        Err(_) => println!("El transmisor fue descartado"),
    }
}

Ejemplo práctico: Servidor de chat simple

Veamos un ejemplo más completo que combina varias características de I/O asíncrono de Tokio para crear un servidor de chat básico:

use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;
use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader, AsyncBufReadExt};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Crear un listener TCP
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Servidor de chat iniciado en 127.0.0.1:8080");
    
    // Canal de broadcast para distribuir mensajes a todos los clientes
    let (tx, _) = broadcast::channel(100);
    
    loop {
        // Aceptar nuevas conexiones
        let (socket, addr) = listener.accept().await?;
        
        // Clonar el transmisor para esta conexión
        let tx_clone = tx.clone();
        // Crear un receptor para esta conexión
        let mut rx = tx.subscribe();
        
        // Manejar cada cliente en una tarea separada
        tokio::spawn(async move {
            // Dividir el socket en reader y writer
            let (reader, mut writer) = socket.into_split();
            let mut reader = BufReader::new(reader);
            let mut linea = String::new();
            
            // Enviar mensaje de bienvenida
            let mensaje_bienvenida = format!("Bienvenido al chat! Tu dirección es: {}\n", addr);
            writer.write_all(mensaje_bienvenida.as_bytes()).await.unwrap();
            
            // Crear dos tareas: una para leer mensajes del cliente, otra para enviarle mensajes
            
            // Tarea 1: Leer mensajes del cliente y distribuirlos
            let tx_clone2 = tx_clone.clone();
            let mut read_task = tokio::spawn(async move {
                loop {
                    linea.clear();
                    // Leer una línea del cliente
                    if reader.read_line(&mut linea).await.unwrap() == 0 {
                        // Conexión cerrada
                        break;
                    }
                    
                    // Formatear y distribuir el mensaje
                    let mensaje = format!("{}: {}", addr, linea.trim());
                    if tx_clone2.send(mensaje).is_err() {
                        // Error al enviar, probablemente no hay receptores
                        break;
                    }
                }
            });
            
            // Tarea 2: Recibir mensajes del canal y enviarlos al cliente
            let mut send_task = tokio::spawn(async move {
                while let Ok(mensaje) = rx.recv().await {
                    // No enviar al cliente sus propios mensajes
                    if !mensaje.starts_with(&addr.to_string()) {
                        writer.write_all(mensaje.as_bytes()).await.unwrap();
                        writer.write_all(b"\n").await.unwrap();
                    }
                }
            });
            
            // Esperar a que cualquiera de las dos tareas termine
            tokio::select! {
                _ = &mut read_task => send_task.abort(),
                _ = &mut send_task => read_task.abort(),
            }
            
            println!("Cliente desconectado: {}", addr);
        });
    }
}

Este servidor de chat demuestra varias características clave del I/O asíncrono en Tokio:

• Aceptación asíncrona de conexiones TCP
• Lectura y escritura asíncrona de datos
• Uso de canales broadcast para distribuir mensajes
• Tareas concurrentes para manejar múltiples aspectos de cada conexión
• Uso de select! para esperar múltiples operaciones asíncronas

Mejores prácticas para I/O asíncrono

Al trabajar con I/O asíncrono en Tokio, es importante seguir estas prácticas:

• Evitar bloqueos: Nunca uses operaciones bloqueantes dentro de tareas asíncronas. Utiliza spawn_blocking si necesitas realizar operaciones sincrónicas.

• Manejar errores adecuadamente: Las operaciones de I/O pueden fallar por muchas razones. Asegúrate de manejar los errores correctamente.

• Limitar recursos: Establece límites para buffers, conexiones y otros recursos para evitar ataques de denegación de servicio.

• Usar timeouts: Aplica timeouts a las operaciones de I/O para evitar que tareas se queden esperando indefinidamente.

// Aplicar timeout a una operación de I/O
let resultado = tokio::time::timeout(
    Duration::from_secs(5),
    socket.read(&mut buffer)
).await??; // Nota el doble ? para manejar tanto el error de timeout como el de I/O

• Considerar el backpressure: Implementa mecanismos para controlar la velocidad a la que se procesan los datos cuando los productores son más rápidos que los consumidores.

• Agrupar operaciones pequeñas: Para operaciones de I/O muy pequeñas, considera agruparlas para reducir la sobrecarga del sistema.

El I/O asíncrono de Tokio proporciona las herramientas necesarias para construir aplicaciones de red y sistemas distribuidos altamente eficientes, aprovechando al máximo los recursos disponibles mientras se mantiene un modelo de programación claro y expresivo.