Share Arc between closures

user_219449 0
Viewed 2388

I'm trying to write a simple tcp server which would read and broadcast messages.
I'm using Tokio, but I think it's more of a general Rust question.

I have an Arc with a shared state:
let state = Arc::new(Mutex::new(Shared::new(server_tx)));

Later I want to spawn 2 threads which would use a reference to that state: 

let server = listener.incoming().for_each(move |socket| {
    // error[E0382]: capture of moved value: `state`
    process(socket, state.clone());
    Ok(())
}).map_err(|err| {
    println!("accept error = {:?}", err);
});

let receive_sensor_messages = sensors_rx.for_each(move |line| {
    println!("Received sensor message, broadcasting: {:?}", line);

    // error[E0597]: borrowed value does not live long enough
    // error[E0507]: cannot move out of borrowed content 
    for (_, tx) in state.clone().lock().unwrap().clients {
        tx.unbounded_send(line.clone()).unwrap();
    }
    Ok(())
}).map_err(|err| {
    println!("line reading error = {:?}", err);
});

(playground)

As far as I understand what it's trying to tell me is that state is borrowed in the first closure listener.incoming().for_each(move |socket| { so when I try to do it again in sensors_rx.for_each(move |line| { it's saying it's not possible.

My question is how do I solve it? Isn't Arc supposed to solve the issue of sharing a variable between threads? I tried different combinations of clone (doing clone outside of the closure and then doing clone inside again), but none worked.

Cheers!

rust
2 Answers

Essentially, your problem can be boiled down to the following MCVE:

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

struct Bar;

fn foo(_ : &Bar){
    println!("foo called");
}

fn main(){
    let example = Arc::new(Mutex::new(Bar));
    std::thread::spawn(move ||{
        let _ = example.clone();
    });
    // --- (1) ---

    std::thread::spawn(move ||{
        foo(&example.clone().lock().unwrap());
    });
}

Now, the first problem here is that example is moved. That is, as soon as we crossed (1), the original example is considered to be moved from. Instead, we need to first clone and then move:

    let example = Arc::new(Mutex::new(Bar));
    let local_state = example.clone();
    std::thread::spawn(move ||{
        let _ = local_state; // now fine!
    });

The other error stems from the short lived Arc. Essentially, it only lives long enough for you to us lock on the underlying Mutex. While we know that there is at least one other Arc pointing to the memory, the compiler cannot prove that. However, if we get rid of the clone() it's fine:

    let local_state = example.clone();        
    std::thread::spawn(move ||{
        foo(&local_state.lock().unwrap());
    });

However, you also loop over your container by consuming its contents (the clients). Instead, use & there, e.g. &local_state().unwrap().clients).

You can find the complete fixed code below or on the playground:

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

struct Bar;

fn foo(_ : &Bar){
    println!("foo called");
}

fn main(){
    let example = Arc::new(Mutex::new(Bar));
    let local_state = example.clone();
    std::thread::spawn(move ||{
        let _ = local_state;
    });
    let local_state = example.clone();
    std::thread::spawn(move ||{
        foo(&local_state.lock().unwrap());
    }).join();
}
user_11396970

For every closure you have to provide its own Arc, so you have to clone your Arc beforehand.

let state = Arc::new(Mutex::new(Shared::new(server_tx)));
let state1 = Arc::clone(&state);
let state2 = Arc::clone(&state);

let server = listener.incoming().for_each(move |socket| {
    process(socket, state1.clone());
    Ok(())
});

let receive_sensor_messages = sensors_rx.for_each(move |line| {
    println!("Received sensor message, broadcasting: {:?}", line);
    let shared = state2.lock().unwrap();
    for (_, tx) in &shared.clients { // better: `for tx in shared.clients.values()`
        tx.unbounded_send(line.clone()).unwrap();
    }
    Ok(())
});

You could omit state1 here, but I find it cleaner to do it like this.

The reason for this is, that you move in the value state into the first closure, so you cannot use it in the second closure, because it is already moved (makes sense, doesn't it?).

user_10219200
Related