How to pass channel into Closure?

[dakom]

Picking up from #1126

Consider the following:

use futures::{Future, Async, Poll};
use futures::sync::oneshot::{Sender, Receiver, channel};
use futures::task::current;
use wasm_bindgen::prelude::*;
use wasm_bindgen::JsCast;
use web_sys::*;

struct AsyncCounter {
    value: f64,
    target: f64,
    state: CounterState,
}

enum CounterState {
    Init,
    Counting { err_receiver: Receiver<JsValue> },
    Error (JsValue)
}
impl Future for AsyncCounter {
    type Item = JsValue;
    type Error = JsValue;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        match &self.state {
            CounterState::Init => {
                //Setup callback for updates
                let (err_sender, err_receiver):(Sender<JsValue>, Receiver<JsValue>) = channel();
                let task = current();

                let closure = Closure::wrap(Box::new(move || {
                    //This is not allowed
                    err_sender.send(JsValue::from_str("foo"));
                    task.notify();
                }) as Box<FnMut()>);

                // TODO: handle cancellation with clearTimeout
                window().unwrap().set_timeout_with_callback_and_timeout_and_arguments_0(closure.as_ref().unchecked_ref(), 1000).unwrap();

                //Move to counting state
                self.state = CounterState::Counting {err_receiver} ;
                let task = current();
                task.notify();
                Ok(Async::NotReady)
            },
            CounterState::Counting {err_receiver}  => {
                if self.value == self.target {
                    Ok(Async::Ready(JsValue::from_f64(self.value as f64)))

                } else {
                    self.value += 1.0;
                    Ok(Async::NotReady)
                }
            },
            CounterState::Error (err) => {
                Err(err.clone())
            }
        }
    }
}

If I remove that line: err_sender.send(JsValue::from_str("foo")); then it compiles fine. But with that line, I can't seem to get around the trait constraints.

Any tips?

