"`self` has an anonymous lifetime `'_` but it needs to satisfy a `'static` lifetime requirement" when access self. fields of struct from thread

user_5397119 0
Viewed 393

I implemented struct, which use some threads inside it. When I try to use self. inside thread, I got an error:

error[E0759]: `self` has an anonymous lifetime `'_` but it needs to satisfy a `'static` lifetime requirement
   --> src/main.rs:37:23
    |
36  |       fn spawn_write_thread(&mut self) -> thread::JoinHandle<()> {
    |                             --------- this data with an anonymous lifetime `'_`...
37  |           thread::spawn(move || {
    |  _______________________^
38  | |             let val = self.receiver.recv().unwrap();
39  | |             self.push(val);
40  | |             thread::sleep(Duration::from_millis(10));
41  | |         })
    | |_________^ ...is used here...
    |
note: ...and is required to live as long as `'static` here
   --> src/main.rs:37:9
    |
37  |         thread::spawn(move || {
    |         ^^^^^^^^^^^^^
note: `'static` lifetime requirement introduced by this bound

This is my code:

use std::thread;
use std::sync::mpsc;
use std::time::Duration;
use rand::Rng;

struct Test {
    data: Vec<u8>,
    sender: mpsc::Sender<u8>, 
    receiver: mpsc::Receiver<u8>,
}

impl Test {
    pub fn new(
        sender: mpsc::Sender<u8>, 
        receiver: mpsc::Receiver<u8>
    ) -> Self {
        Self {
            data: vec![],
            sender,
            receiver,
        }
    }
    pub fn push(&mut self, item: u8) {
        self.data.push(item);
    }

    fn spawn_read_thread(&mut self) -> thread::JoinHandle<()> {
        thread::spawn(move || {
            let val = rand::thread_rng().gen();
            self.sender.send(val).unwrap();
            thread::sleep(Duration::from_millis(10));
        })
    }

    fn spawn_write_thread(&mut self) -> thread::JoinHandle<()> {
        thread::spawn(move || {
            let val = self.receiver.recv().unwrap();
            self.push(val);
            thread::sleep(Duration::from_millis(10));
        })
    }

    pub fn output(&mut self) {
        self.spawn_read_thread();
        self.spawn_write_thread();
    
        for _ in 0..20 {
            println!("{:?}", self.data);
            thread::sleep(Duration::from_millis(100));
        }
    }
}

fn main () {
    let (sender, receiver) = mpsc::channel();
    let test = Test::new(sender.clone(), receiver);
    test.output();
}

This is sandbox implementation.

I tried to use Arc + Mutex on specific self. fields/methods, but this didn't help.

Could you explain, how to use self references inside thread to access struct's fields/methods ?

rust-thread rust
1 Answers

In Rust, every object has to be owned by exactly one owner. Then, you can borrow references to that object, but it must be known to the one borrowing how long the object lives. Dangling references (references that refer to an object that doesn't exist any more) are undefined behaviour, and by definition, Rust does not have any undefined behaviour as long as you don't use unsafe.

Sadly, in your case, the owning context of the test object (which I'll refer to as self from now on) is on a different thread than the context holding the reference (self.sender.send() actually creates a reference to self.sender and by extension to self). But the compiler does not know how long the thread lives, or if the test object outlives the reference. That's why it throws an error.

If you look at the error message:

`self` has an anonymous lifetime `'_` but it needs to satisfy a `'static` lifetime requirement

It says that in order to use a reference in a thread, the reference has to be 'static, meaning, the object reference needs to live for the entire duration of the program. ('static means actually something slightly different, but in the context of this example, that's what it means).

I'd like to point you to a solution in the way of "change those two lines and then it's fixed"; sadly, I don't think such a solution exists.

Yes, it's suggested in the comments to use Arc<Mutex>, which does eliminate the lifetime problem by introducing reference counters, so that the thread itself can keep the objects alive and not just borrow them. But I feel like those are just hotfixes for a more underlying architectural proble.

The actual, underlying problem, is that your code shows no clear concept of ownership trees. That is not a problem in other languages, and that's why it's a little hard to cope with it at the beginning.

Let me demonstrate what I mean. Remember, in Rust it must always be clear where a variable is owned and how long it lives:

  • main owns the Test object
  • Test object owns sender and receiver
  • in a way, conceptually, Test owns the threads, because it spawns them
  • the threads interact with the sender and receiver mutably, so the threads in a way need to own the sender and receiver as well
  • the threads use the Test object mutably, so in a way, the threads also need to own the Test object

It's just a big mess of everyone-owns-everyone, which is probably how you would implement it in C++ and smart pointers, or even raw pointers. In Rust, the chain of ownership has to be much more specific.

So let's ask ourselves, who actually needs to own what?

That's the part where I get a little speculative, because your description is not sufficient to understand the underlying usecase. I will base my assumptions on the code you posted.

  • Does main need to own the Test object?
    • Yes, someone has to. Otherwise nobody would own the Test object.
  • Does Test need to own sender and receiver?
    • That's the first point where I'd disagree with your code. Apart of when it spawns the threads, the Test object never uses sender and receiver again. So I'd argue that the real owners of those should be the threads instead.
  • Does Test need to own the threads?
    • I'd argue "yes", and I'd even say it doesn't own them enough yet. You spawn the threads and then drop the JoinHandles immediately. I'd keep the JoinHandles and make sure to dispose of the threads properly when the Test object gets destroyed.
  • Do the threads need to own the sender and receiver?
    • Yes, definitely. They are the only context that actually interacts with those.
  • Do the threads need to own the Test object?
    • that one is the most tricky one, because this one creates a circular ownership chain. The Test object owns the threads, and the threads own the Test object because they need to call push(). It's a dataflow problem, which is a quite common problem that causes circular ownership problems. There are two ways to solve that in Rust, usually:
      • channels: Break the need of carrying the self object over by changing the callback pattern to a channel pattern. callback patterns in general are discouraged in Rust, due to the ownership problems you just encountered. This is one of the most common programming pattern conversions when porting C++ code to Rust, as you can pass the sender part of a channel to a thread without having to pass its receiver into it.
      • Reference counters, e.g. Arc<Mutex<>>. In this case, as you want to actually change the data from the thread and not just pass events into a pipe, I think this is the pattern I'd go for.

With all those changes applied, this is kind of the direction I would go:

use rand::Rng;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{mpsc, Arc, Mutex};
use std::thread;
use std::time::Duration;

struct Test {
    data: Arc<Mutex<Vec<u8>>>,
    read_thread: Option<thread::JoinHandle<()>>,
    write_thread: Option<thread::JoinHandle<()>>,
    sender_should_stop: Arc<AtomicBool>,
}

impl Test {
    pub fn new(sender: mpsc::Sender<u8>, receiver: mpsc::Receiver<u8>) -> Self {
        let data = Arc::new(Mutex::new(vec![]));
        let sender_should_stop = Arc::new(AtomicBool::new(false));

        let read_thread = Some(Test::spawn_read_thread(sender, sender_should_stop.clone()));
        let write_thread = Some(Test::spawn_write_thread(receiver, data.clone()));

        Self {
            data,
            read_thread,
            write_thread,
            sender_should_stop,
        }
    }

    fn spawn_read_thread(
        sender: mpsc::Sender<u8>,
        should_stop: Arc<AtomicBool>,
    ) -> thread::JoinHandle<()> {
        println!("Spawning sender ...");
        thread::spawn(move || loop {
            let val = rand::thread_rng().gen();
            if let Err(_) = sender.send(val) {
                println!("Channel got closed. Stopping sender ...");
                break;
            }
            thread::sleep(Duration::from_millis(1000));

            // Stop sender if signalled
            if should_stop.load(Ordering::Acquire) {
                println!("Sender received the signal to stop. Stopping sender ...");
                break;
            };
        })
    }

    fn spawn_write_thread(
        receiver: mpsc::Receiver<u8>,
        data: Arc<Mutex<Vec<u8>>>,
    ) -> thread::JoinHandle<()> {
        println!("Spawning receiver ...");
        thread::spawn(move || loop {
            let val = match receiver.recv() {
                Ok(val) => val,
                Err(_) => {
                    println!("Channel got closed. Stopping receiver ...");
                    break;
                }
            };

            data.lock().unwrap().push(val);
        })
    }

    pub fn output(&self) {
        for _ in 0..20 {
            println!("{:?}", self.data.lock().unwrap());
            thread::sleep(Duration::from_millis(333));
        }
    }
}

impl Drop for Test {
    fn drop(&mut self) {
        // Signal the sender to stop.
        println!("Signalling sender to stop ...");
        self.sender_should_stop.store(true, Ordering::Release);

        // Wait for the sender to finish
        std::mem::take(&mut self.read_thread)
            .unwrap()
            .join()
            .unwrap();
        println!("Sender stopped.");

        // Now that the sender is finished and dropped its part of the channel,
        // the receiver will finish as well
        std::mem::take(&mut self.write_thread)
            .unwrap()
            .join()
            .unwrap();
        println!("Receiver stopped.");
    }
}

fn main() {
    let (sender, receiver) = mpsc::channel();
    let test = Test::new(sender, receiver);
    test.output();
}

Spawning sender ...
Spawning receiver ...
[]
[82]
[82]
[82]
[82, 79]
[82, 79]
[82, 79]
[82, 79, 8]
[82, 79, 8]
[82, 79, 8]
[82, 79, 8, 114]
[82, 79, 8, 114]
[82, 79, 8, 114]
[82, 79, 8, 114, 194]
[82, 79, 8, 114, 194]
[82, 79, 8, 114, 194]
[82, 79, 8, 114, 194, 210]
[82, 79, 8, 114, 194, 210]
[82, 79, 8, 114, 194, 210]
[82, 79, 8, 114, 194, 210, 231]
Signalling sender to stop ...
Sender received the signal to stop. Stopping sender ...
Channel got closed. Stopping receiver ...
Sender stopped.
Receiver stopped.

Playground

Now you can clearly see the ownership tree:

  • data: owned by Arc

  • sender_should_stop: owned by Arc

  • main

    • test
      • data: ->Arc
      • sender_should_stop: ->Arc
      • read_thread
        • sender
        • should_stop: ->Arc
      • write_thread
        • receiver
        • data: ->Arc

Annotation: I know that nobody actually owns a thread, it's just a handle and the thread can live without it. I just conceptually think of a thread being owned by the JoinHandle because usually, the JoinHandle is used for error propagation.

user_29028330
Related