What does the Macro futures::ready do?

1

So I got this function (from here):

pub fn poll_recv(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<KcpResult<usize>> {
        loop {
            // Consumes all data in buffer
            if self.recv_buffer_pos < self.recv_buffer_cap {
                let remaining = self.recv_buffer_cap - self.recv_buffer_pos;
                let copy_length = remaining.min(buf.len());

                buf[..copy_length]
                    .copy_from_slice(&self.recv_buffer[self.recv_buffer_pos..self.recv_buffer_pos + copy_length]);
                self.recv_buffer_pos += copy_length;
                return Ok(copy_length).into();
            }

            // Mutex doesn't have poll_lock, spinning on it.
            let mut kcp = self.session.kcp_socket().lock();

            // Try to read from KCP
            // 1. Read directly with user provided `buf`
            let peek_size = kcp.peek_size().unwrap_or(0);

            // 1.1. User's provided buffer is larger than available buffer's size
            if peek_size > 0 && peek_size <= buf.len() {
                match ready!(kcp.poll_recv(cx, buf)) {
                    Ok(n) => {
                        trace!("[CLIENT] recv directly {} bytes", n);
                        return Ok(n).into();
                    }
                    Err(KcpError::UserBufTooSmall) => {}
                    Err(err) => return Err(err).into(),
                }
            }

            // 2. User `buf` too small, read to recv_buffer
            let required_size = peek_size;
            if self.recv_buffer.len() < required_size {
                self.recv_buffer.resize(required_size, 0);
            }

            match ready!(kcp.poll_recv(cx, &mut self.recv_buffer)) {
                Ok(0) => return Ok(0).into(),
                Ok(n) => {
                    trace!("[CLIENT] recv buffered {} bytes", n);
                    self.recv_buffer_pos = 0;
                    self.recv_buffer_cap = n;
                }
                Err(err) => return Err(err).into(),
            }
        }
    }

I tried to debug the code and apprently the match on match ready!(kcp.poll_recv(cx, &mut self.recv_buffer)) does not go no any of the branch.
In fact, there is a loop there, and for whatever reason, the macro ready! return; i.e break the loop.
So what does that macro actually do, and what is the point of having that match branch with the ready! macro?
Thank you.

2

The documentation of futures::ready isn't very verbose, for sure! But what you described is exactly the expected behaviour in case the kcp future (or stream) is not ready yet when you poll it by calling kcp.poll_recv(...). The expanded macro (bound to a variable poll for clarity) looks like this:

pub fn poll_recv(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<KcpResult<usize>> {
        loop {
            // Consumes all data in buffer
            if self.recv_buffer_pos < self.recv_buffer_cap {
                let remaining = self.recv_buffer_cap - self.recv_buffer_pos;
                let copy_length = remaining.min(buf.len());

                buf[..copy_length]
                    .copy_from_slice(&self.recv_buffer[self.recv_buffer_pos..self.recv_buffer_pos + copy_length]);
                self.recv_buffer_pos += copy_length;
                return Ok(copy_length).into();
            }

            // Mutex doesn't have poll_lock, spinning on it.
            let mut kcp = self.session.kcp_socket().lock();

            // Try to read from KCP
            // 1. Read directly with user provided `buf`
            let peek_size = kcp.peek_size().unwrap_or(0);

            // 1.1. User's provided buffer is larger than available buffer's size
            if peek_size > 0 && peek_size <= buf.len() {
                match ready!(kcp.poll_recv(cx, buf)) {
                    Ok(n) => {
                        trace!("[CLIENT] recv directly {} bytes", n);
                        return Ok(n).into();
                    }
                    Err(KcpError::UserBufTooSmall) => {}
                    Err(err) => return Err(err).into(),
                }
            }

            // 2. User `buf` too small, read to recv_buffer
            let required_size = peek_size;
            if self.recv_buffer.len() < required_size {
                self.recv_buffer.resize(required_size, 0);
            }
            
            let poll: Result<_, _> = match kcp.poll_recv(cx, &mut self.recv_buffer) {
                Poll::Ready(r) => r,
                Poll::Pending => return Poll::Pending, // early return here if kcp isn't ready yet
            };

            match poll {
                Ok(0) => return Ok(0).into(),
                Ok(n) => {
                    trace!("[CLIENT] recv buffered {} bytes", n);
                    self.recv_buffer_pos = 0;
                    self.recv_buffer_cap = n;
                }
                Err(err) => return Err(err).into(),
            }
        }
    }

Hope that clarifies what is going on. You only enter your match statement if the future/stream you are polling produces a value. If the future/stream is not ready yet, you return early from your function with a Poll::Pending, telling the caller that you aren't ready to produce a value yet.

1 Like
3

exactly what I need to know, thank you.

1 Like
4

ready macro is now in standard library. It essentially expands to this:

match e {
    Poll::Ready(t) => t,
    Poll::Pending => return Poll::Pending,
}

This can be used to forward poll result from an external function. It's essentially like ? operator, but for Poll.

1 Like
5

This topic was automatically closed 90 days after the last reply. We invite you to open a new topic if you have further questions or comments.