(I've left out the receiver polling just to keep the example a bit more concise - but we can imagine that polling it could move the state to CounterState::Error)

[chpio]

could it be because of:

• the closure being a FnMut, not FnOnce
• and the sender consuming it self in the send method

?

[alexcrichton]

Sounds like @chpio has already solve this! With Closure you can't use FnOnce (yet) so you'll want to wrap the err_sender in an Option which you can take().unwrap() to assert the closure is only called once

[Pauan]

Is this a common Rustism?

Yes. Or at least, I've used it a fair amount.

[chpio]

Is this a common Rustism?

i think it is, replace (take is just a sugar-y wrapper around replace) is used a lot in rust.

[Pauan]

@dakom It's just the natural consequence of Rust's memory management (i.e. no garbage collector).

When the Rust compiler sees a closure, like this...

move || {
    do_something(foo);
}

There are three possibilities: the closure can be Fn, FnMut, or FnOnce.

The Rust compiler tries to make the closure as useful as possible. Since Fn can be used anywhere, Rust first tries to make the closure Fn.

However, if the closure captures mutable references then that doesn't work. So then Rust tries to make the closure FnMut, which works with mutable references.

But if the closure captures a variable and then tries to move it, it cannot be Fn or FnMut. So in that case the Rust compiler makes the closure FnOnce, which is the most restrictive.

Why can't the closure be Fn or FnMut? Consider this closure:

let x = vec![];

let f = move || {
    drop(x);
};

Here we are moving the variable x into the closure f. The closure tries to move the variable x (because it calls the drop function, which consumes its argument by value).

The Rust compiler rightfully says that the type of f is FnOnce(). What would happen if the Rust compiler used Fn() or FnMut() instead? Well, in that case we could drop x twice!

f();  // Drops x
f();  // Drops x again!

Oops, now we have a double-free, and therefore undefined behavior. Double dropping isn't the only bad thing that can happen, use-after-free could happen as well!

This happens because Fn and FnMut can be called multiple times. To prevent that, Rust restricts the closure to be FnOnce, which means it can only be called one time.

How does the Rust compiler enforce that an FnOnce closure is only called once? As an example, let's look at the definition of Option::map:

pub fn map<U, F>(self, f: F) -> Option<U> where F: FnOnce(T) -> U {
    match self {
        Some(x) => Some(f(x)),
        None => None,
    }
}

Notice that the type signature says that it accepts an FnOnce(T) -> U closure.

What that means is that the map function promises it will only call the closure one time, and this is enforced within the body of the function. So if map tries to call f multiple times then it will get a compile error.

What if you changed the map function to use FnMut or Fn instead? In that case it would still work if you called map with an Fn or FnMut closure, however if you tried to call map with an FnOnce closure then it will error.

The way to think of this is that there are two sides: the side which creates the closure, and the side which calls the closure.

For creation, it is most useful to create Fn closures, since they can be called as Fn, FnMut, or FnOnce. It is most restrictive to create FnOnce closures, since they can only be called as FnOnce.

For calling, it is most useful to accept FnOnce closures, since any closure can be FnOnce. And it is most restrictive to accept Fn closures, since then the closure cannot capture mutable references, and it cannot move captured variables.

Another way to think about it is that the caller decides what capabilities the closure has: if a caller (like map) accepts an Fn, then that means the closure can only be Fn.

On the other hand, if the caller accepts FnMut, then that means the closure can be Fn or FnMut. And if the caller accepts FnOnce then that means the closure can be Fn, FnMut, or FnOnce.

Right now wasm-bindgen only accepts FnMut closures, which means you can use Fn or FnMut closures with wasm-bindgen, but you cannot use FnOnce closures.

Using Some(value) and then using .take().unwrap() is one way to "convert" an FnOnce closure into an FnMut closure:

let mut x = Some(vec![]);

let f = move || {
    let y = x.take().unwrap();
    drop(y);
};

Notice that the closure is no longer trying to move x, instead it is moving y. That's okay, because y isn't a captured variable, it's a local variable.

Since the closure doesn't move captured variables, it is no longer FnOnce, instead it is now FnMut.

That means you will get a runtime panic if the closure is called multiple times, but that's better than undefined behavior!

You can read more about the nitty gritty of Rust closures here:

https://krishnasannasi.github.io/rust/syntactic/sugar/2019/01/17/Closures-Magic-Functions.html
http://huonw.github.io/blog/2015/05/finding-closure-in-rust/
https://stevedonovan.github.io/rustifications/2018/08/18/rust-closures-are-hard.html
http://gradebot.org/doc/ipur/closure.html

[Pauan]

P.S. With the final example, you might be wondering why it no longer tries to move x.

The reason is actually really simple: the Option::take method accepts &mut self, not self, and therefore it does not consume its argument, instead it accepts a mutable reference to its argument.

Because x.take() does not consume x, and since nothing else touches x, we can conclude that x is not moved.

On the other hand, the Sender::send method does accept self, and therefore it consumes it. That's why calling err_sender.send(...) causes err_sender to be moved.

This is the sort of reasoning that the Rust compiler uses to determine whether a variable is moved or not. It's all based purely on the types.

[Pauan]

(btw small typo - I think you forgot the F: on where FnOnce(T) -> U )

You're right, I've fixed it, thanks!

[chpio]

let f = move || {
    let y = x.take().unwrap();
    drop(y);
};

x isn't moved there?

yes, it's moved into the closure, but you call the closure (in case of a FnMut) by &mut, so it can only give you &mut access to the captured values (that's because it can be called multiple times).

[Pauan]

@dakom ahhh... and I vaguely remember getting stuck before on that exact same point about not-moving x because it doesn't need to, e.g. the move keyword being more like a hint rather than an explicit instruction.

The move keyword for closures is different from the issue we're referring to.

Without the move keyword, the Rust compiler will infer which variables need to be moved into the closure and which ones don't. It does this inference by looking at how the variables are used in the closure's body.

With the move keyword, you are telling the Rust compiler that all variables should be moved into the closure, regardless of how the variables are used inside the closure.

I think it's really important to understand how closures are implemented in Rust. Let's consider this example:

let mut x: Option<Vec<u32>> = Some(vec![]);

