concurrency - Rust缓存异步特征

Question

当我尝试缓存一个有效值的值并在它变为无效值时对其进行更新时，我遇到了一个问题。我相信问题是由于我试图在异步执行之间共享状态。此外，该组件位于多线程/并发环境中。
我看到的我不知道如何解决的错误是

future is not `Send` as this value is used across an await

以下是一个我可以提出的最小示例(它还具有一些所有权问题)，通常可以捕获我的用例和所遇到的问题。 Here是代码的游乐场。

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use std::sync::{Arc, Mutex};

struct Creds {
    expires_at: DateTime<Utc>,
}

impl Creds {
    fn is_expired(&self) -> bool {
        self.expires_at.le(&Utc::now())
    }
}

#[async_trait]
trait CredsProvider {
    async fn get_creds(&self) -> Creds;
}

struct MyCredsProvider {
    cached_creds: Arc<Mutex<Option<Creds>>>,
}

impl MyCredsProvider {
    fn new() -> Self {
        MyCredsProvider {
            cached_creds: Arc::new(Mutex::new(None)),
        }
    }
    async fn inner_get_creds(&self) -> Creds {
        todo!()
    }
}

#[async_trait]
impl CredsProvider for MyCredsProvider {
    async fn get_creds(&self) -> Creds {
        let mg = self
            .cached_creds
            .lock()
            .expect("Unable to get lock on creds mutex");
        if mg.is_some() && !mg.as_ref().unwrap().is_expired() {
            return mg.unwrap();
        }
        let new_creds = self.inner_get_creds().await;
        *mg = Some(new_creds);
        return new_creds;
    }
}

#[tokio::main]
async fn main() {
    MyCredsProvider::new();
    // Some multi-threaded / concurrent logic to periodically refresh creds
    todo!()
}

我不确定如何在示例中包括它，但是在 main中，想象并发/并行运行的多个工作线程每个都调用 CredsProvider.get_creds，然后使用这些凭据执行一些工作(如果您可以将其添加到完整的工作示例中，那d非常感谢我的启发)。假定 MyCredsProvider.inner_get_creds昂贵，并且仅应在缓存的凭据过期时才调用。
我该如何解决？我以为 Arc<Mutex<>>就足够了，但似乎还不够。在某一时刻，我尝试制作 Creds和trait，以便可以使用 Arc<Mutex<Option<Box<dyn Creds + Send + Sync>>>>，但是那感觉像是错误的路径，因此不起作用。
谢谢。

Answer 1

您可能想切换到tokio::sync::Mutex(playground)。
它解决了

future is not `Send` as this value is used across an await

代码:

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use std::sync::Arc;
use tokio::sync::Mutex;

#[derive(Clone)]
struct Creds {
    expires_at: DateTime<Utc>,
}

impl Creds {
    fn is_expired(&self) -> bool {
        self.expires_at.le(&Utc::now())
    }
}

#[async_trait]
trait CredsProvider {
    async fn get_creds(&self) -> Creds;
}

struct MyCredsProvider {
    cached_creds: Arc<Mutex<Option<Creds>>>,
}

impl MyCredsProvider {
    fn new() -> Self {
        MyCredsProvider {
            cached_creds: Arc::new(Mutex::new(None)),
        }
    }
    async fn inner_get_creds(&self) -> Creds {
        todo!()
    }
}

#[async_trait]
impl CredsProvider for MyCredsProvider {
    async fn get_creds(&self) -> Creds {
        let mut mg = self
            .cached_creds
            .lock()
            .await;
        if mg.is_some() && !mg.as_ref().unwrap().is_expired() {
            return mg.clone().unwrap();
        } else {
            let new_creds = self.inner_get_creds().await;
            *mg = Some(new_creds.clone());
            return new_creds;
        }
    }
}

#[tokio::main]
async fn main() {
    MyCredsProvider::new();
    // Some multi-threaded / concurrent logic to periodically refresh creds
    todo!()
}