重构函数时如何避免出现生命周期问题？

Question

Playground如果你想直接跳转到代码中。

问题

我正在尝试实现一个函数filter_con<T, F>(v: Vec<T>, predicate: F)，它允许通过async 谓词对Vec 进行并发过滤。

也就是说，而不是做：

let arr = vec![...];
let arr_filtered = join_all(arr.into_iter().map(|it| async move {
    if some_getter(&it).await > some_value {
        Some(it)
    } else {
        None
    }
}))
.await
.into_iter()
.filter_map(|it| it)
.collect::<Vec<T>>()

每次我需要过滤Vec 时，我都希望能够：

let arr = vec![...];
let arr_filtered = filter_con(arr, |it| async move { 
  some_getter(&it).await > some_value 
}).await

暂定实施

我已将函数提取到它自己的函数中，但我遇到了生命周期问题

async fn filter_con<T, B, F>(arr: Vec<T>, predicate: F) -> Vec<T>
where
    F: FnMut(&T) -> B,
    B: futures::Future<Output = bool>,
{
    join_all(arr.into_iter().map(|it| async move {
        if predicate(&it).await {
            Some(it)
        } else {
            None
        }
    }))
    .await
    .into_iter()
    .filter_map(|p| p)
    .collect::<Vec<_>>()
}

error[E0507]: cannot move out of a shared reference

我不知道我要从谓词中移出什么？

更多详情，请参阅playground。

Answer 1

您将无法将谓词设为FnOnce，因为如果您的Vec 中有10 个项目，您将需要调用谓词10 次，但FnOnce 只能保证它可以调用一次，这可能会导致这样的事情：

let vec = vec![1, 2, 3];
let has_drop_impl = String::from("hello");

filter_con(vec, |&i| async {
  drop(has_drop_impl);
  i < 5
}

所以F 必须是FnMut 或Fn。标准库Iterator::filter 采用FnMut，尽管这可能会造成混淆（闭包 的捕获变量需要可变引用，而不是迭代器的元素）。

因为谓词是FnMut，所以任何调用者都需要能够获得&mut F。对于Iterator::filter，这可以用来做这样的事情：

let vec = vec![1, 2, 3];
let mut count = 0;

vec.into_iter().filter(|&x| {
  count += 1;  // this line makes the closure an `FnMut`
  x < 2
})

但是，通过将迭代器发送到 join_all，您实际上是允许异步运行时根据需要安排这些调用，可能同时进行，这将导致别名 &mut T，这始终是未定义的行为。此问题是同一问题https://github.com/rust-lang/rust/issues/69446 的略微精简版本。

我仍然不是 100% 了解细节，但似乎编译器在这里很保守，甚至不允许您首先创建闭包以防止出现可靠性问题。

我建议让您的函数只接受Fns。这样，您的运行时可以随意调用该函数。这确实意味着您的闭包不能具有可变状态，但这在 tokio 应用程序中不太可能成为问题。对于计数示例，“正确”的解决方案是使用AtomicUsize（或等效项），它允许通过共享引用进行突变。如果您在 filter 调用中引用可变状态，它应该是线程安全的，并且线程安全数据结构通常允许通过共享引用进行突变。

鉴于该限制，以下给出了您期望的答案：

async fn filter_con<T, B, F>(arr: Vec<T>, predicate: F) -> Vec<T>
where
    F: Fn(&T) -> B,
    B: Future<Output = bool>,
{
    join_all(arr.into_iter().map(|it| async {
        if predicate(&it).await {
            Some(it)
        } else {
            None
        }
    }))
    .await
    .into_iter()
    .filter_map(|p| p)
    .collect::<Vec<_>>()
}

Playground