rust - Problemas de por vida para compartir referencias entre hilos

lifetime

Question

Tengo un hilo que lanza hilos de trabajo, se espera que todos vivan para siempre. Cada subproceso de trabajo mantiene su propia lista de Socket s.

Algunas operaciones requieren que atraviese todos los sockets actualmente vivos, pero estoy teniendo problemas con vidas que intentan crear una lista maestra de sockets que contengan un puntero a un socket propiedad de otra lista.

use std::{str, thread};
use std::thread::JoinHandle;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::ops::DerefMut;
use std::sync::mpsc::{channel, Sender, Receiver, TryRecvError};
use self::socketlist::SocketList;
use self::mastersocketlist::MasterSocketList;

pub struct Socket {
    user: String,
    stream: TcpStream,
}

mod socketlist {
    use self::SocketList::{Node, End};
    use super::Socket;

    pub enum SocketList {
        Node(Socket, Box<SocketList>),
        End,
    }

    impl SocketList {
        pub fn new() -> SocketList {
            End
        }

        pub fn add(self, socket: Socket) -> SocketList {
            Node(socket, Box::new(self))
        }

        pub fn newest<''a>(&''a mut self) -> Result<&''a Socket, String> {
            match *self {
                Node(ref mut socket, ref mut next) => Ok(socket),
                End => Err("No socket available".to_string()),
            }
        }
    }
}

mod mastersocketlist {
    use self::MasterSocketList::{Node, End};
    use super::Socket;

    pub enum MasterSocketList<''a> {
        Node(Box<&''a Socket>, Box<MasterSocketList<''a>>),
        End,
    }

    impl<''a> MasterSocketList<''a> {
        pub fn new() -> MasterSocketList<''a> {
            End
        }

        pub fn add(self, socket: &''a Socket) -> MasterSocketList<''a> {
            MasterSocketList::Node(Box::new(&socket), Box::new(self))
        }
    }
}

pub struct SlotManager {
    prox: JoinHandle<()>,
    prox_tx: Sender<TcpStream>,
}

impl SlotManager {
    pub fn new() -> SlotManager {
        let (tx, rx): (Sender<TcpStream>, Receiver<TcpStream>) = channel();

        let tx_clone = tx.clone();
        let prox = thread::spawn(move || SlotManager::event_loop(tx, rx));

        SlotManager {
            prox: prox,
            prox_tx: tx_clone,
        }
    }

    pub fn sender(&self) -> Sender<TcpStream> {
        self.prox_tx.clone()
    }

    fn event_loop(tx: Sender<TcpStream>, rx: Receiver<TcpStream>) {
        let socket_list = Arc::new(Mutex::new(MasterSocketList::new()));
        let mut slot = Slot::new(socket_list.clone());
        loop {
            match rx.try_recv() {
                Ok(stream) => slot.new_connection(stream),
                Err(e) => {}
            }
        }
    }
}

pub struct Slot {
    prox: JoinHandle<()>,
    prox_tx: Sender<TcpStream>,
}

impl Slot {
    pub fn new(master_socket_list: Arc<Mutex<MasterSocketList>>) -> Slot {
        let (tx, rx): (Sender<TcpStream>, Receiver<TcpStream>) = channel();

        let tx_clone = tx.clone();
        let prox = thread::spawn(move || Slot::event_loop(tx, rx, master_socket_list));

        Slot {
            prox: prox,
            prox_tx: tx_clone,
        }
    }

    pub fn new_connection(&self, stream: TcpStream) {
        self.prox_tx.send(stream);
    }

    fn event_loop(tx: Sender<TcpStream>,
                  rx: Receiver<TcpStream>,
                  master_socket_list: Arc<Mutex<MasterSocketList>>) {

        let mut sockets = SocketList::new();
        loop {
            // Check for new connections
            match rx.try_recv() {
                Ok(stream) => {
                    let mut socket = Socket {
                        user: "default".to_string(),
                        stream: stream,
                    };
                    sockets = sockets.add(socket);

                    let mut msl_guard = match master_socket_list.lock() {
                        Ok(guard) => guard,
                        Err(poisoned) => poisoned.into_inner(),
                    };
                    let mut msl_handle = msl_guard.deref_mut();
                    *msl_handle = msl_handle.add(sockets.newest().unwrap());
                }
                Err(e) => {}
            }
        }
    }
}

