Passing self into a thread

Question

I have the following code:

use std::error::Error;
use std::collections::HashMap;
use std::fmt::{Display, Formatter, Error as FmtError};
use std::sync::{Arc, Mutex};
use std::thread;

use amiquip::{ConsumerMessage};
use serde::Deserialize;

use crate::rabbit_mq;
use crate::rabbit_mq::RmqChannel;
use super::agv::*;

pub type AGVControllerError = Box;
pub type AGVControllerResult = Result;


pub struct AGVController<'a> {
    agv_map: HashMap<&'a str, AGV>,
    //rmq: &'a mut rabbit_mq::RMQ
    rmq: Arc>
}

impl<'a> AGVController<'a> {
    // Create a new AGV Controller
    //pub fn new(rmq: &'a mut rabbit_mq::RMQ) -> Self {
    pub fn new(rmq: Arc>) -> Self {
        Self {
            agv_map: HashMap::new(),
            rmq
        }
    }

    // Listen for messages related to the agv controller.
    pub fn listen(&'static mut self, routing_key: &'static str) -> AGVControllerResult<()> {
        let mut rmq_channel = self.rmq.lock().unwrap().create_channel()?;

        thread::spawn(move || {
            //TODO: need to handle any error returned here
            //listen_on_consumer(routing_key, rmq_channel).unwrap();
            let consumer = rmq_channel.create_consumer(routing_key).unwrap();
            let test = self.agv_map.get("rr");
            for (i, message) in consumer.receiver().iter().enumerate() {
                match message {
                    ConsumerMessage::Delivery(delivery) => {
                        let body = String::from_utf8_lossy(&delivery.body);
                        println!("({:>3}) Received [{}]", i, body);
                        consumer.ack(delivery).unwrap();
                    }
                    other => {
                        println!("Consumer ended: {:?}", other);
                        break;
                    }
                }
            }
        });

        return Ok(());
    }
}

So in the listen function I create a consumer from rmq and i just read off messages as they come. That all works fine. But when a message is read off I need access to the struct information, data and functions. Making the self parameter use 'static seems to fix it. Is this okay? static means it stays alive as long as the program exists, right? Which seems correct since i want it to be alive as long as the thread is going. I just want to make sure Im doing this correctly

UPDATE:

So I tried to clone self and use that in the thread but I am getting:

error[E0495]: cannot infer an appropriate lifetime due to conflicting requirements
   --> src/agv/agv_controller.rs:47:35
    |
47  |         let mut self_clone = self.clone();
    |                                   ^^^^^
    |
note: first, the lifetime cannot outlive the lifetime `'a` as defined here...
   --> src/agv/agv_controller.rs:35:6
    |
35  | impl<'a> AGVController<'a> {
    |      ^^
note: ...so that the types are compatible
   --> src/agv/agv_controller.rs:47:35
    |
47  |         let mut self_clone = self.clone();
    |                                   ^^^^^
    = note: expected `&agv_controller::AGVController<'_>`
               found `&agv_controller::AGVController<'a>`
    = note: but, the lifetime must be valid for the static lifetime...
note: ...so that the type `[closure@src/agv/agv_controller.rs:48:23: 66:10]` will meet its required lifetime bounds...
   --> src/agv/agv_controller.rs:48:9
    |
48  |         thread::spawn(move || {
    |         ^^^^^^^^^^^^^
note: ...that is required by this bound
   --> /Users/conordowney/.rustup/toolchains/nightly-aarch64-apple-darwin/lib/rustlib/src/rust/library/std/src/thread/mod.rs:646:15
    |
646 |     F: Send + 'static,
    |               ^^^^^^^

user_13843935 · Answer

I had to butcher your program to get it to compile on my machine (for future reference, a minimal reproducable example that can be dropped in a playground is super helpful), but I think I found the root cause.

It didn't like that you are trying to pass something inside a Mutex through to another thread, so you should clone the Arc (its main purpose is to be cloned), then lock it in the thread.

I can't tell what an AGV is, but it looks like you are only reading from the map, so hopefully the unwrap/deref allows it to be moved thread-safely. If you need the whole map, you should be able to substitute a call to .clone().

use std::error::Error;
use std::collections::HashMap;
use std::fmt::{Display, Formatter, Error as FmtError};
use std::sync::{Arc, Mutex};
use std::thread;

use amiquip::{ConsumerMessage};
use serde::Deserialize;

use crate::rabbit_mq{self, RmqChannel};
use super::agv::*;

pub type AGVControllerError = Box;
pub type AGVControllerResult = Result;


pub struct AGVController<'a> {
    agv_map: HashMap<&'a str, AGV>,
    rmq: Arc>
}

impl<'a> AGVController<'a> {
    // Create a new AGV Controller
    pub fn new(rmq: Arc>) -> Self {
        Self {
            agv_map: HashMap::new(),
            rmq
        }
    }

    // Listen for messages related to the agv controller.
    pub fn listen(&self, routing_key: &'static str) -> AGVControllerResult<()> {
        //let mut rmq_channel = self.rmq.lock().unwrap().create_channel()?;
        let rmq_channel_mtx = self.rmq.clone();
        let rr = *self.agv_map.get("rr").unwrap()

        thread::spawn(move || {
            let rmq_channel = rmq_channel_mtx.lock().unwrap();
            let consumer = rmq_channel.create_consumer(routing_key).unwrap();

            let test = rr;

            for (i, message) in consumer.receiver().iter().enumerate() {
                match message {
                    ConsumerMessage::Delivery(delivery) => {
                        let body = String::from_utf8_lossy(&delivery.body);
                        println!("({:>3}) Received [{}]", i, body);
                        consumer.ack(delivery).unwrap();
                    }
                    other => {
                        println!("Consumer ended: {:?}", other);
                        break;
                    }
                }
            }
        });

        return Ok(());
    }
}

Side note, unless you really need a &str, I find its usually easier to deal with owned Strings in structures like HashMap, plus you won't need those lifetime annotations everywhere.

Passing self into a thread

1 Answers