the ugly org - backend - fundamentos del Dao De Jing (道德經)

desu #1 20 Nov

The way you can go
isn’t the real way.
The name you can say
isn’t the real name.

Heaven and earth
begin in the unnamed:
name’s the mother
of the ten thousand things.

So the unwanting soul
sees what’s hidden,
and the ever-wanting soul
sees only what it wants.

Two things, one origin,
but different in name,
whose identity is mystery.
Mystery of all mysteries!
The door to the hidden.

1
Steps:
1. Create our socket
2. Bind it to an address
3. Put it in "server" mode (i.e. call listen on it)
4. Accept a connection
5. Application logic involving reading/writing/closing to the socket
6. Goto step 4

2
Steps:
1. Create our socket
2. Bind it to an address
3. Put it in "server" mode (i.e. call listen on it)
4. Accept a connection in a loop
5. In a new thread: Application logic involving reading/writing/closing the socket
6. Goto step 4

3
Steps:
1. Create our socket
2. Bind it to an address
3. Put it in "server" mode (i.e. call listen on it)
4. Create an event queue
5. Add the server socket to the event queue
6. Wait for events in a loop
7. Application logic involving reading/writing/closing the socket
8. Remove socket from queue when done
9. Goto step 6

4
Steps:
1. Create our socket
2. Bind it to an address
3. Put it in "server" mode (i.e. call listen on it)
4. Create an event queue
5. Add the server socket to the event queue
6. Create a thread pool
7. Wait for events in a loop
8. Each thread does application logic involving reading/writing/closing the socket
8. Remove socket from queue when done
9. Goto step 6

4

desu #2 20 Nov

Vamos a empezar con 1.

1 Steps:
1. Create our socket
2. Bind it to an address
3. Put it in "server" mode (i.e. call listen on it)
4. Accept a connection
5. Application logic involving reading/writing/closing to the socket
6. Goto step 4

Escribimos:

Y ejecutamos:

El servidor se desconecta. No podemos seguir mandando mensajes y solo respondemos a una conexión.

Añadimos un loop:

Ahora podemos mandar multiples mensajes:

Pero el servidor, no detecta que la conexión ha terminado, y hace un loop infinito para un socket cerrado que siempre devuelve [].

Si leemos la documentación de read, si n == 0, probablemente, podemos cerrar.

Ahora el servidor termina cuando el cliente termina la conexión.

Vamos a arreglar el print en servidor del mensaje, convirtiéndolo a string:

Good. Ahora vamos a mover el loop, para que en lugar de terminar, acepte una nueva conexión.

Y ahora el servidor puede aceptar multiples clientes de manera secuencial.

Seguro? Quizas no, el bucle no esta bien, el codigo espera que el ciente mande un mensaje, se desconecte, se vuelva a conectar, mande un mensaje... no puede leer multiples mensajes.

Añadimos otro loop para leer, donde lo teniamos antes, y voy a añadir un goto para salir, los gotos son geniales. Solo los malos programadores les tienen miedo porque no saben usarlos.

Y ahora si:

Ahora si? Seguro?

Nada es seguro en esta vida excepto la muerte. He cambiado el tamaño del buffer en el que leemos a 2 para que se vea mas facil:

Si nos llega un mensaje mas grande de lo que podemos leer, llenaremos el buffer, lo escribiremos, y volveremos a leer, escribiremos y volveremos a leer.

El motivo por el cual en la terminal nuestro cliente ve el mensaje completo, es por los '\n' y como ncat tiene un buffer. Segun el tipo de mensajes que enviemos este comportamiento es incorrecto. Lo que se conoce como el problema: message boundery. Que es un mensaje? Como se que he recibido un mensaje? Normalmente los protocolos tienen un campo acordado que indica la longitud esperada del mensaje. Y aqui entraria el problema de errores y ataques, que vamos a ignorar.

Problemas:

Message boundery, como leer y escribir un socket
Como tener multiples sockets

True goodness
is like water.
Water’s good
for everything.
It doesn’t compete.

It goes right
to the low loathsome places,
and so finds the way.