fn main() {
    let mut slot_manager = SlotManager::new();
    let listener = TcpListener::bind("127.0.0.1:1234").unwrap();
    for stream in listener.incoming() {
        match stream {
            Ok(stream) => {
                let sender = slot_manager.sender();
                thread::spawn(move || {
                    sender.send(stream);
                    //process_new_connection(stream, sender)
                });
            }
            Err(e) => println!("Connection error: {}", e),
        }
    }
    drop(listener);
}

Los errores que recibo ...

error[E0477]: the type `[closure@src/main.rs:107:34: 107:86 tx:std::sync::mpsc::Sender<std::net::TcpStream>, rx:std::sync::mpsc::Receiver<std::net::TcpStream>, master_socket_list:std::sync::Arc<std::sync::Mutex<mastersocketlist::MasterSocketList<''_>>>]` does not fulfill the required lifetime
   --> src/main.rs:107:20
    |
107 |         let prox = thread::spawn(move || Slot::event_loop(tx, rx, master_socket_list));
    |                    ^^^^^^^^^^^^^
    |
    = note: type must outlive the static lifetime

Ni siquiera sé si lo que estoy tratando de hacer es posible como código seguro.

Quiero que mastersocketlist contenga un puntero a un socket donde la vida útil del socket se define por el hilo que lo creó. Creo que eso es lo que significan todos esos errores, pero no tengo idea de cómo proporcionar las anotaciones de por vida adecuadas para solucionarlo.

Answer 1

Una gran cosa acerca de Rust es que la verificación de tipo entre funciones se realiza únicamente mediante la firma de la función. ¡Eso significa que puede reemplazar la mayoría de los cuerpos de funciones con unimplemented!() Y preservar los errores de verificación de tipo.

Repita ese proceso varias veces y terminará sin invocar muchas funciones: elimínelas. La inclusión de módulos y la reducción de estructuras / enumeraciones también pueden ayudar.

En algún momento su error desaparecerá, ¡la primera pista sobre el problema! Sigue así y obtendrás una pequeña reproducción:

use std::sync::{Arc, Mutex};
use std::thread;

pub enum MasterSocketList<''a> {
    One(&''a u8),
}

pub struct Slot;

impl Slot {
    pub fn new<''a>(master_socket_list: Arc<Mutex<MasterSocketList<''a>>>) -> Slot {
        thread::spawn(move || {
            master_socket_list;
        });
        unimplemented!();
    }
}

fn main() {}

Al revisar el error, aún coincide:

error[E0477]: the type `[closure@src/main.rs:12:23: 14:10 master_socket_list:std::sync::Arc<std::sync::Mutex<MasterSocketList<''a>>>]` does not fulfill the required lifetime
  --> src/main.rs:12:9
   |
12 |         thread::spawn(move || {
   |         ^^^^^^^^^^^^^
   |
   = note: type must satisfy the static lifetime

Vamos a ver los documentos para la firma de thread::spawn :

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
where
    F: FnOnce() -> T,
    F: Send + ''static,
    T: Send + ''static, 

El punto clave aquí es F: Send + ''static : el cierre que le da a spawn solo debe contener referencias que duren toda la vida del programa . Esto se debe a que la spawn puede crear hilos que se desprenden . Una vez separado, el hilo podría vivir para siempre, por lo que todas las referencias deben vivir al menos ese tiempo, de lo contrario, obtendría referencias colgantes , ¡algo malo! Rust salva el día, una vez más!

Si desea garantizar que los subprocesos terminarán en un punto conocido, puede usar subprocesos con ámbito , como los proporcionados por scoped-threadpool o crossbeam .

Si su código no tenía una variable con una vida útil dentro de él, usar algún tipo de propiedad compartida como Arc emparejado con algo que garantice que solo un hilo pueda mutar la variable, como Mutex hubiera sido suficiente. Esto permite que cada subproceso posea el valor compartido y finalmente lo suelte cada vez que salga el último subproceso. Consulte ¿Cómo comparto un objeto mutable entre subprocesos? para detalles.