Rust 中的多线程记忆

Question

我正在用 Rust 开发一个我想要多线程的算法。该算法的本质是它产生重叠子问题的解决方案，因此我正在寻找一种实现多线程记忆的方法。

Pritchard 在this article 中介绍了（单线程）记忆的实现。

我希望扩展此功能：

1. 无论何时必须调用底层函数，包括递归，都会在新线程上异步评估结果。
2. 从上一点继续，假设我们有一些记忆函数f，和f(x)，需要递归调用f(x1)，f(x2)，...f(xn)。应该可以在单独的线程上同时评估所有这些递归调用。
3. 如果在当前正在评估其结果的输入上调用 memoised 函数，则当前线程应阻塞该线程，并在释放后以某种方式获取结果。这样可以确保我们不会出现多个线程尝试评估相同的结果。
4. 有一种方法可以在不阻塞当前线程的情况下强制评估和缓存f(x)（如果尚未缓存）。这允许程序员抢先开始对他们知道以后需要（或可能需要）的特定值的结果进行评估。

Answer 1

您可以做到这一点的一种方法是存储HashMap，其中键是f 的参数，值是包含结果的一次性消息的接收者。然后对于您需要的任何值：

• 如果地图中已经有接收器，请等待。
• 否则，生成一个 future 以开始计算结果，并将接收器存储在地图中。

这是一个非常人为的示例，花费的时间比应有的时间长，但成功运行 (Playground)：

use futures::{
    future::{self, BoxFuture},
    prelude::*,
    ready,
};
use std::{
    collections::HashMap,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll},
};
use tokio::sync::{oneshot, Mutex};

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct MemoInput(usize);

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct MemoReturn(usize);

/// This is necessary in order to make a concrete type for the `HashMap`.
struct OneshotReceiverUnwrap<T>(oneshot::Receiver<T>);

impl<T> Future for OneshotReceiverUnwrap<T> {
    type Output = T;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // Don't worry too much about this part
        Poll::Ready(ready!(Pin::new(&mut self.0).poll(cx)).unwrap())
    }
}

type MemoMap = Mutex<HashMap<MemoInput, future::Shared<OneshotReceiverUnwrap<MemoReturn>>>>;

/// Compute (2^n)-1, super inefficiently.
fn compute(map: Arc<MemoMap>, x: MemoInput) -> BoxFuture<'static, MemoReturn> {
    async move {
        // First, get all dependencies.
        let dependencies: Vec<MemoReturn> = future::join_all({
            let map2 = map.clone();
            let mut map_lock = map.lock().await;

            // This is an iterator of futures that resolve to the results of the
            // dependencies.
            (0..x.0).map(move |i| {
                let key = MemoInput(i);
                let key2 = key.clone();

                (*map_lock)
                    .entry(key)
                    .or_insert_with(|| {
                        // If the value is not currently being calculated (ie.
                        // is not in the map), start calculating it
                        let (tx, rx) = oneshot::channel();
                        let map3 = map2.clone();
                        tokio::spawn(async move {
                            // Compute the value, then send it to the receiver
                            // that we put in the map. This will awake all
                            // threads that were awaiting it.
                            tx.send(compute(map3, key2).await).unwrap();
                        });
                        // Return a shared future so that multiple threads at a
                        // time can await it
                        OneshotReceiverUnwrap(rx).shared()
                    })
                    .clone() // Clone one instance of the shared future for us
            })
        })
        .await;

        // At this point, all dependencies have been resolved!

        let result = dependencies.iter().map(|r| r.0).sum::<usize>() + x.0;
        MemoReturn(result)
    }
    .boxed() // Box in order to prevent a recursive type
}

#[tokio::main]
async fn main() {
    let map = Arc::new(MemoMap::default());
    let result = compute(map, MemoInput(10)).await.0;
    println!("{}", result); // 1023
}

注意：这当然可以更好地优化，这只是一个 POC 示例。