diff --git a/src/sys/windows/named_pipe.rs b/src/sys/windows/named_pipe.rs index 8d9d16efa..eeddcbdea 100644 --- a/src/sys/windows/named_pipe.rs +++ b/src/sys/windows/named_pipe.rs @@ -7,8 +7,8 @@ use std::sync::{Arc, Mutex}; use std::{fmt, mem, slice}; use windows_sys::Win32::Foundation::{ - ERROR_BROKEN_PIPE, ERROR_IO_INCOMPLETE, ERROR_IO_PENDING, ERROR_NO_DATA, ERROR_PIPE_CONNECTED, - ERROR_PIPE_LISTENING, HANDLE, INVALID_HANDLE_VALUE, + ERROR_BROKEN_PIPE, ERROR_IO_INCOMPLETE, ERROR_IO_PENDING, ERROR_MORE_DATA, ERROR_NO_DATA, + ERROR_PIPE_CONNECTED, ERROR_PIPE_LISTENING, HANDLE, INVALID_HANDLE_VALUE, }; use windows_sys::Win32::Storage::FileSystem::{ ReadFile, WriteFile, FILE_FLAG_FIRST_PIPE_INSTANCE, FILE_FLAG_OVERLAPPED, PIPE_ACCESS_DUPLEX, @@ -705,10 +705,13 @@ impl Inner { /// Schedules a read to happen in the background, executing an overlapped /// operation. /// - /// This function returns `true` if a normal error happens or if the read - /// is scheduled in the background. If the pipe is no longer connected - /// (ERROR_PIPE_LISTENING) then `false` is returned and no read is - /// scheduled. + /// This function returns `true` if either of the following conditions are met: + /// * A normal error happens + /// * The read is scheduled in the background + /// * Data is already available to be read (ERROR_MORE_DATA) + /// + /// If the pipe is no longer connected (ERROR_PIPE_LISTENING) then `false` is + /// returned and no read is scheduled. fn schedule_read(me: &Arc, io: &mut Io, events: Option<&mut Vec>) -> bool { // Check to see if a read is already scheduled/completed match io.read { @@ -736,6 +739,20 @@ impl Inner { // we just need to wait for a connect. Err(ref e) if e.raw_os_error() == Some(ERROR_PIPE_LISTENING as i32) => false, + // If ERROR_MORE_DATA is returned, it means the slice of unused capacity of the + // buffer provided is less than the amount of data available to be read. So + // prioritize draining the buffer before scheduling a new read. + // + // Return `true` to indicate that an overlapped read was scheduled "successfully", + // without actually scheduling it. Instead, update `io.read` to `State::Ok(buf, 0)` + // to ensure that the next `std::io::Read::read` call is presented still with the + // unread data to read from. + Err(ref e) if e.raw_os_error() == Some(ERROR_MORE_DATA as i32) => { + io.read = State::Ok(buf, 0); + mem::forget(me.clone()); + true + } + // If some other error happened, though, we're now readable to give // out the error. Err(e) => { @@ -898,6 +915,13 @@ fn read_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec>) { buf.set_len(status.bytes_transferred() as usize); io.read = State::Ok(buf, 0); } + // This is non-fatal. The buffer was simply too small for the entire message. + // Deliver the bytes we got, and if the caller wants to read the rest of the + // message, they can initiate another read. + Err(e) if e.raw_os_error() == Some(ERROR_MORE_DATA as i32) => { + buf.set_len(status.bytes_transferred() as usize); + io.read = State::Ok(buf, 0); + } Err(e) => { debug_assert_eq!(status.bytes_transferred(), 0); io.read = State::Err(e); diff --git a/tests/win_named_pipe.rs b/tests/win_named_pipe.rs index c495a6080..f4564a351 100644 --- a/tests/win_named_pipe.rs +++ b/tests/win_named_pipe.rs @@ -10,6 +10,9 @@ use mio::windows::NamedPipe; use mio::{Events, Interest, Poll, Token}; use windows_sys::Win32::{Foundation::ERROR_NO_DATA, Storage::FileSystem::FILE_FLAG_OVERLAPPED}; +mod util; +use util::{expect_events, ExpectEvent}; + fn _assert_kinds() { fn _assert_send() {} fn _assert_sync() {} @@ -347,3 +350,101 @@ fn reregister_deregister_different_poll() { io::ErrorKind::AlreadyExists, ); } + +#[test] +fn read_message_larger_than_internal_buffer() { + let (mut server, mut client) = pipe(); + let mut poll = t!(Poll::new()); + t!(poll.registry().register( + &mut server, + Token(0), + Interest::READABLE | Interest::WRITABLE, + )); + t!(poll.registry().register( + &mut client, + Token(1), + Interest::READABLE | Interest::WRITABLE, + )); + let mut events = Events::with_capacity(128); + t!(poll.poll(&mut events, None)); + + // Send message larger than the IPC kernel buffer (4096 bytes) + let expected_msg = vec![0x5u8; 8192]; + assert_eq!(t!(client.write(&expected_msg)), 8192); + + expect_events( + &mut poll, + &mut events, + vec![ExpectEvent::new(Token(0), Interest::READABLE)], + ); + + let mut buf = [0u8; 4000]; + let mut actual_msg = Vec::new(); + + loop { + match server.read(&mut buf) { + Ok(0) => break, + Ok(n) => { + actual_msg.extend_from_slice(&buf[..n]); + if actual_msg.len() >= expected_msg.len() { + break; + } + } + Err(e) if e.kind() == io::ErrorKind::WouldBlock => { + t!(poll.poll(&mut events, Some(Duration::from_secs(1)))); + } + Err(e) => panic!("error reading message: {e}"), + } + } + + assert_eq!(expected_msg, actual_msg); +} + +#[test] +fn read_with_small_buffer_provided() { + let (mut server, mut client) = pipe(); + let mut poll = t!(Poll::new()); + t!(poll.registry().register( + &mut server, + Token(0), + Interest::READABLE | Interest::WRITABLE, + )); + t!(poll.registry().register( + &mut client, + Token(1), + Interest::READABLE | Interest::WRITABLE, + )); + + let mut events = Events::with_capacity(128); + t!(poll.poll(&mut events, None)); + + let expected_msg = vec![1u8; 10000]; + assert_eq!(t!(client.write(&expected_msg)), 10000); + + expect_events( + &mut poll, + &mut events, + vec![ExpectEvent::new(Token(0), Interest::READABLE)], + ); + + let mut buf = [0u8; 128]; + let mut actual_msg = Vec::new(); + + loop { + match server.read(&mut buf) { + Ok(0) => break, + Ok(n) => { + actual_msg.extend_from_slice(&buf[..n]); + if actual_msg.len() >= expected_msg.len() { + break; + } + } + Err(e) if e.kind() == io::ErrorKind::WouldBlock => { + t!(poll.poll(&mut events, Some(Duration::from_millis(100)))); + } + Err(e) => panic!("error reading message: {e}"), + } + } + + assert_eq!(actual_msg, expected_msg); +}