Multithreading

Rune is thread safe, but the Vm does not implement Sync so cannot directly be shared across threads. This section details instead how you are intended to use Rune in a multithreaded environment.

Compiling a Unit and a RuntimeContext are expensive operations compared to the cost of calling a function. So you should try to do this as little as possible. It is appropriate to recompile a script when the source of the script changes. See the Hot reloading section for more information on this.

Once you have a Unit and a RuntimeContext they are thread safe and can be used by multiple threads simultaneously through Arc<Unit> and Arc<RuntimeContext>. Constructing a Vm with these through Vm::new is a very cheap operation.

#![allow(unused)]
fn main() {
let unit: Arc<Unit> = /* todo */;
let context: Arc<RuntimeContext> = /* todo */;

std::thread::spawn(move || {
    let mut vm = Vm::new(unit, context);
    let value = vm.call(["function"], (42,))?;
    Ok(())
});
}

Virtual machines do allocate memory. To avoide this overhead you’d have to employ more advanced techniques, such as storing virtual machines in a pool or thread locals. Once a machine has been acquired the Unit and RuntimeContext associated with it can be swapped out to the ones you need using Vm::unit_mut and Vm::context_mut respectively.

Using Vm::send_execute is a way to assert that a given execution is thread safe. And allows you to use Rune in asynchronous multithreaded environments, such as Tokio. This is achieved by ensuring that all captured arguments are ConstValue’s, which in contrast to Value’s are guaranteed to be thread-safe:

use rune::alloc::prelude::*;
use rune::sync::Arc;
use rune::termcolor::{ColorChoice, StandardStream};
use rune::{Diagnostics, Vm};

#[tokio::main]
async fn main() -> rune::support::Result<()> {
    let context = rune_modules::default_context()?;
    let runtime = Arc::try_new(context.runtime()?)?;

    let mut sources = rune::sources! {
        entry => {
            async fn main(timeout) {
                time::sleep(time::Duration::from_secs(timeout)).await
            }
        }
    };

    let mut diagnostics = Diagnostics::new();

    let result = rune::prepare(&mut sources)
        .with_context(&context)
        .with_diagnostics(&mut diagnostics)
        .build();

    if !diagnostics.is_empty() {
        let mut writer = StandardStream::stderr(ColorChoice::Always);
        diagnostics.emit(&mut writer, &sources)?;
    }

    let unit = result?;
    let unit = Arc::try_new(unit)?;
    let vm = Vm::new(runtime, unit);

    let execution = vm.try_clone()?.send_execute(["main"], (5u32,))?;
    let t1 = tokio::spawn(async move {
        execution.complete().await.unwrap();
        println!("timer ticked");
    });

    let execution = vm.try_clone()?.send_execute(["main"], (2u32,))?;
    let t2 = tokio::spawn(async move {
        execution.complete().await.unwrap();
        println!("timer ticked");
    });

    tokio::try_join!(t1, t2).unwrap();
    Ok(())
}

Statics across threads

Static storage follows the same rule as the Vm: a Globals holds Value’s, so it cannot be shared between threads. It is built per machine, alongside the machine, out of the shared Unit:

#![allow(unused)]
fn main() {
let globals = Globals::new(unit.clone())?;
let mut vm = Vm::new(runtime, unit).with_globals(globals);
}

This is not a limitation so much as a division of labour. What makes a static shared across threads is the value you put in it, not the storage holding it. Assign a host type whose interior is synchronized, such as an Arc<Mutex<..>>, an Arc<RwLock<..>>, an atomic, or a channel sender, and it is up to that value to decide how concurrent access is handled. Each thread gets its own slot, and every slot points at the same interior:

#![allow(unused)]
fn main() {
#[derive(Debug, Clone, Any)]
struct Counter {
    inner: Arc<Mutex<i64>>,
}
}

Cloning Counter produces another handle to the same Mutex, so each thread can be handed its own clone before it builds its machine. Note that the clone has to happen on the outside: a Value cannot cross a thread boundary, so each thread converts its own clone into one.

This composes with the pool described above. A pooled machine keeps the storage it was constructed with, so a worker which picks a machine out of the pool already has the statics associated with it, so there is nothing to re-inject per task. Warming the pool once means each machine resolves its statics once, and every one of them still reaches the same shared interior.

//! Sharing state between virtual machines on different threads through a
//! `static`.
//!
//! Static storage cannot cross a thread boundary, so every thread builds its
//! own [`Globals`] out of the same [`Unit`]. What makes the state shared is the
//! *value* placed in the slot: a host type whose interior is an
//! `Arc<Mutex<..>>`, cloned once per thread.

use std::sync::{Arc as StdArc, Mutex};
use std::thread;

use rune::runtime::Globals;
use rune::sync::Arc;
use rune::termcolor::{ColorChoice, StandardStream};
use rune::{Any, ContextError, Diagnostics, Module, Vm};

/// A counter shared by every virtual machine, on every thread.
///
/// Cloning it produces another handle to the same interior, which is what makes
/// it usable from more than one thread at a time.
#[derive(Debug, Clone, Any)]
struct Counter {
    inner: StdArc<Mutex<i64>>,
}

impl Counter {
    /// Bump the counter and return its new value.
    #[rune::function]
    fn increment(&self) -> i64 {
        let mut count = self.inner.lock().unwrap();
        *count += 1;
        *count
    }
}

fn module() -> Result<Module, ContextError> {
    let mut m = Module::new();
    m.ty::<Counter>()?;
    m.function_meta(Counter::increment)?;
    Ok(m)
}

const THREADS: usize = 4;
const CALLS: usize = 1000;

fn main() -> rune::support::Result<()> {
    let mut context = rune_modules::default_context()?;
    context.install(module()?)?;
    let runtime = Arc::try_new(context.runtime()?)?;

    let mut sources = rune::sources!(
        entry => {
            static COUNTER;

            pub fn main() {
                COUNTER.increment()
            }
        }
    );

    let mut diagnostics = Diagnostics::new();

    let result = rune::prepare(&mut sources)
        .with_context(&context)
        .with_diagnostics(&mut diagnostics)
        .build();

    if !diagnostics.is_empty() {
        let mut writer = StandardStream::stderr(ColorChoice::Always);
        diagnostics.emit(&mut writer, &sources)?;
    }

    // The unit and the runtime are `Send + Sync`, so they are compiled once and
    // shared. The storage and the virtual machines are not, so they are built
    // per thread.
    let unit = Arc::try_new(result?)?;

    let counter = Counter {
        inner: StdArc::new(Mutex::new(0)),
    };

    let mut handles = Vec::new();

    for _ in 0..THREADS {
        let unit = unit.clone();
        let runtime = runtime.clone();
        // A new handle to the same interior.
        let counter = counter.clone();

        handles.push(thread::spawn(move || -> rune::support::Result<()> {
            let globals = Globals::new(unit.clone())?;
            globals.set(["COUNTER"], rune::to_value(counter)?)?;

            let mut vm = Vm::new(runtime, unit).with_globals(globals);

            for _ in 0..CALLS {
                vm.call(["main"], ())?;
            }

            Ok(())
        }));
    }

    for handle in handles {
        handle.join().unwrap()?;
    }

    let total = *counter.inner.lock().unwrap();
    println!("total: {total}");
    Ok(())
}

$> cargo run --example statics_threads
total: 4000

Finally Function::into_sync exists to coerce a function into a SyncFunction, which is a thread-safe variant of a regular Function. This is a fallible operation since all values which are captured in the function-type in case its a closure has to be coerced to ConstValue. If this is not the case, the conversion will fail.