rust - 无法推断合适的生命周期 : Borrowing mutable buffers to Futures

Question

我实现了一个具有以下高级逻辑的 Tokio Future:

1. 调用函数recv_dgram。这应该返回一个 Future，它在收到一个小数据报时就绪。
2. 当 Future 就绪时，保留小数据报。
3. 如果我们有足够的小数据报来重构大数据报，将Future设置为就绪并返回重构的数据报。否则，返回步骤 1。

我有一些终生无法理解的问题。我创建了一段独立的简化代码来演示它。

在下面的代码中，RecvMsg 是当接收到足够小的数据报时就绪的 Future。 recv_dgram 是一个返回 Future 的函数，当一个小数据报到达时，它就准备好了。

我正在尝试编译以下代码 ( playground ):

extern crate futures;
use self::futures::{Future, Poll, Async};

struct RecvMsg<'d,R>
where 
    R: for <'r> FnMut(&'r mut [u8]) -> Box<Future<Item=&'r mut [u8], Error=()> + 'r>,
{
    recv_dgram: R,
    temp_buff: Vec<u8>,
    opt_read_future: Option<Box<Future<Item=&'d mut [u8], Error=()> + 'd>>,
}

impl<'d,R> Future for RecvMsg<'d,R>
where 
    R: for <'r> FnMut(&'r mut [u8]) -> Box<Future<Item=&'r mut [u8], Error=()> + 'r>,
{
    type Item = ();
    type Error = ();

    fn poll(&mut self) -> Poll<Self::Item, ()> {

        // Obtain a future datagram, 
        let mut fdgram = (self.recv_dgram)(&mut self.temp_buff);
        // Code compiles if this line is commented out:
        self.opt_read_future = Some(fdgram);
        return Ok(Async::NotReady);
    }
}

fn main() {}

这是我收到的错误信息:

error[E0495]: cannot infer an appropriate lifetime for borrow expression due to conflicting requirements
  --> src/main.rs:25:44
   |
25 |         let mut fdgram = (self.recv_dgram)(&mut self.temp_buff);
   |                                            ^^^^^^^^^^^^^^^^^^^
   |
note: first, the lifetime cannot outlive the anonymous lifetime #1 defined on the method body at 22:5...
  --> src/main.rs:22:5
   |
22 | /     fn poll(&mut self) -> Poll<Self::Item, ()> {
23 | |
24 | |         // Obtain a future datagram, 
25 | |         let mut fdgram = (self.recv_dgram)(&mut self.temp_buff);
...  |
28 | |         return Ok(Async::NotReady);
29 | |     }
   | |_____^
note: ...so that reference does not outlive borrowed content
  --> src/main.rs:25:44
   |
25 |         let mut fdgram = (self.recv_dgram)(&mut self.temp_buff);
   |                                            ^^^^^^^^^^^^^^^^^^^
note: but, the lifetime must be valid for the lifetime 'd as defined on the impl at 13:1...
  --> src/main.rs:13:1
   |
13 | / impl<'d,R> Future for RecvMsg<'d,R>
14 | | where 
15 | |     R: for <'r> FnMut(&'r mut [u8]) -> Box<Future<Item=&'r mut [u8], Error=()> + 'r>,
16 | | {
...  |
29 | |     }
30 | | }
   | |_^
note: ...so that expression is assignable (expected std::option::Option<std::boxed::Box<futures::Future<Error=(), Item=&'d mut [u8]> + 'd>>, found std::option::Option<std::boxed::Box<futures::Future<Error=(), Item=&mut [u8]>>>)
  --> src/main.rs:27:32
   |
27 |         self.opt_read_future = Some(fdgram);
   |                                ^^^^^^^^^^^^

我对可能出错的地方有一些想法。我知道如果我注释掉这一行:

reading_state.opt_read_future = Some(fdgram);

代码编译成功。另外，我怀疑这里使用 reading_state.temp_buff 作为参数:

let mut fdgram = (reading_state.frag_msg_receiver.recv_dgram)(
                &mut reading_state.temp_buff);

与问题有关。 (另请参阅 Why can't I store a value and a reference to that value in the same struct?)

Answer 1

类型实现 Future但没有 'static生命周期非常无用，因为事件循环将无法运行。

另外我认为生命周期在Future::Item不好；我看不出类型系统如何帮助您在控制流中确定 future 实际完成释放生命周期的时间。

所以你可能想要 R: FnMut(Vec<u8>) -> Box<Future<Item=Vec<u8>, Error=()>>而不是 recv_dgram功能。它将接收一个缓冲区，它应该将新接收到的数据附加到(它可以使用 reserve() 和“不安全”缓冲区填充 + set_len() 进行优化)。您还可以将 Vec 包装在一些自定义类型中以强制执行“仅附加”语义。

现在您应该意识到可变性不会对这个函数有太大帮助 - 您可以连续调用它 10 次而无需等待返回的 futures 完成。相反，您可能希望传递显式状态。还让我们避免装箱，并接受任何可使用 IntoFuture 转换的结果:

// `S` is the state, `F` something convertible to a future `F::Future`.
pub struct RecvMsg<R, F, S>
where
    F: IntoFuture<Item=(S, Vec<u8>), Error=()>,
    R: FnMut(S, Vec<u8>) -> F,
{
    recv_dgram: R,
    opt_read_future: Option<F::Future>,
}

当前挂起的读取操作现在变为 Option<F::Future> ，并且缓冲区存在于该挂起的操作中。

只要您的包装器被轮询(您还没有完全完成)，就需要轮询这个挂起的读取操作!

总而言之，它可能看起来像这样:

Playground

extern crate futures;
use self::futures::{IntoFuture,Future, Poll, Async};

pub struct RecvMsg<R, F, S>
where
    F: IntoFuture<Item=(S, Vec<u8>), Error=()>,
    R: FnMut(S, Vec<u8>) -> F,
{
    recv_dgram: R,
    pending: Option<F::Future>,
}

impl<R, F, S> RecvMsg <R, F, S>
where
    F: IntoFuture<Item=(S, Vec<u8>), Error=()>,
    R: FnMut(S, Vec<u8>) -> F,
{
    pub fn new(mut recv_dgram: R, initial_state: S) -> Self {
        let start = recv_dgram(initial_state, Vec::new()).into_future();
        RecvMsg{
            recv_dgram: recv_dgram,
            pending: Some(start)
        }
    }
}

impl<R, F, S> Future for RecvMsg <R, F, S>
where
    F: IntoFuture<Item=(S, Vec<u8>), Error=()>,
    R: FnMut(S, Vec<u8>) -> F,
{
    type Item = Vec<u8>;
    type Error = ();

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        match self.pending.take() {
            Some(mut p) => {
                match p.poll()? {
                    Async::Ready((state, buf)) => {
                        if buf.len() > 1024 {
                            // enough data
                            Ok(Async::Ready(buf))
                        } else {
                            // not enough data, read more
                            let next = (self.recv_dgram)(state, buf).into_future();
                            self.pending = Some(next);
                            Ok(Async::NotReady)
                        }
                    },
                    Async::NotReady => {
                        // still waiting for more data
                        self.pending = Some(p);
                        Ok(Async::NotReady)
                    },
                }
            },
            None => Ok(Async::NotReady), // already completed
        }
    }
}