diff --git a/src/sys/windows/named_pipe.rs b/src/sys/windows/named_pipe.rs index 2a6e9a239..3fe4796b9 100644 --- a/src/sys/windows/named_pipe.rs +++ b/src/sys/windows/named_pipe.rs @@ -3,7 +3,7 @@ use std::io::{self, Read, Write}; use std::os::windows::io::{AsRawHandle, FromRawHandle, RawHandle}; use std::sync::atomic::Ordering::{Relaxed, SeqCst}; use std::sync::atomic::{AtomicBool, AtomicUsize}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, MutexGuard}; use std::{fmt, mem, slice}; use windows_sys::Win32::Foundation::{ @@ -86,7 +86,9 @@ struct Inner { write: Overlapped, event: Overlapped, // END NOTE. - handle: Handle, + // `None` once the owning `NamedPipe` has been dropped and the handle + // closed; in-flight completions may still reference this `Inner`. + handle: Mutex>, connecting: AtomicBool, io: Mutex, pool: Mutex, @@ -100,6 +102,18 @@ unsafe impl Send for Inner {} // resources that are thread-safe in `Inner`. unsafe impl Sync for Inner {} +/// Borrow the handle from a locked `Inner::handle`, failing if it was already +/// closed by `NamedPipe::drop`. +fn check_handle<'a>(guard: &'a MutexGuard<'_, Option>) -> io::Result<&'a Handle> { + match &**guard { + Some(handle) => Ok(handle), + None => Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "named pipe handle is closed", + )), + } +} + impl Inner { /// Converts a pointer to `Inner.connect` to a pointer to `Inner`. /// @@ -149,7 +163,9 @@ impl Inner { /// valid until the I/O operation is completed, typically via completion /// ports and waiting to receive the completion notification on the port. pub unsafe fn connect_overlapped(&self, overlapped: *mut OVERLAPPED) -> io::Result { - if ConnectNamedPipe(self.handle.raw(), overlapped) != 0 { + let guard = self.handle.lock().unwrap(); + let handle = check_handle(&guard)?; + if ConnectNamedPipe(handle.raw(), overlapped) != 0 { return Ok(true); } @@ -165,7 +181,9 @@ impl Inner { /// Disconnects this named pipe from any connected client. pub fn disconnect(&self) -> io::Result<()> { - if unsafe { DisconnectNamedPipe(self.handle.raw()) } == 0 { + let guard = self.handle.lock().unwrap(); + let handle = check_handle(&guard)?; + if unsafe { DisconnectNamedPipe(handle.raw()) } == 0 { Err(io::Error::last_os_error()) } else { Ok(()) @@ -203,9 +221,11 @@ impl Inner { buf: &mut [u8], overlapped: *mut OVERLAPPED, ) -> io::Result> { + let guard = self.handle.lock().unwrap(); + let handle = check_handle(&guard)?; let len = std::cmp::min(buf.len(), u32::MAX as usize) as u32; let res = ReadFile( - self.handle.raw(), + handle.raw(), buf.as_mut_ptr() as *mut _, len, std::ptr::null_mut(), @@ -219,7 +239,7 @@ impl Inner { } let mut bytes = 0; - let res = GetOverlappedResult(self.handle.raw(), overlapped, &mut bytes, 0); + let res = GetOverlappedResult(handle.raw(), overlapped, &mut bytes, 0); if res == 0 { let err = io::Error::last_os_error(); if err.raw_os_error() == Some(ERROR_IO_INCOMPLETE as i32) { @@ -263,9 +283,11 @@ impl Inner { buf: &[u8], overlapped: *mut OVERLAPPED, ) -> io::Result> { + let guard = self.handle.lock().unwrap(); + let handle = check_handle(&guard)?; let len = std::cmp::min(buf.len(), u32::MAX as usize) as u32; let res = WriteFile( - self.handle.raw(), + handle.raw(), buf.as_ptr() as *const _, len, std::ptr::null_mut(), @@ -279,7 +301,7 @@ impl Inner { } let mut bytes = 0; - let res = GetOverlappedResult(self.handle.raw(), overlapped, &mut bytes, 0); + let res = GetOverlappedResult(handle.raw(), overlapped, &mut bytes, 0); if res == 0 { let err = io::Error::last_os_error(); if err.raw_os_error() == Some(ERROR_IO_INCOMPLETE as i32) { @@ -305,14 +327,20 @@ impl Inner { /// This function is unsafe as `overlapped` must have previously been used /// to execute an operation for this handle, and it must also be a valid /// pointer to an `Overlapped` instance. + /// + /// Returns `None` if the handle has already been closed by + /// `NamedPipe::drop`, in which case no result can be retrieved and the + /// completion should simply be discarded. #[inline] - unsafe fn result(&self, overlapped: *mut OVERLAPPED) -> io::Result { + unsafe fn result(&self, overlapped: *mut OVERLAPPED) -> Option> { + let guard = self.handle.lock().unwrap(); + let handle = check_handle(&guard).ok()?; let mut transferred = 0; - let r = GetOverlappedResult(self.handle.raw(), overlapped, &mut transferred, 0); + let r = GetOverlappedResult(handle.raw(), overlapped, &mut transferred, 0); if r == 0 { - Err(io::Error::last_os_error()) + Some(Err(io::Error::last_os_error())) } else { - Ok(transferred as usize) + Some(Ok(transferred as usize)) } } } @@ -494,7 +522,7 @@ impl FromRawHandle for NamedPipe { unsafe fn from_raw_handle(handle: RawHandle) -> NamedPipe { NamedPipe { inner: Arc::new(Inner { - handle: Handle::new(handle as HANDLE), + handle: Mutex::new(Some(Handle::new(handle as HANDLE))), connect: Overlapped::new(connect_done), connecting: AtomicBool::new(false), read: Overlapped::new(read_done), @@ -680,7 +708,10 @@ impl Source for NamedPipe { impl AsRawHandle for NamedPipe { fn as_raw_handle(&self) -> RawHandle { - self.inner.handle.raw() as RawHandle + let guard = self.inner.handle.lock().unwrap(); + // Cannot panic: the handle is only taken by `NamedPipe::drop`, and this + // `NamedPipe` is still alive. + guard.as_ref().unwrap().raw() as RawHandle } } @@ -692,18 +723,35 @@ impl fmt::Debug for NamedPipe { impl Drop for NamedPipe { fn drop(&mut self) { - // Cancel pending reads/connects, but don't cancel writes to ensure that - // everything is flushed out. + // NOTE: `io` must be locked before `handle`; every other code path + // takes the locks in that order. + let io = self.inner.io.lock().unwrap(); + let mut handle = self.inner.handle.lock().unwrap(); + // Cannot panic: the handle is only taken here, and this `NamedPipe` is + // still alive. + let raw_handle = handle.as_ref().unwrap(); + + // Cancel pending reads/connects. unsafe { if self.inner.connecting.load(SeqCst) { - drop(cancel(&self.inner.handle, &self.inner.connect)); + drop(cancel(raw_handle, &self.inner.connect)); } - let io = self.inner.io.lock().unwrap(); if let State::Pending(..) = io.read { - drop(cancel(&self.inner.handle, &self.inner.read)); + drop(cancel(raw_handle, &self.inner.read)); } } + + // Close the handle here rather than relying on the last `Arc` + // reference going away. A pending overlapped operation holds a + // reference that is only returned when its completion is witnessed, + // which is not guaranteed to happen (for example when `Poll` is dropped + // with operations still in flight). Leaving the close to that reference + // leaks the handle permanently. + // + // Closing the handle also ends any in-flight write, so a write is no + // longer guaranteed to be flushed out after the `NamedPipe` is dropped. + *handle = None; } } @@ -877,8 +925,11 @@ fn connect_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec>) { debug_assert_eq!(status.bytes_transferred(), 0); unsafe { match me.result(status.overlapped()) { - Ok(n) => debug_assert_eq!(n, 0), - Err(e) => me.io.lock().unwrap().connect_error = Some(e), + Some(Ok(n)) => debug_assert_eq!(n, 0), + Some(Err(e)) => me.io.lock().unwrap().connect_error = Some(e), + // The `NamedPipe` was dropped and the handle closed; there is + // nobody left to deliver an event to. + None => return, } } @@ -916,7 +967,7 @@ fn read_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec>) { }; unsafe { match me.result(status.overlapped()) { - Ok(n) => { + Some(Ok(n)) => { debug_assert_eq!(status.bytes_transferred() as usize, n); buf.set_len(status.bytes_transferred() as usize); io.read = State::Ok(buf, 0); @@ -924,14 +975,17 @@ fn read_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec>) { // 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) => { + Some(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) => { + Some(Err(e)) => { debug_assert_eq!(status.bytes_transferred(), 0); io.read = State::Err(e); } + // The `NamedPipe` was dropped and the handle closed; there is + // nobody left to deliver an event to. + None => return, } } @@ -971,7 +1025,7 @@ fn write_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec>) { unsafe { match me.result(status.overlapped()) { - Ok(n) => { + Some(Ok(n)) => { debug_assert_eq!(status.bytes_transferred() as usize, n); let new_pos = pos + (status.bytes_transferred() as usize); if new_pos == buf.len() { @@ -981,11 +1035,14 @@ fn write_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec>) { Inner::schedule_write(&me, buf, new_pos, &mut io, events); } } - Err(e) => { + Some(Err(e)) => { debug_assert_eq!(status.bytes_transferred(), 0); io.write = State::Err(e); io.notify_writable(&me, events); } + // The `NamedPipe` was dropped and the handle closed; there is + // nobody left to deliver an event to. + None => {} } } } diff --git a/tests/win_named_pipe.rs b/tests/win_named_pipe.rs index f4564a351..c546f822b 100644 --- a/tests/win_named_pipe.rs +++ b/tests/win_named_pipe.rs @@ -4,7 +4,7 @@ use std::fs::OpenOptions; use std::io::{self, Read, Write}; use std::os::windows::fs::OpenOptionsExt; use std::os::windows::io::{FromRawHandle, IntoRawHandle}; -use std::time::Duration; +use std::time::{Duration, Instant}; use mio::windows::NamedPipe; use mio::{Events, Interest, Poll, Token}; @@ -267,20 +267,22 @@ fn connect_twice() { let mut events = Events::with_capacity(128); - loop { - t!(poll.poll(&mut events, None)); - let events = events.iter().collect::>(); - if let Some(event) = events.iter().find(|e| e.token() == Token(0)) { - if event.is_readable() { - let mut buf = [0; 10]; - - match server.read(&mut buf) { - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue, - Ok(0) => break, - res => panic!("{:?}", res), + 'wait_for_eof: loop { + 'wait_for_readable: loop { + t!(poll.poll(&mut events, None)); + let events = events.iter().collect::>(); + for event in &events { + if event.is_readable() && event.token() == Token(0) { + break 'wait_for_readable; } } } + let mut buf = [0; 10]; + match server.read(&mut buf) { + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => continue 'wait_for_eof, + Ok(0) => break 'wait_for_eof, + res => panic!("{:?}", res), + } } t!(server.disconnect()); @@ -448,3 +450,73 @@ fn read_with_small_buffer_provided() { assert_eq!(actual_msg, expected_msg); } + +#[test] +fn handle_closed_on_drop() { + const TIMEOUT: Duration = Duration::from_secs(2); + + let (mut server, mut client) = pipe(); + let mut server_poll = t!(Poll::new()); + + t!(server_poll.registry().register( + &mut server, + Token(0), + Interest::READABLE | Interest::WRITABLE, + )); + t!(server.connect()); + + { + // Create another Poll as if we are in a separate process running a separate event loop. + let client_poll = t!(Poll::new()); + t!(client_poll.registry().register( + &mut client, + Token(1), + Interest::READABLE | Interest::WRITABLE, + )); + + let mut spam = b"spam".to_vec(); + spam.resize(1024 * 1024, 0); // 1MiB to make sure it blocks on server-side read + + // first write should not return WouldBlock + t!(client.write(&spam)); + // now there's an OVERLAPPED that will hold a ref to Arc indefinitely + + // order and presence of these 3 lines makes no difference: + let _ = client_poll.registry().deregister(&mut client); + drop(client); + drop(client_poll); + // Either way, client_poll will get dropped by the end of this block. + // Inside the drop, client_poll will make a (vain) attempt to drain the IOCP of all events + // and release all references to `client` named pipe. + // But the large write we just submited will not complete in this timeframe, and there will be + // one reference to `client.inner` that will never get released. + // + // Doing a large write is the reliable way to reproduce this, but writing thru `server` pipe in busy loop + // in a separate process and reading from `client` can also lead to leaked handles (race condition + // between CancelIoEx and GetQueuedCompletionStatusEx). + } + + // As server, read until eof. + // Since client's NamedPipe and even Poll got dropped, we should eventually get an EOF. + let mut events = Events::with_capacity(128); + let mut buf = vec![0; 1024]; + let start_time = Instant::now(); + 'wait_for_eof: loop { + t!(server_poll.poll(&mut events, Some(TIMEOUT))); + if events.is_empty() { + panic!( + "timed out waiting for eof after {}ms", + start_time.elapsed().as_millis() + ); + } + 'drain: loop { + match server.read(&mut buf) { + Err(e) if e.kind() == io::ErrorKind::WouldBlock => { + break 'drain; + } + Err(_) | Ok(0) => break 'wait_for_eof, + Ok(_) => (), + } + } + } +}