desu #3 20 Nov

Voy a hacer una solución multi-threading.

2
Steps:
1. Create our socket
2. Bind it to an address
3. Put it in "server" mode (i.e. call listen on it)
4. Accept a connection in a loop
5. In a new thread: Application logic involving reading/writing/closing the socket
6. Goto step 4

Para la mayoría de aplicaciones CPU-bound esto es mas que suficiente.

Y ya tenemos multiples clientes

Si ahora tiro un benchmark, de clientes que envían peticiones HTTP, creo que mucha gente se sorprendería con el numero de peticiones por segundo que somos capaces de lograr. En eficiencia y rendimiento, menos es mas.

https://github.com/patrykstefanski/async-bench

framework	reqs/s
threads	930,142
Boost.Asio	1,011,690
Go	1,265,821
Tokio	1,120,036
async-std	974,002

Como vimos en otro diario: https://github.com/vrnvu/rust-minikeyvalue, Go > Tokio
Pero es que con threads tenemos casi 1M req/s :-) Un approx 75% lo que ofrece Go que seria el sweet spot.

Podemos aprovechar este momento para limpiar algunas cosas, el loop con el break, como esta dentro de un thread puede ser un return.

Y Rust te proporciona un iterador lazy sobre los sockets, de esta manera en lugar de un loop infinito podemos tener un for.

Y un problema seria establecer cual es el numero maximo de threads que queremos spawnear. Esto siempre lo debemos fijar como upper limit segundo el numero de CPU y threads fisicos de nuestro sistema. Voy a poner 1 numero maximo de threads. Cuando llegue una peticion que no pueda procesarse, se bloqueara, y mi sistema actuara como un FIFO. Los mensajes igualmente llegan a nuestro socket, pero no los estamos leyendo.

He puesto solo un thread para demostrarlo y fijaros que el buffer vuelve a ser 1024:

desu #4 20 Nov

Message boundery.

spoiler

use std::io::{self, Read, Write};

#[derive(Debug)]
pub struct Connection<T: Read + Write> {
    pub(crate) socket: T,
    buf: Vec<u8>,
    last_read_n: usize,
}

impl<T: Read + Write> Connection<T> {
    pub fn with_capacity(socket: T, capacity: usize) -> Self {
        Self {
            socket,
            buf: vec![0; capacity],
            last_read_n: 0,
        }
    }

pub fn read(&mut self) -> io::Result<usize> {
    if self.last_read_n >= self.buf.len() {
        return Ok(0);
    }

    let n = self.socket.read(&mut self.buf[self.last_read_n..])?;
    self.last_read_n += n;
    Ok(n)
}

pub fn write(&mut self) -> io::Result<()> {
    let n = self.socket.write(&self.buf[..self.last_read_n])?;
    if n > 0 {
        if n < self.last_read_n {
            self.buf.copy_within(n..self.last_read_n, 0);
        }
        self.last_read_n -= n;
    }

    Ok(())
}
}

#[cfg(test)]
mod tests {
    use super::*;

// Simple socket that reads/writes all data at once
struct StreamSocket {
    read_data: Vec<u8>,
    written_data: Vec<u8>,
    read_pos: usize,
}

impl StreamSocket {
    fn new(read_data: Vec<u8>) -> Self {
        Self {
            read_data,
            written_data: Vec::new(),
            read_pos: 0,
        }
    }
}

impl Read for StreamSocket {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if self.read_pos >= self.read_data.len() {
            return Ok(0);
        }
        let available = &self.read_data[self.read_pos..];
        let n = std::cmp::min(available.len(), buf.len());
        buf[..n].copy_from_slice(&available[..n]);
        self.read_pos += n;
        Ok(n)
    }
}

