Struct tokio::task::LocalSet[−][src]

pub struct LocalSet { /* fields omitted */ }

This is supported on crate feature rt only.

Expand description

A set of tasks which are executed on the same thread.

In some cases, it is necessary to run one or more futures that do not implement Send and thus are unsafe to send between threads. In these cases, a local task set may be used to schedule one or more !Send futures to run together on the same thread.

For example, the following code will not compile:

use std::rc::Rc;

#[tokio::main]
async fn main() {
    // `Rc` does not implement `Send`, and thus may not be sent between
    // threads safely.
    let unsend_data = Rc::new("my unsend data...");

    let unsend_data = unsend_data.clone();
    // Because the `async` block here moves `unsend_data`, the future is `!Send`.
    // Since `tokio::spawn` requires the spawned future to implement `Send`, this
    // will not compile.
    tokio::spawn(async move {
        println!("{}", unsend_data);
        // ...
    }).await.unwrap();
}

Use with `run_until`

To spawn !Send futures, we can use a local task set to schedule them on the thread calling Runtime::block_on. When running inside of the local task set, we can use task::spawn_local, which can spawn !Send futures. For example:

use std::rc::Rc;
use tokio::task;

#[tokio::main]
async fn main() {
    let unsend_data = Rc::new("my unsend data...");

    // Construct a local task set that can run `!Send` futures.
    let local = task::LocalSet::new();

    // Run the local task set.
    local.run_until(async move {
        let unsend_data = unsend_data.clone();
        // `spawn_local` ensures that the future is spawned on the local
        // task set.
        task::spawn_local(async move {
            println!("{}", unsend_data);
            // ...
        }).await.unwrap();
    }).await;
}

Note: The run_until method can only be used in #[tokio::main], #[tokio::test] or directly inside a call to Runtime::block_on. It cannot be used inside a task spawned with tokio::spawn.

Awaiting a `LocalSet`

Additionally, a LocalSet itself implements Future, completing when all tasks spawned on the LocalSet complete. This can be used to run several futures on a LocalSet and drive the whole set until they complete. For example,

use tokio::{task, time};
use std::rc::Rc;

#[tokio::main]
async fn main() {
    let unsend_data = Rc::new("world");
    let local = task::LocalSet::new();

    let unsend_data2 = unsend_data.clone();
    local.spawn_local(async move {
        // ...
        println!("hello {}", unsend_data2)
    });

    local.spawn_local(async move {
        time::sleep(time::Duration::from_millis(100)).await;
        println!("goodbye {}", unsend_data)
    });

    // ...

    local.await;
}

Note: Awaiting a LocalSet can only be done inside #[tokio::main], #[tokio::test] or directly inside a call to Runtime::block_on. It cannot be used inside a task spawned with tokio::spawn.

Use inside `tokio::spawn`

The two methods mentioned above cannot be used inside tokio::spawn, so to spawn !Send futures from inside tokio::spawn, we need to do something else. The solution is to create the LocalSet somewhere else, and communicate with it using an mpsc channel.

The following example puts the LocalSet inside a new thread.

use tokio::runtime::Builder;
use tokio::sync::{mpsc, oneshot};
use tokio::task::LocalSet;

// This struct describes the task you want to spawn. Here we include
// some simple examples. The oneshot channel allows sending a response
// to the spawner.
#[derive(Debug)]
enum Task {
    PrintNumber(u32),
    AddOne(u32, oneshot::Sender<u32>),
}

#[derive(Clone)]
struct LocalSpawner {
   send: mpsc::UnboundedSender<Task>,
}

impl LocalSpawner {
    pub fn new() -> Self {
        let (send, mut recv) = mpsc::unbounded_channel();

        let rt = Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();

        std::thread::spawn(move || {
            let local = LocalSet::new();

            local.spawn_local(async move {
                while let Some(new_task) = recv.recv().await {
                    tokio::task::spawn_local(run_task(new_task));
                }
                // If the while loop returns, then all the LocalSpawner
                // objects have have been dropped.
            });

            // This will return once all senders are dropped and all
            // spawned tasks have returned.
            rt.block_on(local);
        });

        Self {
            send,
        }
    }

    pub fn spawn(&self, task: Task) {
        self.send.send(task).expect("Thread with LocalSet has shut down.");
    }
}

// This task may do !Send stuff. We use printing a number as an example,
// but it could be anything.
//
// The Task struct is an enum to support spawning many different kinds
// of operations.
async fn run_task(task: Task) {
    match task {
        Task::PrintNumber(n) => {
            println!("{}", n);
        },
        Task::AddOne(n, response) => {
            // We ignore failures to send the response.
            let _ = response.send(n + 1);
        },
    }
}

#[tokio::main]
async fn main() {
    let spawner = LocalSpawner::new();

    let (send, response) = oneshot::channel();
    spawner.spawn(Task::AddOne(10, send));
    let eleven = response.await.unwrap();
    assert_eq!(eleven, 11);
}

Struct tokio::task::LocalSet[−][src]

Use with run_until

Awaiting a LocalSet

Use inside tokio::spawn

Implementations

Notable traits for LocalSet

Notable traits for JoinHandle<T>

Panics

Notes

Examples

Trait Implementations

Notable traits for LocalSet

Auto Trait Implementations

Blanket Implementations

Use with `run_until`

Awaiting a `LocalSet`

Use inside `tokio::spawn`