线程之间共享引用的生命周期问题

Question

我有一个启动工作线程的线程，所有线程都应该永远存在。每个工作线程维护它自己的 Sockets.

列表

有些操作要求我遍历所有当前活动的套接字，但我在尝试创建包含指向另一个列表拥有的套接字的指针的套接字主列表时遇到了生命周期问题。

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);
}

我收到的错误...

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

我什至不知道我正在尝试的是否可以作为安全代码。

我希望 mastersocketlist 包含一个指向套接字的指针，套接字的生命周期由创建它的线程定义。我相信这就是所有这些错误的意思，但我不知道如何提供正确的生命周期注释来修复它。

Answer 1

Rust 的一大优点是跨函数的类型检查完全由函数签名完成。这意味着您可以用 unimplemented!() 替换大部分函数体并保留类型检查错误。

重复这个过程几次，你最终没有调用很多函数 - 删除那些。内联模块和减少结构/枚举也有帮助。

在某些时候你的错误会消失 - 问题的第一条线索！坚持下去，你会得到一个微小的复制品：

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() {}

检查错误，依然匹配：

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

让我们检查文档以获取 thread::spawn 的签名：

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

这里的关键点是 F: Send + 'static - 您给 spawn 的闭包必须只包含持续整个程序生命周期的引用 。这是因为 spawn 可以创建成为 分离的线程 。一旦分离，线程可以永远存在，所以所有引用必须至少存在那么久，否则你会得到 悬空引用 ，这是一件坏事！ Rust 再次拯救世界！

如果您想保证线程将在已知点终止，您可以使用 作用域线程，例如 scoped-threadpool or crossbeam.[=23 提供的线程=]

如果您的代码中没有包含生命周期的变量，请使用某种类型的共享所有权，例如 Arc paired with something that will ensure only one thread can mutate the variable, like Mutex would have been sufficient. This allows each thread to own the shared value, finally dropping it whenever the last thread exits. See 了解详细信息。

线程之间共享引用的生命周期问题

Lifetime troubles sharing references between threads

rust

lifetime