monocoque_core/
timeout.rs

1//! Timeout utilities for I/O operations
2//!
3//! Provides timeout wrappers for async read/write operations using compio's timeout support.
4
5use crate::rt::timeout;
6use compio_io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
7use std::io;
8use std::time::Duration;
9
10/// Execute an async `read_exact` operation with a timeout.
11///
12/// Reads exactly the full buffer or returns an error.
13pub async fn read_exact_with_timeout<S, B>(
14    stream: &mut S,
15    buf: B,
16    duration: Option<Duration>,
17) -> io::Result<compio_buf::BufResult<(), B>>
18where
19    S: AsyncRead + Unpin,
20    B: compio_buf::IoBufMut,
21{
22    match duration {
23        None => {
24            // No timeout, block indefinitely
25            Ok(stream.read_exact(buf).await)
26        }
27        Some(d) if d.is_zero() => {
28            // A zero budget cannot complete an exact read. These helpers serve
29            // the handshake, where that means the step is out of time; report a
30            // timeout rather than pretend a non-blocking read is in progress.
31            // (User-facing RCVTIMEO=0 non-blocking recv is handled in
32            // SocketBase::read_raw, which returns WouldBlock after draining any
33            // already-buffered frames - it never reaches this helper.)
34            Err(io::Error::new(
35                io::ErrorKind::TimedOut,
36                "Read operation timed out (zero timeout budget)",
37            ))
38        }
39        Some(d) => {
40            // Timeout mode
41            match timeout(d, stream.read_exact(buf)).await {
42                Ok(result) => Ok(result),
43                Err(_elapsed) => Err(io::Error::new(
44                    io::ErrorKind::TimedOut,
45                    "Read operation timed out",
46                )),
47            }
48        }
49    }
50}
51
52/// Execute an async `write_all` operation with a timeout.
53///
54/// Writes the entire buffer or returns an error.
55pub async fn write_all_with_timeout<S, B>(
56    stream: &mut S,
57    buf: B,
58    duration: Option<Duration>,
59) -> io::Result<compio_buf::BufResult<(), B>>
60where
61    S: AsyncWrite + Unpin,
62    B: compio_buf::IoBuf,
63{
64    match duration {
65        None => {
66            // No timeout, block indefinitely
67            Ok(stream.write_all(buf).await)
68        }
69        Some(d) if d.is_zero() => {
70            // A zero budget cannot complete a full write. See the read helper:
71            // in the handshake this means the step is out of time. User-facing
72            // SNDTIMEO=0 non-blocking send is handled in SocketBase, not here.
73            Err(io::Error::new(
74                io::ErrorKind::TimedOut,
75                "Write operation timed out (zero timeout budget)",
76            ))
77        }
78        Some(d) => {
79            // Timeout mode
80            match timeout(d, stream.write_all(buf)).await {
81                Ok(result) => Ok(result),
82                Err(_elapsed) => Err(io::Error::new(
83                    io::ErrorKind::TimedOut,
84                    "Write operation timed out",
85                )),
86            }
87        }
88    }
89}
90
91#[cfg(test)]
92mod tests {
93    use super::*;
94
95    // Note: These are compile-time tests to ensure the API is sound
96    // Full integration tests would require actual I/O operations
97
98    #[test]
99    fn test_timeout_types() {
100        let infinite: Option<Duration> = None;
101        assert!(infinite.is_none());
102        let nonblocking = Some(Duration::ZERO);
103        assert_eq!(nonblocking, Some(Duration::ZERO));
104        let timed = Some(Duration::from_secs(5));
105        assert_eq!(timed, Some(Duration::from_secs(5)));
106    }
107}
monocoque_core/timeout.rs

monocoque_core/
timeout.rs