let f = || {
    let y = x.take().unwrap();
    drop(y);
};

What does Rust compile the above code into? Well, it will be something like this:

let mut x: Option<Vec<u32>> = Some(vec![]);

let f = {
    struct Closure13925 {
        x: &mut Option<Vec<u32>>,
    }

    impl FnMut<()> for Closure13925 {
        type Output = ();

        fn call_mut(&mut self, args: ()) -> () {
            let y = self.x.take().unwrap();
            drop(y);
        }
    }

    Closure13925 {
        x: &mut x,
    }
};

Notice that the closure actually got converted into a struct (which has a unique name so it doesn't collide with other structs). The links I posted go into more detail about this.

Each captured variable (in this case x) gets put as a field in the struct. And the body of the closure gets converted into a trait method (in this case FnMut).

Another important thing is that captured variables get replaced with struct lookups: as you can see it replaced x with self.x.

Lastly, notice that the struct contains a mutable reference to x, but it doesn't contain x itself.

What happens when you instead use the move keyword on the f closure?

let mut x: Option<Vec<u32>> = Some(vec![]);

let f = {
    struct Closure13925 {
        x: Option<Vec<u32>>,
    }

    impl FnMut<()> for Closure13925 {
        type Output = ();

        fn call_mut(&mut self, args: ()) -> () {
            let y = self.x.take().unwrap();
            drop(y);
        }
    }

    Closure13925 {
        x,
    }
};

In this case it's exactly the same, except rather than the struct containing a mutable reference to x, it instead contains x itself. In other words, x has been moved into the struct!

However, notice that the call_mut method is still using &mut self, so it can only access self.x mutably, it cannot move self.x.

So there are really two different kinds of movement here:

1. Moving variables into the closure struct (which is controlled with the move keyword).

2. Moving captured variables (which are just struct fields) out of the closure struct (which can only be done with FnOnce closures).

[Pauan]

Incidentally, your original closure...

move || {
    err_sender.send(JsValue::from_str("foo"));
    task.notify();
}

...would get compiled into something like this:

struct Closure13925 {
    err_sender: Sender<JsValue>,
    task: Task,
}

impl FnOnce<()> for Closure13925 {
    type Output = ();
    
    fn call_once(self, args: ()) -> () {
        self.err_sender.send(JsValue::from_str("foo"));
        self.task.notify();
    }
}

Closure13925 {
    err_sender,
    task,
}

Since you used the move keyword it moved err_sender and task into the closure struct.

And since Sender::send consumes self, that means the closure implements FnOnce (not FnMut).

You'll note that the FnOnce trait has a call_once method which accepts self (not &mut self!).

That means that when you call the closure it will consume the entire closure struct. That has two implications:

1. It guarantees that the closure can only be called once, because calling the closure moves the closure struct, and Rust prevents you from using things after they have been moved.

2. It allows you to move fields out of the closure struct, which is what allows you to call Sender::send.

However, since the closure struct implements FnOnce, but not FnMut or Fn, that means that you cannot use it with wasm-bindgen, since wasm-bindgen requires the closure to be Fn or FnMut.

Once you understand that closures are really just structs in disguise, everything makes perfect sense.

[Pauan]

Is the idea that once something consumes itself, it means the parent must also be consuming itself too, since it can't have a non-existent property?

Yes, the value must always exist. Rust does not have undefined values (except for unsafe code, of course).

The reason .take() works is that it's replacing the value with None. So it still has a value, it's just a useless value.

That's very different from having no value, which would be undefined behavior.

(and following from that, since the parent must be consuming itself in that situation, for closure's that's call_once() and therefore Rust knows it to be FnOnce?)

Yes, but it's actually the other way around:

1. The Rust compiler looks at the closure's body.

2. The Rust compiler sees err_sender.send(...), and it knows that err_sender is a captured variable, and it knows that Sender::send consumes self.

3. Therefore the Rust compiler knows that the closure must be FnOnce.

4. Therefore the Rust compiler generates the FnOnce impl for the closure.

And when you call a closure like f(...), Rust will replace the call with Fn::call(&f, ...), FnMut::call_mut(&mut f, ...), or FnOnce::call_once(f, ...) as appropriate.

The Fn, FnMut, and FnOnce traits are implementation details.

It's good to understand them, but you can't use them in your actual code, instead you use the magical Fn(Foo) -> Bar syntax, which desugars into the Fn, FnMut, and FnOnce traits.

So it's all just syntax sugar for structs + traits.

[alanpoon]

Hi I got runtime error. I followed the take and drop advice.

pub fn start_rpc_client_thread(
  url: String,
  jsonreq: String,
  result_in: ThreadOut<String>,
  on_message_fn: OnMessageFn,
) {
  console_log!("jsonreq: {:?}",jsonreq);
  let ws = WebSocket::new(&url).unwrap();
  let ws_c = ws.clone();
  let mut result_in_o = Some(result_in);
  let on_message = {
    Closure::wrap(Box::new(move |evt: MessageEvent| {
        let msgg = evt.data()
                    .as_string()
        .expect("Can't convert received data to a string");
        console_log!("message event, received data: {:?}", msgg);
        let res_e = (on_message_fn)(&msgg);
        match res_e {
          ResultE::None=>{},
          ResultE::Close=>{
            ws_c.close_with_code(1000).unwrap();
          },
          ResultE::S(s)=>{
            let y = result_in_o.take().unwrap();
            y.send(s).unwrap();
            drop(y);
          },
          ResultE::SClose(s)=>{
            let y = result_in_o.take().unwrap();
            y.send(s).unwrap();
            drop(y);
            ws_c.close_with_code(1000).unwrap();
          }
        }
    }) as Box<dyn FnMut(MessageEvent)>)
  };
  
  ws.set_onmessage(Some(on_message.as_ref().unchecked_ref()));
  on_message.forget();
}

Uncaught (in promise) RuntimeError: unreachable
at __rust_start_panic (wasm-function[600]:1)
at rust_panic (wasm-function[430]:31)
at std::panicking::rust_panic_with_hook::h3b27a5077fe826e8 (wasm-function[218]:286)
at std::panicking::begin_panic::h85bd2bcd3c2f6125 (wasm-function[417]:40)
at std::sys::wasm::condvar::Condvar::wait::h6e9f7dfa883b7206 (wasm-function[522]:13)
at std::sys_common::condvar::Condvar::wait::h90274ef8e52a9b85 (wasm-function[557]:7)
at std::thread::park::h451a3db3ca79bdf5 (wasm-function[123]:366)
at std::sync::mpsc::blocking::WaitToken::wait::hc140bbc00b818195 (wasm-function[347]:30)
at std::sync::mpsc::shared::Packet::recv::h4622443f21479084 (wasm-function[73]:286)
at std::sync::mpsc::Receiver::recv::hfeae20b228f11744 (wasm-function[45]:1592)
at substrate_api_client::Api

::_get_request::h54ce5d76aa34c112 (wasm-function[108]:177)

wasm-0012db0a-218:124 Uncaught RuntimeError: unreachable
at std::panicking::rust_panic_with_hook::h3b27a5077fe826e8 (wasm-function[218]:280)
at std::panicking::begin_panic::h85bd2bcd3c2f6125 (wasm-function[417]:40)
at std::thread::Thread::unpark::ha05dcf7e67a102ee (wasm-function[239]:199)
at std::sync::mpsc::blocking::SignalToken::signal::hc84301e060392bae (wasm-function[388]:47)
at std::sync::mpsc::shared::Packet::send::hb608d9669bdbf8c4 (wasm-function[132]:252)
at std::sync::mpsc::Sender::send::h595dc87ac604d02b (wasm-function[44]:1100)
at substrate_api_client::rpc::client::websys_client::start_rpc_client_thread::{{closure}}::hb3dcc934a3df8467 (wasm-function[46]:1012)
at <dyn core::ops::function::FnMut<(A,)>+Output = R as wasm_bindgen::closure::WasmClosure>::describe::invoke::h4e82836d4451b155 (wasm-function[448]:17)

[alanpoon]

Fixed replacing channel with futures signal