impl Write for StreamSocket {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.written_data.extend_from_slice(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

// Socket that simulates partial reads/writes
struct ChunkedSocket {
    read_data: Vec<u8>,
    written_data: Vec<u8>,
    read_pos: usize,
    chunk_size: usize,
}

impl ChunkedSocket {
    fn new(read_data: Vec<u8>, chunk_size: usize) -> Self {
        Self {
            read_data,
            written_data: Vec::new(),
            read_pos: 0,
            chunk_size,
        }
    }
}

impl Read for ChunkedSocket {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if self.read_pos >= self.read_data.len() {
            return Ok(0);
        }
        let available = &self.read_data[self.read_pos..];
        let n = std::cmp::min(std::cmp::min(available.len(), buf.len()), self.chunk_size);
        buf[..n].copy_from_slice(&available[..n]);
        self.read_pos += n;
        Ok(n)
    }
}

impl Write for ChunkedSocket {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        let n = std::cmp::min(buf.len(), self.chunk_size);
        self.written_data.extend_from_slice(&buf[..n]);
        Ok(n)
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

#[test]
fn test_connection_read() -> io::Result<()> {
    let test_data = b"hello world";
    let socket = StreamSocket::new(test_data.to_vec());
    let mut connection = Connection::with_capacity(socket, 16);

    let n = connection.read()?;
    assert_eq!(n, test_data.len());
    assert_eq!(&connection.buf[..n], test_data);
    Ok(())
}

#[test]
fn test_connection_write() -> io::Result<()> {
    let socket = StreamSocket::new(Vec::new());
    let mut connection = Connection::with_capacity(socket, 16);

    let test_data = b"hello world";
    connection.buf[..test_data.len()].copy_from_slice(test_data);
    connection.last_read_n = test_data.len();

    connection.write()?;
    assert_eq!(&connection.socket.written_data, test_data);
    Ok(())
}

#[test]
fn test_connection_partial_write() -> io::Result<()> {
    let socket = ChunkedSocket::new(Vec::new(), 5);
    let mut connection = Connection::with_capacity(socket, 16);

    let test_data = b"hello world";
    connection.buf[..test_data.len()].copy_from_slice(test_data);
    connection.last_read_n = test_data.len();

    connection.write()?;
    assert_eq!(&connection.buf[..connection.last_read_n], b" world");
    assert_eq!(&connection.socket.written_data, b"hello");
    Ok(())
}

#[test]
fn test_connection_chunked_read() -> io::Result<()> {
    let test_data = b"hello world";
    let socket = ChunkedSocket::new(test_data.to_vec(), 5);
    let mut connection = Connection::with_capacity(socket, 16);

    let n = connection.read()?;
    assert_eq!(n, 5);
    assert_eq!(&connection.buf[..n], b"hello");

    let n = connection.read()?;
    assert_eq!(n, 5);
    assert_eq!(&connection.buf[5..5 + n], b" worl");

    let n = connection.read()?;
    assert_eq!(n, 1);
    assert_eq!(&connection.buf[10..10 + n], b"d");
    Ok(())
}
}

Antes de explicar la solución que todo el mundo quiere ver, el famoso async, hago un paréntesis con esta pequeña ayuda que me he hecho para definir conexiones. He escrito unos tests para tratar lecturas y escrituras parciales.

Un socket es un file, como (casi) todo, tansolo necesitamos implementar Read y Write para poder leer y escribir. Asi que una conexión acepta un generico T que sea Read y Write. No tiene ninguna misterio.

Tenemos un buffer como ya hemos visto, en mi he decidido tener un tamaño fijo, y si me llega un mensaje mas grande, no leo mas. Una alternativa habitual es re-alocar y asignar un buffer mas grande. Y una optimización habitual seria tener un pool de buffers para no alocar a cada conexión.

desu #5 20 Nov

"async, wow, async async async"

3
Steps:
1. Create our socket
2. Bind it to an address
3. Put it in "server" mode (i.e. call listen on it)
4. Create an event queue
5. Add the server socket to the event queue
6. Wait for events in a loop
7. Application logic involving reading/writing/closing the socket
8. Remove socket from queue when done
9. Goto step 6

Creo que no hay nada que le guste mas a los ignorantes en software que hablar de async. Necesitamos una API async! Compañero, necesito que las llamadas sean asyncronas. Estamos migrando los servidores HTTP a async!

Voy a usar mio, que es una abstracción encima de epoll, kqueue, IOCP. La desgracia que es poll me la voy a saltar, bastante tenemos que sufrir con la porqueria de epoll, de lo peor que se ha escrito en al historia de la informática. Y de io_uring, de momento, mejor no hablemos.

La manera en que funciona una cola es muy simple. Simplificando para nuestro caso de uso, el kernel nos avisara mediante la cola cuando podamos leer o escribir sin bloquear. Imaginaros este caso:

mi socket esta ocupado
trato de leer del socket, como esta ocupado me bloqueo

Lo que vamos a hacer es:

registrar el socket, avisame cuando pueda escribir y leer
si el socket esta ocupado, mi codigo hace otras cosas
la cola me avisa que el socekt ya esta listo, lo uso y no me bloqueo

Wealth, status, pride,
are their own ruin.
To do good, work well, and lie low
is the way of the blessing.

Para usar la cola, necesitamos definir un tamaño máximo como upper bound. Voy a elegir 1024 clientes. Cada uno de los clientes tendra un Token asignado de 0..1023. Y necesitamos además un Token para nuestro server, voy a elegir el 1025.

const MAX_CLIENTS: usize = 1024;
const LISTENER: Token = Token(MAX_CLIENTS + 1);

Asi todos nuestros sockets iran del 0, 1, 2, 3, 4... 1023.
Cada vez que hagamos poll, la cola nos devolvera una lista de eventos que se podra escribir o leer.
Oye el socket 1, 3 y 6 puedes leer, el socket 2 puedes escribir.
Si no me llega ningúna notificaciones es que no puedo realizar operaciones sin bloquear.

Veamos como debemos inicializarlo con mio:

Tenemos:

Un servidor, listener, asignado a nuestro puerto 8080
Un Poll de mio, abstrae: epoll, kqueue, IOPC
Una lista de eventos, necesario para estas estructuras de datos, MAX_CLIENTS + 1, que es el +1? Lo acabamos de ver, el servidor.
Un Slab de Connection, (https://crates.io/crates/slab) un slab es una especie de array que nos sera muy util.

Que queremos? Seguiremos los pasos descritos arriba. El primero es que nuestro servidor acepte conexiónes.

Registro el listener como READABLE, cada vez que se pueda leer del listener nos llegara un evento. Esto NO significa que haya llegado una nueva conexión, no lo sabemos, lo que sabemos es que podremos leer de ese socket y la operación funcionara.

El codigo es muy simple, en un loop hacemos poll, este tiene un timeout que bloquea hasta que el kernel nos notifica que hay un evento. Como le hemos pasado toda la lista de eventos que estamos interesados, el kernel nos devolvera los que podemos leer y escribir, tanto servidor como clientes, como hemos visto en el ejemplo anterior.

Cada evento tiene un token que lo identifica, en este caso tenemos nuestro servidor LISTENER, que es el Token(1025) y token para los clientes que van de Token(0) a Token(1023) como hemos visto.

Para cada evento de LISTENER vamos a aceptar una conexión y añadirla a nuestros eventos y connections.
Para cada evento de un cliente, ya sea leer o escribir, haremos lo que toque.

Aqui es donde el codigo se vuelve loco.

El servidor es muy simple, para cada conexión, la añadimos en nuestro slab, y actualizamos nuestro registry como READABLE. Nos interesa leer de ese cliente.

uff que duro este codigo, y no esta ni al 50%.

hacemos exactamente lo que haciamos antes bloqueando, peor ahora tenemos que actualizar el poll y gestionar los errores.

cliente lee
si lee n == 0, es que no hay nada que hacer, limpia la conexion del registry y del slab
si lee n > 0, es que hay algo que escribir, no sabemos si esto es parcial o no, solo sabemos que hemos leido algo, asique marcamos como WRITEABLE
si se da el caso de que el event es WRITEABLE, escribimos

cuantos bugs tiene este codigo? infinitos. voy a resolver unos cuantos.

Funciona? Hombre, si escribimos a mano con un par de clientes funciona. Pero no esta bien.

En Rust usamos ? cuando queremos devolver un error, estos errores... no se pueden devolver nunca asi como asi... hay que gestionarlos. En el codigo del cliente tenemos 4 ?, en el del server 2?...

Can you keep your soul in its body,
hold fast to the one,
and so learn to be whole?
Can you center your energy,
be soft, tender,
and so learn to be a baby?

Empecemos con el server:

Primero que pasa si el listener.accept() falla? Lo voy a ignorar.
Segundo que pasa si el registry().register() falla? Lo voy a re-intentar hasta que funcione.

Ahora vamos a seguir con el cliente:

Primero que pasa si el connection.read() falla? Lo voy a ignorar.
Que pasa si el registry().deregister() falla? Lo voy a re-intentar hasta que funcione.
Segundo que pasa si el connection.write() falla? Lo voy a ignorar.

Esta esto bien?

spoiler

#[cfg(test)]
mod tests {

use super::*;
use std::{
    io::{Read, Write},
    net::{SocketAddr, TcpStream},
    thread::{self},
};

fn setup_server() -> Result<(Server, SocketAddr)> {
    let server_addr = "127.0.0.1:0".parse::<SocketAddr>()?;
    let server = Server::bind(server_addr)?;
    let local_addr = server.listener.local_addr()?;
    Ok((server, local_addr))
}

#[test]
fn test_server_single_message() -> Result<()> {
    let (mut server, local_addr) = setup_server()?;
    thread::spawn(move || {
        server.run().unwrap();
    });

    let mut stream = TcpStream::connect(local_addr)?;
    stream.set_nodelay(true)?;

    let test_msg = b"Hello, Server!";
    stream.write_all(test_msg)?;
    stream.flush()?;

    let mut buf = vec![0; test_msg.len()];
    stream.read_exact(&mut buf)?;
    assert_eq!(&buf, test_msg);

    Ok(())
}

#[test]
fn test_server_multiple_writes() -> Result<()> {
    let (mut server, local_addr) = setup_server()?;
    thread::spawn(move || server.run().unwrap());

    let mut stream = TcpStream::connect(local_addr)?;
    stream.set_nodelay(true)?;

    let test_msg1 = b"First";
    let test_msg2 = b"Second";
    stream.write_all(test_msg1)?;
    stream.write_all(test_msg2)?;
    stream.flush()?;

    let expected = [&test_msg1[..], &test_msg2[..]].concat();
    let mut buf = vec![0; expected.len()];
    stream.read_exact(&mut buf)?;
    assert_eq!(buf, expected);

    Ok(())
}

#[test]
fn test_server_large_message() -> Result<()> {
    let (mut server, local_addr) = setup_server()?;
    thread::spawn(move || server.run().unwrap());

    let mut stream = TcpStream::connect(local_addr)?;
    stream.set_nodelay(true)?;

    let test_msg = vec![b'X'; 1024];
    stream.write_all(&test_msg)?;
    stream.flush()?;

    let mut buf = vec![0; test_msg.len()];
    stream.read_exact(&mut buf)?;
    assert_eq!(buf, test_msg);

    Ok(())
}

#[test]
fn test_server_message_larger_than_buffer() -> Result<()> {
    let (mut server, local_addr) = setup_server()?;
    thread::spawn(move || server.run().unwrap());

    let mut stream = TcpStream::connect(local_addr)?;
    stream.set_nodelay(true)?;

    let buffer_size = 1024;
    let test_msg = vec![b'X'; buffer_size + 500];
    stream.write_all(&test_msg)?;
    stream.flush()?;

    let mut buf = vec![0; buffer_size];
    stream.read_exact(&mut buf)?;
    assert_eq!(&buf[..], &test_msg[..buffer_size]);

    Ok(())
}

#[test]
fn test_server_concurrent_clients() -> Result<()> {
    let (mut server, local_addr) = setup_server()?;
    thread::spawn(move || server.run().unwrap());

    let mut handles = vec![];
    let num_clients = 5;

    for i in 0..num_clients {
        let addr = local_addr;
        handles.push(thread::spawn(move || {
            let mut stream = TcpStream::connect(addr).unwrap();
            stream.set_nodelay(true).unwrap();

            let msg = format!("Client{}", i);
            stream.write_all(msg.as_bytes()).unwrap();
            stream.flush().unwrap();

            let mut buf = vec![0; msg.len()];
            stream.read_exact(&mut buf).unwrap();
            assert_eq!(buf, msg.as_bytes());
        }));
    }

    for handle in handles {
        handle.join().unwrap();
    }

    Ok(())
}

#[test]
fn test_server_client_disconnect_reconnect() -> Result<()> {
    let (mut server, local_addr) = setup_server()?;
    thread::spawn(move || server.run().unwrap());

    // First connection
    let mut stream = TcpStream::connect(local_addr)?;
    stream.write_all(b"hello")?;
    stream.flush()?;
    drop(stream); // Force disconnect

    // Small delay to ensure disconnect is processed
    thread::sleep(std::time::Duration::from_millis(10));

    // Reconnect
    let mut stream2 = TcpStream::connect(local_addr)?;
    stream2.write_all(b"world")?;
    stream2.flush()?;

    let mut buf = vec![0; 5];
    stream2.read_exact(&mut buf)?;
    assert_eq!(&buf, b"world");

    Ok(())
}

#[test]
fn test_server_zero_length_message() -> Result<()> {
    let (mut server, local_addr) = setup_server()?;
    thread::spawn(move || server.run().unwrap());

    let mut stream = TcpStream::connect(local_addr)?;
    stream.set_nodelay(true)?;

    stream.write_all(b"")?;
    stream.flush()?;

    // Write a real message after empty one
    let msg = b"test";
    stream.write_all(msg)?;
    stream.flush()?;

    let mut buf = vec![0; msg.len()];
    stream.read_exact(&mut buf)?;
    assert_eq!(&buf, msg);

    Ok(())
}

#[test]
fn test_server_rapid_connect_disconnect() -> Result<()> {
    let (mut server, local_addr) = setup_server()?;
    thread::spawn(move || server.run().unwrap());

    for _ in 0..10 {
        let stream = TcpStream::connect(local_addr)?;
        drop(stream);
    }

    // Final connection to verify server still works
    let mut stream = TcpStream::connect(local_addr)?;
    stream.write_all(b"test")?;
    stream.flush()?;

    let mut buf = vec![0; 4];
    stream.read_exact(&mut buf)?;
    assert_eq!(&buf, b"test");

    Ok(())
}
}

Lo suyo es escribir unos tests y bueno, todo funciona con estos, ya hemos testeado a parte que nuestra Connection tenga partial read y partial writes y funcione. El problema aqui esta en los edge cases asyncronos del kernel.

solucion

use std::net::SocketAddr;

use anyhow::Result;
use log::{debug, error};

mod connection;
use connection::Connection;
use mio::{
    net::{TcpListener, TcpStream},
    Events, Poll, Token,
};
use slab::Slab;

const MAX_CLIENTS: usize = 1024;
const LISTENER: Token = Token(MAX_CLIENTS + 1);

struct Server {
    listener: TcpListener,
    poll: Poll,
    events: Events,
    connections: Slab<Connection<TcpStream>>,
}

impl Server {
    fn bind(addr: SocketAddr) -> Result<Self> {
        let mut listener = TcpListener::bind(addr)?;
        debug!("server listening on {:?}", listener.local_addr()?);

    let poll = Poll::new()?;
    let events = Events::with_capacity(MAX_CLIENTS + 1);
    let connections = Slab::with_capacity(MAX_CLIENTS);

    poll.registry()
        .register(&mut listener, LISTENER, mio::Interest::READABLE)?;

    Ok(Self {
        listener,
        poll,
        events,
        connections,
    })
}

fn run(&mut self) -> Result<()> {
    loop {
        self.poll.poll(&mut self.events, None)?;
        for event in self.events.iter() {
            match event.token() {
                LISTENER => {
                    if let Ok((socket, addr)) = self.listener.accept() {
                        debug!("server: Connection from {:?}", addr);

                        let connection = Connection::with_capacity(socket, 1024);
                        let key = self.connections.insert(connection);
                        while let Err(e) = self.poll.registry().register(
                            &mut self.connections[key].socket,
                            Token(key),
                            mio::Interest::READABLE,
                        ) {
                            error!("server: Failed to register client: {:?}", e);
                        }
                    } else {
                        debug!("server: Failed to accept connection");
                    }
                }
                token => {
                    if let Some(connection) = self.connections.get_mut(token.0) {
                        if event.is_readable() {
                            if let Ok(n) = connection.read() {
                                if n == 0 {
                                    debug!("Client disconnected");
                                    while let Err(e) =
                                        self.poll.registry().deregister(&mut connection.socket)
                                    {
                                        error!("server: Failed to deregister client: {:?}", e);
                                    }
                                    self.connections.remove(token.0);
                                    continue;
                                } else {
                                    while let Err(e) = self.poll.registry().reregister(
                                        &mut connection.socket,
                                        token,
                                        mio::Interest::READABLE.add(mio::Interest::WRITABLE),
                                    ) {
                                        error!("server: Failed to reregister client: {:?}", e);
                                    }
                                }
                            }
                        } else if event.is_writable() {
                            if let Err(e) = connection.write() {
                                error!("server: Failed to write to client: {:?}", e);
                            }
                        }
                    }
                }
            }
        }
    }
}
}

fn main() -> Result<()> {
    env_logger::init();

let mut server = Server::bind("127.0.0.1:8080".parse()?)?;
server.run()?;

Ok(())
}

desu #6 20 Nov

Preguntas?

Hasta aqui hoy, voy a ir arreglando el codigo y el texto para que se entienda mejor, cuando tenga tiempo.

Me queda el multi-thread + async.

desu #7 20 Nov

Final. Multi-thread + async. Parte 1.

Esto no seria la ugly org si no os doy una master class. No hagais multi-thread + async. Hacer multi proceso + async. Seguramente esto es suficiente para el 99% de vuestros casos de uso.

Que signifca?

Ejecuta multiples servidores :8080, :8081, :8082, :8083
Cada servidor correra en un core, y sera async
Mete un nginx delante que balance el trafico
???
Profit!

Hold fast to the great thought
and all the world will come to you,
harmless, peaceable, serene.

Tendrás algo mejor que Spring Boot + Netty en menos de 500 lineas de código, si no te lo crees, hazlo. Si usas Go lo puedes tener en 50 lineas de código.

1

desu #8 20 Nov

Final. Multi-thread + async. Parte 2.

4
Steps:
1. Create our socket
2. Bind it to an address
3. Put it in "server" mode (i.e. call listen on it)
4. Create an event queue
5. Add the server socket to the event queue
6. Create a thread pool
7. Wait for events in a loop
8. Each thread does application logic involving reading/writing/closing the socket
8. Remove socket from queue when done
9. Goto step 6

desu #9 20 Nov

TODO: implement a full runtime

For reference see: tokio, std-async, golang, C#/Java futures, python asyncio, C libuv, Scala finagle...

Hipnos #10 20 Nov Dungeon Master

Me resulta curioso qué tiene que ver la alquimia interna china con programar.

1 respuesta

desu #11 21 Nov

#10

For being and nonbeing
arise together;
hard and easy
complete each other;
high and low
depend on each other;
note and voice
make the music together;
before and after
follow each other.

Fundamentos.
Escribir y leer de IO. File descriptors.
Y nada mas.

Este hilo no pretende ser un tutorial.
La ultima vez +20 personas votaron que querian que les enseñase un server tcp/http en un proyecto, y pense que hacerlo a mano con las syscall fundamentales es un ejercicio que todo el mundo debe hacer para comprender después lo que pasa.
Aquí esta unos ejemplitos que todo el mundo puede hacer en sus lenguajes preferidos.

2

the ugly org - backend - fundamentos del Dao De Jing (道德經)

TODO: implement a full runtime

For reference see: tokio, std-async, golang, C#/Java futures, python asyncio, C libuv, Scala finagle...

Usuarios habituales

Tags