use crate::{poll, Registry};
use crate::event::Source;
use crate::sys::windows::{Event, Overlapped};
use winapi::um::minwinbase::OVERLAPPED_ENTRY;
use std::ffi::OsStr;
use std::fmt;
use std::io::{self, Read, Write};
use std::mem;
use std::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
use std::slice;
use std::sync::atomic::{AtomicUsize, AtomicBool};
use std::sync::atomic::Ordering::{Relaxed, SeqCst};
use std::sync::{Arc, Mutex};
use crate::{Interest, Token};
use miow::iocp::{CompletionPort, CompletionStatus};
use miow::pipe;
use winapi::shared::winerror::{ERROR_BROKEN_PIPE, ERROR_PIPE_LISTENING};
use winapi::um::ioapiset::CancelIoEx;
macro_rules! offset_of {
($t:ty, $($field:ident).+) => (
&(*(0 as *const $t)).$($field).+ as *const _ as usize
)
}
macro_rules! overlapped2arc {
($e:expr, $t:ty, $($field:ident).+) => ({
let offset = offset_of!($t, $($field).+);
debug_assert!(offset < mem::size_of::<$t>());
Arc::from_raw(($e as usize - offset) as *mut $t)
})
}
pub struct NamedPipe {
inner: Arc<Inner>,
}
#[repr(C)]
struct Inner {
handle: pipe::NamedPipe,
connect: Overlapped,
connecting: AtomicBool,
read: Overlapped,
write: Overlapped,
io: Mutex<Io>,
pool: Mutex<BufferPool>,
}
struct Io {
cp: Option<Arc<CompletionPort>>,
token: Option<Token>,
read: State,
read_interest: bool,
write: State,
write_interest: bool,
connect_error: Option<io::Error>,
}
#[derive(Debug)]
enum State {
None,
Pending(Vec<u8>, usize),
Ok(Vec<u8>, usize),
Err(io::Error),
}
static NEXT_TOKEN: AtomicUsize = AtomicUsize::new(1);
fn would_block() -> io::Error {
io::ErrorKind::WouldBlock.into()
}
impl NamedPipe {
pub fn new<A: AsRef<OsStr>>(
addr: A,
) -> io::Result<NamedPipe> {
let pipe = pipe::NamedPipe::new(addr)?;
Ok(unsafe { NamedPipe::from_raw_handle(pipe.into_raw_handle()) })
}
pub fn connect(&self) -> io::Result<()> {
if self.inner.connecting.swap(true, SeqCst) {
return Err(would_block());
}
let res = unsafe {
let overlapped = self.inner.connect.as_ptr() as *mut _;
self.inner.handle.connect_overlapped(overlapped)
};
match res {
Ok(true) => {
self.inner.connecting.store(false, SeqCst);
Inner::post_register(&self.inner, None);
Ok(())
}
Ok(false) => {
mem::forget(self.inner.clone());
Err(would_block())
}
Err(e) => {
self.inner.connecting.store(false, SeqCst);
Err(e)
}
}
}
pub fn take_error(&self) -> io::Result<Option<io::Error>> {
Ok(self.inner.io.lock().unwrap().connect_error.take())
}
pub fn disconnect(&self) -> io::Result<()> {
self.inner.handle.disconnect()
}
}
impl FromRawHandle for NamedPipe {
unsafe fn from_raw_handle(
handle: RawHandle,
) -> NamedPipe {
NamedPipe {
inner: Arc::new(Inner {
handle: pipe::NamedPipe::from_raw_handle(handle),
connect: Overlapped::new(connect_done),
connecting: AtomicBool::new(false),
read: Overlapped::new(read_done),
write: Overlapped::new(write_done),
io: Mutex::new(Io {
cp: None,
token: None,
read: State::None,
read_interest: false,
write: State::None,
write_interest: false,
connect_error: None,
}),
pool: Mutex::new(BufferPool::with_capacity(2)),
}),
}
}
}
impl Read for NamedPipe {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
<&NamedPipe as Read>::read(&mut &*self, buf)
}
}
impl Write for NamedPipe {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
<&NamedPipe as Write>::write(&mut &*self, buf)
}
fn flush(&mut self) -> io::Result<()> {
<&NamedPipe as Write>::flush(&mut &*self)
}
}
impl<'a> Read for &'a NamedPipe {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let mut state = self.inner.io.lock().unwrap();
if state.token.is_none() {
return Err(would_block());
}
match mem::replace(&mut state.read, State::None) {
State::None => {
Err(would_block())
}
State::Pending(buf, amt) => {
state.read = State::Pending(buf, amt);
Err(would_block())
}
State::Ok(data, cur) => {
let n = {
let mut remaining = &data[cur..];
remaining.read(buf)?
};
let next = cur + n;
if next != data.len() {
state.read = State::Ok(data, next);
} else {
self.inner.put_buffer(data);
Inner::schedule_read(&self.inner, &mut state, None);
}
Ok(n)
}
State::Err(e) => {
Inner::schedule_read(&self.inner, &mut state, None);
if e.raw_os_error() == Some(ERROR_BROKEN_PIPE as i32) {
Ok(0)
} else {
Err(e)
}
}
}
}
}
impl<'a> Write for &'a NamedPipe {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let mut io = self.inner.io.lock().unwrap();
if io.token.is_none() {
return Err(would_block());
}
match io.write {
State::None => {}
_ => {
return Err(would_block());
}
}
let mut owned_buf = self.inner.get_buffer();
owned_buf.extend(buf);
Inner::schedule_write(&self.inner, owned_buf, 0, &mut io, None);
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
impl Source for NamedPipe {
fn register(&mut self, registry: &Registry, token: Token, interest: Interest) -> io::Result<()> {
let mut io = self.inner.io.lock().unwrap();
io.check_association(registry, false)?;
if io.token.is_some() {
return Err(io::Error::new(
io::ErrorKind::AlreadyExists,
"I/O source already registered with a `Registry`",
));
}
if io.cp.is_none() {
io.cp = Some(poll::selector(registry).clone_port());
let inner_token = NEXT_TOKEN.fetch_add(2, Relaxed) + 2;
poll::selector(registry).inner.cp.add_handle(inner_token, &self.inner.handle)?;
}
io.token = Some(token);
io.read_interest = interest.is_readable();
io.write_interest = interest.is_writable();
drop(io);
Inner::post_register(&self.inner, None);
Ok(())
}
fn reregister(&mut self, registry: &Registry, token: Token, interest: Interest) -> io::Result<()> {
let mut io = self.inner.io.lock().unwrap();
io.check_association(registry, true)?;
io.token = Some(token);
io.read_interest = interest.is_readable();
io.write_interest = interest.is_writable();
drop(io);
Inner::post_register(&self.inner, None);
Ok(())
}
fn deregister(&mut self, registry: &Registry) -> io::Result<()> {
let mut io = self.inner.io.lock().unwrap();
io.check_association(registry, true)?;
if io.token.is_none() {
return Err(io::Error::new(
io::ErrorKind::NotFound,
"I/O source not registered with `Registry`",
));
}
io.token = None;
Ok(())
}
}
impl AsRawHandle for NamedPipe {
fn as_raw_handle(&self) -> RawHandle {
self.inner.handle.as_raw_handle()
}
}
impl fmt::Debug for NamedPipe {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.inner.handle.fmt(f)
}
}
impl Drop for NamedPipe {
fn drop(&mut self) {
unsafe {
if self.inner.connecting.load(SeqCst) {
drop(cancel(&self.inner.handle, &self.inner.connect));
}
let io = self.inner.io.lock().unwrap();
match io.read {
State::Pending(..) => {
drop(cancel(&self.inner.handle, &self.inner.read));
}
_ => {}
}
}
}
}
impl Inner {
fn schedule_read(me: &Arc<Inner>, io: &mut Io, events: Option<&mut Vec<Event>>) -> bool {
match io.read {
State::None => {}
_ => return true,
}
let mut buf = me.get_buffer();
let e = unsafe {
let overlapped = me.read.as_ptr() as *mut _;
let slice = slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.capacity());
me.handle.read_overlapped(slice, overlapped)
};
match e {
Ok(_) => {
io.read = State::Pending(buf, 0);
mem::forget(me.clone());
true
}
Err(ref e) if e.raw_os_error() == Some(ERROR_PIPE_LISTENING as i32) => false,
Err(e) => {
io.read = State::Err(e);
io.notify_readable(events);
true
}
}
}
fn schedule_write(me: &Arc<Inner>, buf: Vec<u8>, pos: usize, io: &mut Io, events: Option<&mut Vec<Event>>) {
let e = unsafe {
let overlapped = me.write.as_ptr() as *mut _;
me.handle.write_overlapped(&buf[pos..], overlapped)
};
match e {
Ok(_) => {
io.write = State::Pending(buf, pos);
mem::forget(me.clone())
}
Err(e) => {
io.write = State::Err(e);
io.notify_writable(events);
}
}
}
fn post_register(me: &Arc<Inner>, mut events: Option<&mut Vec<Event>>) {
let mut io = me.io.lock().unwrap();
if Inner::schedule_read(&me, &mut io, events.as_mut().map(|ptr| &mut **ptr)) {
if let State::None = io.write {
io.notify_writable(events);
}
}
}
fn get_buffer(&self) -> Vec<u8> {
self.pool.lock().unwrap().get(4 * 1024)
}
fn put_buffer(&self, buf: Vec<u8>) {
self.pool.lock().unwrap().put(buf)
}
}
unsafe fn cancel<T: AsRawHandle>(handle: &T, overlapped: &Overlapped) -> io::Result<()> {
let ret = CancelIoEx(handle.as_raw_handle(), overlapped.as_ptr() as *mut _);
if ret == 0 {
Err(io::Error::last_os_error())
} else {
Ok(())
}
}
fn connect_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec<Event>>) {
let status = CompletionStatus::from_entry(status);
let me = unsafe { overlapped2arc!(status.overlapped(), Inner, connect) };
let prev = me.connecting.swap(false, SeqCst);
assert!(prev, "NamedPipe was not previously connecting");
debug_assert_eq!(status.bytes_transferred(), 0);
unsafe {
match me.handle.result(status.overlapped()) {
Ok(n) => debug_assert_eq!(n, 0),
Err(e) => me.io.lock().unwrap().connect_error = Some(e),
}
}
Inner::post_register(&me, events);
}
fn read_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec<Event>>) {
let status = CompletionStatus::from_entry(status);
let me = unsafe { overlapped2arc!(status.overlapped(), Inner, read) };
let mut io = me.io.lock().unwrap();
let mut buf = match mem::replace(&mut io.read, State::None) {
State::Pending(buf, _) => buf,
_ => unreachable!(),
};
unsafe {
match me.handle.result(status.overlapped()) {
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);
}
Err(e) => {
debug_assert_eq!(status.bytes_transferred(), 0);
io.read = State::Err(e);
}
}
}
io.notify_readable(events);
}
fn write_done(status: &OVERLAPPED_ENTRY, events: Option<&mut Vec<Event>>) {
let status = CompletionStatus::from_entry(status);
let me = unsafe { overlapped2arc!(status.overlapped(), Inner, write) };
let mut io = me.io.lock().unwrap();
let (buf, pos) = match mem::replace(&mut io.write, State::None) {
State::Pending(buf, pos) => (buf, pos),
_ => unreachable!(),
};
unsafe {
match me.handle.result(status.overlapped()) {
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() {
me.put_buffer(buf);
io.notify_writable(events);
} else {
Inner::schedule_write(&me, buf, new_pos, &mut io, events);
}
}
Err(e) => {
debug_assert_eq!(status.bytes_transferred(), 0);
io.write = State::Err(e);
io.notify_writable(events);
}
}
}
}
impl Io {
fn check_association(&self, registry: &Registry, required: bool) -> io::Result<()> {
match self.cp {
Some(ref cp) if !poll::selector(registry).same_port(cp) => {
Err(io::Error::new(
io::ErrorKind::AlreadyExists,
"I/O source already registered with a different `Registry`"
))
}
None if required => {
Err(io::Error::new(
io::ErrorKind::NotFound,
"I/O source not registered with `Registry`"
))
}
_ => Ok(()),
}
}
fn notify_readable(&self, events: Option<&mut Vec<Event>>) {
if let Some(token) = self.token {
let mut ev = Event::new(token);
ev.set_readable();
if let Some(events) = events {
events.push(ev);
} else {
let _ = self.cp.as_ref().unwrap().post(ev.to_completion_status());
}
}
}
fn notify_writable(&self, events: Option<&mut Vec<Event>>) {
if let Some(token) = self.token {
let mut ev = Event::new(token);
ev.set_writable();
if let Some(events) = events {
events.push(ev);
} else {
let _ = self.cp.as_ref().unwrap().post(ev.to_completion_status());
}
}
}
}
struct BufferPool {
pool: Vec<Vec<u8>>,
}
impl BufferPool {
fn with_capacity(cap: usize) -> BufferPool {
BufferPool {
pool: Vec::with_capacity(cap),
}
}
fn get(&mut self, default_cap: usize) -> Vec<u8> {
self.pool
.pop()
.unwrap_or_else(|| Vec::with_capacity(default_cap))
}
fn put(&mut self, mut buf: Vec<u8>) {
if self.pool.len() < self.pool.capacity() {
unsafe {
buf.set_len(0);
}
self.pool.push(buf);
}
}
}