Error in Async closure: Lifetime may not live long enough; returning this value requires that `'1` must outlive `'2`

Viewed 71

I'm trying to use a Mutex<Sender> inside an async closure but I'm not entirely sure why I'm getting the error below:

error: lifetime may not live long enough
   --> src/main.rs:120:41
    |
120 |       io.add_method(MARK_ITSELF, move |_| async {
    |  ________________________________--------_^
    | |                                |      |
    | |                                |      return type of closure `impl std::future::Future<Output = Result<jsonrpc::serde_json::Value, jsonrpc_http_server::jsonrpc_core::Error>>` contains a lifetime `'2`
    | |                                lifetime `'1` represents this closure's body
121 | |         trace!("Mark itself!");
122 | |         let tx = sender.lock().map_err(to_internal)?;
123 | |         tx.send(Action::MarkItself)
124 | |             .map_err(to_internal)
125 | |             .map(|_| Value::Bool(true))
126 | |     });
    | |_____^ returning this value requires that `'1` must outlive `'2`
    |
    = note: closure implements `Fn`, so references to captured variables can't escape the closure

My main function looks like this:

use jsonrpc::serde_json::Value;
use jsonrpc::Error as ClientError;
use jsonrpc::{
    serde_json::value::RawValue,
    simple_http::{self, SimpleHttpTransport},
    Client,
};
use jsonrpc_http_server::jsonrpc_core::{Error as ServerError, IoHandler};
use jsonrpc_http_server::ServerBuilder;
use log::{debug, error, trace};
use serde::Deserialize;
use std::rc::Rc;
use std::sync::mpsc::Receiver;
use std::sync::Mutex;
use std::thread;
use std::{
    env, fmt,
    net::SocketAddr,
    sync::mpsc::{channel, Sender},
};

const START_ROLL_CALL: &str = "start_roll_call";
const MARK_ITSELF: &str = "mark_itself";

fn create_client(url: &str, user: &str, pass: &str) -> Result<Client, simple_http::Error> {
    let t = SimpleHttpTransport::builder()
        .url(url)?
        .auth(user, Some(pass))
        .build();

    Ok(Client::with_transport(t))
}

fn spawn_worker() -> Result<Sender<Action>, failure::Error> {
    let (tx, rx): (Sender<Action>, Receiver<Action>) = channel();
    let next: SocketAddr = env::var("NEXT")?.parse()?;
    thread::spawn(move || {
        let remote = Remote::new(next).unwrap();
        let mut in_roll_call = false;
        for action in rx.iter() {
            match action {
                Action::StartRollCall => {
                    if !in_roll_call {
                        if remote.start_roll_call().is_ok() {
                            debug!("ON");
                            in_roll_call = true;
                        }
                    } else {
                        if remote.mark_itself().is_ok() {
                            debug!("OFF");
                            in_roll_call = false;
                        }
                    }
                }
                Action::MarkItself => {
                    if in_roll_call {
                        if remote.mark_itself().is_ok() {
                            debug!("OFF");
                            in_roll_call = false;
                        }
                    } else {
                        debug!("SKIP");
                    }
                }
            }
        }
    });
    Ok(tx)
}

enum Action {
    StartRollCall,
    MarkItself,
}

struct Remote {
    client: Client,
}

impl Remote {
    fn new(addr: SocketAddr) -> Result<Self, simple_http::Error> {
        let url = format!("http://{}", addr);
        let client = create_client(&url, "", "")?;
        Ok(Self { client })
    }

    fn call_method<T>(&self, method: &str, params: &[Box<RawValue>]) -> Result<T, ClientError>
    where
        T: for<'de> Deserialize<'de>,
    {
        let request = self.client.build_request(method, params);
        self.client
            .send_request(request)
            .and_then(|res| res.result::<T>())
    }

    fn start_roll_call(&self) -> Result<bool, ClientError> {
        self.call_method(START_ROLL_CALL, &[])
    }

    fn mark_itself(&self) -> Result<bool, ClientError> {
        self.call_method(MARK_ITSELF, &[])
    }
}

fn main() -> Result<(), failure::Error> {
    env_logger::init();
    let tx = spawn_worker()?;
    let addr: SocketAddr = env::var("ADDRESS")?.parse()?;
    let mut io = IoHandler::default();
    let sender = Mutex::new(tx.clone());
    io.add_method(START_ROLL_CALL, move |_| async move {
        trace!("Starting roll call!");
        let tx = sender.lock().map_err(to_internal)?;
        tx.send(Action::StartRollCall)
            .map_err(to_internal)
            .map(|_| Value::Bool(true))
    });
    let sender = Mutex::new(tx.clone());
    io.add_method(MARK_ITSELF, move |_| async {
        trace!("Mark itself!");
        let tx = sender.lock().map_err(to_internal)?;
        tx.send(Action::MarkItself)
            .map_err(to_internal)
            .map(|_| Value::Bool(true))
    });
    let server = ServerBuilder::new(io).start_http(&addr)?;
    Ok(server.wait())
}

fn to_internal<E: fmt::Display>(err: E) -> ServerError {
    error!("Error: {}", err);
    ServerError::internal_error()
}

The main idea is to pass the mspc sender to the closure so that the method can send the Action(An enum). Is there something I'm doing wrong?

1 Answers

The problem is add_method requires that

  • your closure is static, and
  • the Future returned from your closure is static.

Try this

let sender = Arc::new(Mutex::new(tx.clone()));
io.add_method(START_ROLL_CALL, move |_| {
    let cloned_sender = sender.clone();
    async move {
        trace!("Starting roll call!");
        let tx = cloned_sender.lock().map_err(to_internal)?;
        tx.send(Action::StartRollCall)
            .map_err(to_internal)
            .map(|_| Value::Bool(true))
    }
});

Side note: you are using sync Mutex in an async environment. This undermines the benefits of async. Consider using async Mutex such as Tokio Mutex or async-lock Mutex.

Related