Rust: 別のスレッドでの処理結果を Future にするには?

Question

前提・実現したいこと

非同期処理の結果を Future で返す方法に付いて教えて下さい。

1. ２ つのスレッド Main と Worker が存在し、ジョブキューを共有しています。つまり channel の Sender, Receiver で通信しています。
2. Worker はジョブキューを監視して処理が来たらキューから取り出し、Worker スレッド内でそれを実行します。
3. Main スレッドから呼ばれるある関数は、処理をジョブキューに投入して非同期で Worker で実行させ、関数自体は即座に Future を返します。

単に非同期処理を行うだけであれば async を使うことができますが、この質問の背景では別の非同期処理と組み合わせているためこのような構成になっています (具体的には epoll() や select() を使うケースのように Worker スレッドがイベントループしている想定です)。

発生している問題・エラーメッセージ

後述のソースコードの enqueue() 関数の返値を Future<R> にして let result = exec(); の result を非同期で取得するにはどうしたら良いでしょうか? 最終的には以下のように記述できるようにしたいと思っています。

let now = euqueue(&sender, move || SystemTime::now()).await;

該当のソースコード

use std::sync::mpsc::{channel, Receiver, Sender};
use std::thread::{sleep, spawn};
use std::time::{Duration, SystemTime};
use std::fmt::Debug;

type TASK = Box<dyn FnMut() + Send>;

fn main() {
  let (sender, receiver) = channel::<TASK>();
  spawn(move || event_loop(receiver));
  loop {
    sleep(Duration::from_secs(1));
    euqueue(&sender, move || SystemTime::now());
  }
}

fn euqueue<R, F>(sender: &Sender<TASK>, mut exec: F) where F: FnMut() -> R + Send, F: 'static, R:Debug {
  sender.send(Box::new(move || {
    let result = exec();   // How to return this result as Future for function enqueue()?
    println!("task has been executed with: {:?}", result);
  })).unwrap();
}

fn event_loop(receiver: Receiver<TASK>) {
  loop {
    let mut exec = receiver.recv().unwrap();
    exec();
  }
}

補足情報（FW/ツールのバージョンなど）

> rustc --version
rustc 1.48.0 (7eac88abb 2020-11-16)

Answer 1

Futureトレイトを実装した型を定義して、それを返すことで実現できそうです。

Rust公式ドキュメントのひとつである「Asynchronous Programming in Rust」に載っている例が参考になると思います。

• Applied: Build a Timer -- Asynchronous Programming in Rust

その例ではTimerFutureという型を定義してFutureトレイトを実装します。TimerFutureは指定した時間が経過すると終了するFutureで、内部では別スレッドを起動してthread::sleep()で時間を計ります。

以下のように動作します。

1. TimerFuture::new(Duration)を呼ぶとTimerFutureのインスタンスが作られ、スレッドが起動される。TimerFutureインスタンスと、起動したスレッド上で実行されるクロージャとの間では、Arc<Mutex<SharedState>>というデータを共有することで連携できるようになっている
2. TimerFutureのインスタンスを非同期ランタイムで実行すると、非同期ランタイムはTimerFutureインスタンスのFuture::poll()メソッドを呼ぶ
3. TimerFutureインスタンスは、まだ指定時間が経過してないなら、poll()で渡されたWakerというオブジェクトをSharedStateに保存してからPoll::Pendingを返す
4. TimerFutureのインスタンスが起動したスレッドでは、指定時間が経過したら（thread::sleep()が終わったら）SharedStateにセットされたWakerのwake()メソッドを呼ぶ
5. 非同期ランタイムはwake()が呼ばれたことでTimerFutureのpoll()を再度呼ぶようスケジューリングする
6. 非同期ランタイムはTimerFutureインスタンスのpoll()メソッドを呼ぶ
7. TimerFutureインスタンスは指定時間が経過したので、Poll::Ready(計算結果の値)を返す。（この例では計算結果自体は意味を持たないので()値を返している）
8. 非同期ランタイムはTimerFutureインスタンスの実行を終了する

この例と同じようにFutureを実装した型を定義して、new()で処理をジョブキューに投入します。

その型と処理オブジェクトの間ではArc<Mutex<_>>のようなものでデータを共有するようにして、処理が終わるときにWaker::wake()を呼ぶようにします。

そして上の例と同じようにpoll()メソッドを実装すれば完成です。

追記

ご提示のコードに上の内容を実装してみました。

Futureを実装した型を定義して、new()で処理をジョブキューに投入します。

このコードではTaskFutureという型を定義しましたが、それのnew()で処理（Task）をジョブキューに投入するのではなく、enqueue()関数内で行うようにしました。

// Cargo.tomlに以下を追加する
// [dependencies]
// tokio = { version = "0.3", features = ["rt", "rt-multi-thread"] }

use std::{
    fmt::Debug,
    future::Future,
    pin::Pin,
    sync::mpsc::{channel, Receiver, Sender},
    sync::{Arc, Mutex},
    task::{Context, Poll, Waker},
    thread::{sleep, spawn},
    time::{Duration, SystemTime},
};
use tokio::runtime::Runtime;

struct Task {
    exec: Box<dyn Send + FnMut() -> Box<dyn Debug + Send>>,
    shared_state: Arc<Mutex<SharedState>>,
}

impl Task {
    fn new<E>(exec: E) -> Self
    where
        E: Send + 'static + FnMut() -> Box<dyn Debug + Send>,
    {
        Self {
            exec: Box::new(exec),
            shared_state: Arc::new(Mutex::new(SharedState {
                result: None,
                waker: None,
            })),
        }
    }

    fn execute(&mut self) {
        let result = (self.exec)();
        println!("task has been executed with: {:?}", result);
        let mut shared_state = self.shared_state.lock().unwrap();
        shared_state.result = Some(result);
        if let Some(waker) = shared_state.waker.take() {
            waker.wake()
        }
    }
}

struct TaskFuture {
    shared_state: Arc<Mutex<SharedState>>,
}

impl Future for TaskFuture {
    type Output = Box<dyn Debug>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let mut shared_state = self.shared_state.lock().unwrap();
        if let Some(result) = shared_state.result.take() {
            Poll::Ready(result)
        } else {
            shared_state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

struct SharedState {
    result: Option<Box<dyn Debug + Send>>,
    waker: Option<Waker>,
}

fn main() {
    let (sender, receiver) = channel::<Task>();
    spawn(move || event_loop(receiver));

    let rt = Runtime::new().unwrap();

    loop {
        sleep(Duration::from_secs(1));
        let task = Task::new(|| Box::new(SystemTime::now()));
        // tokioの非同期ランタイムでasyncブロックを実行する
        rt.block_on(async {
            let now = enqueue(&sender, task).await;
            println!("result has been received: {:?}", now);
        });
    }
}

fn enqueue(sender: &Sender<Task>, task: Task) -> TaskFuture {
    let future = TaskFuture {
        shared_state: Arc::clone(&task.shared_state),
    };
    sender.send(task).unwrap();
    future
}

fn event_loop(receiver: Receiver<Task>) {
    loop {
        let mut task = receiver.recv().unwrap();
        task.execute();
    }
}

Comments

[退会済みユーザー]

イベントループ時のパターンとして理解しやすいソースとともに詳しい解説をありがとうございます。
この回答を見ていくつか調べましたが、Rust の Future は Future そのものから実装しないといけないんですね。Scala の Future や Java の CompletableFuture、その他の言語の同等機能に慣れていたので、スレッド間で Future/Promise を共有して Worker が set すれば Main 側で get できるようになる使い方をイメージしていました。