pty: implement awaitable conpty for windows

A bit of a PITA, but it appears to be functional

pty/Cargo.toml

@@ -19,6 +19,7 @@ serde_derive = {version="1.0", optional=true}
 serde = {version="1.0", optional=true}
 serial = "0.4"
 ssh2 = {optional=true, version="0.6"}
+tokio = { version = "0.2", features = ["io-driver", "io-util"] }
 
 [features]
 default = []
@@ -27,7 +28,6 @@ ssh = ["ssh2"]
 
 [target."cfg(unix)".dependencies]
 mio = { version = "0.6" }
-tokio = { version = "0.2", features = ["io-driver", "io-util"] }
 
 [target."cfg(windows)".dependencies]
 bitflags = "1.0"

pty/examples/whoami_async.rs

@@ -39,7 +39,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         // implementation synthesizes title change escape sequences
         // in the output stream and it can be confusing to see those
         // printed out raw in another terminal.
-        print!("output: ");
+        print!("output: len={} ", line.len());
         for c in line.escape_debug() {
             print!("{}", c);
         }

pty/src/awaitable.rs

@@ -22,7 +22,7 @@ pub trait MasterPty: tokio::io::AsyncWrite {
 pub trait Child:
     std::fmt::Debug + std::future::Future<Output = anyhow::Result<ExitStatus>>
 {
-    /// Terminate the child process
+    /// Request termination of the child process
     fn kill(&mut self) -> IoResult<()>;
 }
 

pty/src/win/awaitable.rs

@@ -0,0 +1,414 @@
+use crate::cmdbuilder::CommandBuilder;
+use crate::win::conpty::ConPtySystem;
+use crate::win::psuedocon::PsuedoCon;
+use crate::win::readbuf::ReadBuffer;
+use crate::PtySize;
+use anyhow::anyhow;
+use async_trait::async_trait;
+use filedescriptor::{FileDescriptor, Pipe};
+use std::io::{self, Error as IoError, Read, Write};
+use std::os::windows::io::AsRawHandle;
+use std::os::windows::raw::HANDLE;
+use std::pin::Pin;
+use std::sync::mpsc::{channel, sync_channel, Receiver, Sender, SyncSender, TryRecvError};
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll, Waker};
+use winapi::um::namedpipeapi::PeekNamedPipe;
+use winapi::um::wincon::COORD;
+
+struct AwaitableConPtySlavePty {
+    inner: Arc<Mutex<AwaitableInner>>,
+}
+
+struct AwaitableConPtyMasterPty {
+    inner: Arc<Mutex<AwaitableInner>>,
+}
+
+enum WriteRequest {
+    Data(Vec<u8>),
+    Resize(PtySize, Waker, Sender<anyhow::Result<()>>),
+}
+
+struct AwaitableInner {
+    con: Arc<PsuedoCon>,
+    size: PtySize,
+    write_tx: Sender<WriteRequest>,
+    reader: Arc<Mutex<AwaitableReader>>,
+}
+
+/// PTYs on Windows are restricted to synchronous operation, so we cannot
+/// simply use IOCP to manage our asynchronous reads or writes.
+/// The AwaitableReader implements a little facade that allows us to
+/// schedule a blocking read of a desirable size in a worker thread.
+/// We can then use non-blocking calls on a channel to poll for completion.
+struct AwaitableReader {
+    wait_for_read: Sender<(Waker, usize)>,
+    read_buffer: ReadBuffer,
+    read_results_rx: Receiver<std::io::Result<Vec<u8>>>,
+}
+
+#[async_trait(?Send)]
+impl crate::awaitable::SlavePty for AwaitableConPtySlavePty {
+    async fn spawn_command(
+        &self,
+        cmd: CommandBuilder,
+    ) -> anyhow::Result<Pin<Box<dyn crate::awaitable::Child>>> {
+        let inner = self.inner.lock().unwrap();
+        let child = inner.con.spawn_command(cmd)?;
+        Ok(Box::pin(child))
+    }
+}
+
+impl tokio::io::AsyncRead for AwaitableConPtyMasterPty {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        self.inner
+            .lock()
+            .unwrap()
+            .reader
+            .lock()
+            .unwrap()
+            .poll_read_impl(cx, buf)
+    }
+}
+
+impl tokio::io::AsyncRead for AwaitableReaderArc {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
+        self.0.lock().unwrap().poll_read_impl(cx, buf)
+    }
+}
+
+impl tokio::io::AsyncWrite for AwaitableConPtyMasterPty {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        self.inner.lock().unwrap().poll_write_impl(cx, buf)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        _cx: &mut Context,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+}
+
+#[async_trait(?Send)]
+impl crate::awaitable::MasterPty for AwaitableConPtyMasterPty {
+    async fn resize(&self, size: PtySize) -> anyhow::Result<()> {
+        enum ResizeState {
+            NotRequested,
+            Waiting(Receiver<anyhow::Result<()>>),
+            Done,
+        }
+
+        struct ResizeFuture {
+            state: ResizeState,
+            inner: Arc<Mutex<AwaitableInner>>,
+            size: PtySize,
+        }
+
+        impl std::future::Future for ResizeFuture {
+            type Output = anyhow::Result<()>;
+
+            fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+                match std::mem::replace(&mut self.state, ResizeState::Done) {
+                    ResizeState::NotRequested => {
+                        let inner = self.inner.lock().unwrap();
+                        let (tx, rx) = channel();
+                        if let Err(err) = inner.write_tx.send(WriteRequest::Resize(
+                            self.size.clone(),
+                            cx.waker().clone(),
+                            tx,
+                        )) {
+                            return Poll::Ready(Err(anyhow!(
+                                "sending write request to pty failed: {}",
+                                err
+                            )));
+                        }
+
+                        drop(inner);
+                        self.state = ResizeState::Waiting(rx);
+                        Poll::Pending
+                    }
+                    ResizeState::Waiting(rx) => match rx.try_recv() {
+                        Ok(res) => {
+                            // We just successfully changed the size, so
+                            // record the new size
+                            self.inner.lock().unwrap().size = self.size;
+                            Poll::Ready(res)
+                        }
+                        Err(TryRecvError::Empty) => {
+                            self.state = ResizeState::Waiting(rx);
+                            Poll::Pending
+                        }
+                        Err(err) => Poll::Ready(Err(anyhow!(
+                            "receiving write results from pty failed: {}",
+                            err
+                        ))),
+                    },
+                    ResizeState::Done => Poll::Ready(Err(anyhow!("polling a completed future?"))),
+                }
+            }
+        }
+
+        let future = ResizeFuture {
+            state: ResizeState::NotRequested,
+            inner: Arc::clone(&self.inner),
+            size,
+        };
+        future.await
+    }
+
+    async fn get_size(&self) -> anyhow::Result<PtySize> {
+        let inner = self.inner.lock().unwrap();
+        Ok(inner.size.clone())
+    }
+
+    fn try_clone_reader(&self) -> anyhow::Result<Pin<Box<dyn tokio::io::AsyncRead + Send>>> {
+        let inner = self.inner.lock().unwrap();
+        Ok(Box::pin(AwaitableReaderArc(inner.reader.clone())))
+    }
+}
+
+#[async_trait(?Send)]
+impl crate::awaitable::PtySystem for ConPtySystem {
+    async fn openpty(&self, size: PtySize) -> anyhow::Result<crate::awaitable::PtyPair> {
+        let stdin = Pipe::new()?;
+        let stdout = Pipe::new()?;
+
+        let con = PsuedoCon::new(
+            COORD {
+                X: size.cols as i16,
+                Y: size.rows as i16,
+            },
+            stdin.read,
+            stdout.write,
+        )?;
+
+        let master = AwaitableConPtyMasterPty {
+            inner: Arc::new(Mutex::new(AwaitableInner::new(
+                con,
+                stdout.read,
+                stdin.write,
+                size,
+            ))),
+        };
+
+        let slave = AwaitableConPtySlavePty {
+            inner: master.inner.clone(),
+        };
+
+        Ok(crate::awaitable::PtyPair {
+            master: Box::pin(master),
+            slave: Box::pin(slave),
+        })
+    }
+}
+
+fn peek_pipe_len(pipe: HANDLE) -> std::io::Result<usize> {
+    let mut bytes_avail = 0;
+
+    let res = unsafe {
+        PeekNamedPipe(
+            pipe,
+            std::ptr::null_mut(),
+            0,
+            std::ptr::null_mut(),
+            &mut bytes_avail,
+            std::ptr::null_mut(),
+        )
+    };
+    if res == 0 {
+        return Err(IoError::last_os_error());
+    }
+
+    Ok(bytes_avail as usize)
+}
+
+impl AwaitableInner {
+    fn new(
+        con: PsuedoCon,
+        readable: FileDescriptor,
+        writable: FileDescriptor,
+        size: PtySize,
+    ) -> Self {
+        let con = Arc::new(con);
+        let (write_tx, write_rx) = channel();
+        let write_thread_con = Arc::clone(&con);
+        std::thread::spawn(move || Self::writer_thread(write_rx, write_thread_con, writable));
+
+        let (wait_for_read, wait_rx) = channel();
+
+        let (read_results_tx, read_results_rx) = sync_channel(1);
+        std::thread::spawn(move || {
+            AwaitableReader::reader_thread(wait_rx, readable, read_results_tx)
+        });
+
+        Self {
+            con,
+            size,
+            write_tx,
+            reader: Arc::new(Mutex::new(AwaitableReader {
+                wait_for_read,
+                read_results_rx,
+                read_buffer: ReadBuffer::new(),
+            })),
+        }
+    }
+
+    fn writer_thread(
+        to_write: Receiver<WriteRequest>,
+        con: Arc<PsuedoCon>,
+        mut writable: FileDescriptor,
+    ) {
+        while let Ok(item) = to_write.recv() {
+            match item {
+                WriteRequest::Data(data) => {
+                    if let Err(_err) = writable.write_all(&data) {
+                        // FIXME: set errored flag?
+                        // Right now we defer error detection to
+                        // the read side
+                    }
+                }
+                WriteRequest::Resize(size, waker, results) => {
+                    let res = con.resize(COORD {
+                        X: size.cols as i16,
+                        Y: size.rows as i16,
+                    });
+                    let res = results.send(res);
+                    waker.wake();
+                    if res.is_err() {
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    pub fn poll_write_impl(
+        &mut self,
+        _cx: &mut Context,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        // The poll_write API has EAGAIN semantics which are too
+        // awkward to emulate here.  We'll simply buffer up the
+        // write and claim that it succeeded.
+        let len = buf.len();
+        if let Err(err) = self.write_tx.send(WriteRequest::Data(buf.to_vec())) {
+            Poll::Ready(Err(std::io::Error::new(
+                std::io::ErrorKind::UnexpectedEof,
+                format!("unable to queue write ({}); treating as EOF", err),
+            )))
+        } else {
+            Poll::Ready(Ok(len))
+        }
+    }
+}
+
+impl AwaitableReader {
+    fn poll_read_impl(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>> {
+        // Try to satisfy from the buffer
+        let len = self.read_buffer.consume(buf);
+        if len > 0 {
+            return Poll::Ready(Ok(len));
+        }
+
+        match self.read_results_rx.try_recv() {
+            // A successful read; store to the buffer and then return
+            // an appropriate portion of it to our caller.
+            Ok(Ok(recv_buf)) => {
+                self.read_buffer.append(&recv_buf);
+
+                let len = self.read_buffer.consume(buf);
+                if len > 0 {
+                    return Poll::Ready(Ok(len));
+                }
+
+                // Probably impossible, but if we get here, we'll fall
+                // through below and request some more data to be read
+            }
+
+            // Map BrokenPipe errors to EOF for easier POSIX compatibility
+            Ok(Err(err)) if err.kind() == std::io::ErrorKind::BrokenPipe => {
+                return Poll::Ready(Ok(0))
+            }
+
+            // Other errors are returned to the caller
+            Ok(Err(err)) => return Poll::Ready(Err(err)),
+
+            Err(TryRecvError::Empty) => {
+                // There are no reasults ready to read yet.
+                // We'll queue up a read below.
+            }
+
+            // There's a problem with the channel, most likely we're partially
+            // destroyed: relay this as an EOF error (distinct from EOF) so
+            // that it bubbles up as an actual error condition
+            Err(err) => {
+                return Poll::Ready(Err(std::io::Error::new(
+                    std::io::ErrorKind::UnexpectedEof,
+                    format!("unable to receive read results ({}); treating as EOF", err),
+                )))
+            }
+        }
+
+        // Ask the reader thread to do some IO for us
+        match self.wait_for_read.send((cx.waker().clone(), buf.len())) {
+            Ok(_) => Poll::Pending,
+            Err(err) => Poll::Ready(Err(std::io::Error::new(
+                std::io::ErrorKind::UnexpectedEof,
+                format!("unable to queue read ({}); treating as EOF", err),
+            ))),
+        }
+    }
+
+    fn reader_thread(
+        wait_requests: Receiver<(Waker, usize)>,
+        mut readable: FileDescriptor,
+        read_results: SyncSender<std::io::Result<Vec<u8>>>,
+    ) {
+        while let Ok((waker, size)) = wait_requests.recv() {
+            // If there is data already present in the pipe, we know that
+            // our read will immediately succeed for that size, so prefer
+            // to size our local vector to match.
+            // If there is no data available to read, take the size from
+            // the caller.
+            let res = match peek_pipe_len(readable.as_raw_handle()) {
+                Ok(avail) => {
+                    let size = if avail == 0 { size } else { size.min(avail) };
+
+                    let mut buf = vec![0u8; size];
+
+                    readable.read(&mut buf).map(|size| {
+                        buf.resize(size, 0);
+                        buf
+                    })
+                }
+                Err(err) => Err(err),
+            };
+
+            let broken = read_results.send(res).is_err();
+            waker.wake();
+            if broken {
+                break;
+            }
+        }
+    }
+}
+
+#[derive(Clone)]
+struct AwaitableReaderArc(pub Arc<Mutex<AwaitableReader>>);

pty/src/win/conpty.rs

@@ -1,29 +1,12 @@
-use super::WinChild;
 use crate::cmdbuilder::CommandBuilder;
+use crate::win::psuedocon::PsuedoCon;
 use crate::{Child, MasterPty, PtyPair, PtySize, PtySystem, SlavePty};
-use anyhow::{bail, ensure, Error};
-use filedescriptor::{FileDescriptor, OwnedHandle, Pipe};
-use lazy_static::lazy_static;
-use shared_library::shared_library;
-use std::ffi::OsString;
-use std::io::{self, Error as IoError};
-use std::mem;
-use std::os::windows::ffi::OsStringExt;
-use std::os::windows::io::{AsRawHandle, FromRawHandle};
-use std::os::windows::raw::HANDLE;
-use std::path::Path;
-use std::ptr;
+use anyhow::Error;
+use filedescriptor::{FileDescriptor, Pipe};
+use std::io;
 use std::sync::{Arc, Mutex};
-use winapi::shared::minwindef::DWORD;
-use winapi::shared::winerror::{HRESULT, S_OK};
-use winapi::um::handleapi::*;
-use winapi::um::processthreadsapi::*;
-use winapi::um::winbase::STARTUPINFOEXW;
-use winapi::um::winbase::{CREATE_UNICODE_ENVIRONMENT, EXTENDED_STARTUPINFO_PRESENT};
 use winapi::um::wincon::COORD;
 
-const PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE: usize = 0x00020016;
-
 #[derive(Default)]
 pub struct ConPtySystem {}
 
@@ -37,8 +20,8 @@ impl PtySystem for ConPtySystem {
                 X: size.cols as i16,
                 Y: size.rows as i16,
             },
-            &stdin.read,
-            &stdout.write,
+            stdin.read,
+            stdout.write,
         )?;
 
         let master = ConPtyMasterPty {
@@ -61,132 +44,6 @@ impl PtySystem for ConPtySystem {
     }
 }
 
-struct ProcThreadAttributeList {
-    data: Vec<u8>,
-}
-
-impl ProcThreadAttributeList {
-    pub fn with_capacity(num_attributes: DWORD) -> Result<Self, Error> {
-        let mut bytes_required: usize = 0;
-        unsafe {
-            InitializeProcThreadAttributeList(
-                ptr::null_mut(),
-                num_attributes,
-                0,
-                &mut bytes_required,
-            )
-        };
-        let mut data = Vec::with_capacity(bytes_required);
-        // We have the right capacity, so force the vec to consider itself
-        // that length.  The contents of those bytes will be maintained
-        // by the win32 apis used in this impl.
-        unsafe { data.set_len(bytes_required) };
-
-        let attr_ptr = data.as_mut_slice().as_mut_ptr() as *mut _;
-        let res = unsafe {
-            InitializeProcThreadAttributeList(attr_ptr, num_attributes, 0, &mut bytes_required)
-        };
-        ensure!(
-            res != 0,
-            "InitializeProcThreadAttributeList failed: {}",
-            IoError::last_os_error()
-        );
-        Ok(Self { data })
-    }
-
-    pub fn as_mut_ptr(&mut self) -> LPPROC_THREAD_ATTRIBUTE_LIST {
-        self.data.as_mut_slice().as_mut_ptr() as *mut _
-    }
-
-    pub fn set_pty(&mut self, con: HPCON) -> Result<(), Error> {
-        let res = unsafe {
-            UpdateProcThreadAttribute(
-                self.as_mut_ptr(),
-                0,
-                PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE,
-                con,
-                mem::size_of::<HPCON>(),
-                ptr::null_mut(),
-                ptr::null_mut(),
-            )
-        };
-        ensure!(
-            res != 0,
-            "UpdateProcThreadAttribute failed: {}",
-            IoError::last_os_error()
-        );
-        Ok(())
-    }
-}
-
-impl Drop for ProcThreadAttributeList {
-    fn drop(&mut self) {
-        unsafe { DeleteProcThreadAttributeList(self.as_mut_ptr()) };
-    }
-}
-
-type HPCON = HANDLE;
-
-shared_library!(ConPtyFuncs,
-    pub fn CreatePseudoConsole(
-        size: COORD,
-        hInput: HANDLE,
-        hOutput: HANDLE,
-        flags: DWORD,
-        hpc: *mut HPCON
-    ) -> HRESULT,
-    pub fn ResizePseudoConsole(hpc: HPCON, size: COORD) -> HRESULT,
-    pub fn ClosePseudoConsole(hpc: HPCON),
-);
-
-lazy_static! {
-    static ref CONPTY: ConPtyFuncs = ConPtyFuncs::open(Path::new("kernel32.dll")).expect(
-        "this system does not support conpty.  Windows 10 October 2018 or newer is required"
-    );
-}
-
-struct PsuedoCon {
-    con: HPCON,
-}
-unsafe impl Send for PsuedoCon {}
-unsafe impl Sync for PsuedoCon {}
-impl Drop for PsuedoCon {
-    fn drop(&mut self) {
-        unsafe { (CONPTY.ClosePseudoConsole)(self.con) };
-    }
-}
-impl PsuedoCon {
-    fn new(size: COORD, input: &FileDescriptor, output: &FileDescriptor) -> Result<Self, Error> {
-        let mut con: HPCON = INVALID_HANDLE_VALUE;
-        let result = unsafe {
-            (CONPTY.CreatePseudoConsole)(
-                size,
-                input.as_raw_handle(),
-                output.as_raw_handle(),
-                0,
-                &mut con,
-            )
-        };
-        ensure!(
-            result == S_OK,
-            "failed to create psuedo console: HRESULT {}",
-            result
-        );
-        Ok(Self { con })
-    }
-    fn resize(&self, size: COORD) -> Result<(), Error> {
-        let result = unsafe { (CONPTY.ResizePseudoConsole)(self.con, size) };
-        ensure!(
-            result == S_OK,
-            "failed to resize console to {}x{}: HRESULT: {}",
-            size.X,
-            size.Y,
-            result
-        );
-        Ok(())
-    }
-}
-
 struct Inner {
     con: PsuedoCon,
     readable: FileDescriptor,
@@ -253,42 +110,7 @@ impl io::Write for ConPtyMasterPty {
 impl SlavePty for ConPtySlavePty {
     fn spawn_command(&self, cmd: CommandBuilder) -> anyhow::Result<Box<dyn Child>> {
         let inner = self.inner.lock().unwrap();
-
-        let mut si: STARTUPINFOEXW = unsafe { mem::zeroed() };
-        si.StartupInfo.cb = mem::size_of::<STARTUPINFOEXW>() as u32;
-
-        let mut attrs = ProcThreadAttributeList::with_capacity(1)?;
-        attrs.set_pty(inner.con.con)?;
-        si.lpAttributeList = attrs.as_mut_ptr();
-
-        let mut pi: PROCESS_INFORMATION = unsafe { mem::zeroed() };
-
-        let (mut exe, mut cmdline) = cmd.cmdline()?;
-        let cmd_os = OsString::from_wide(&cmdline);
-        let res = unsafe {
-            CreateProcessW(
-                exe.as_mut_slice().as_mut_ptr(),
-                cmdline.as_mut_slice().as_mut_ptr(),
-                ptr::null_mut(),
-                ptr::null_mut(),
-                0,
-                EXTENDED_STARTUPINFO_PRESENT | CREATE_UNICODE_ENVIRONMENT,
-                cmd.environment_block().as_mut_slice().as_mut_ptr() as *mut _,
-                ptr::null_mut(),
-                &mut si.StartupInfo,
-                &mut pi,
-            )
-        };
-        if res == 0 {
-            let err = IoError::last_os_error();
-            bail!("CreateProcessW `{:?}` failed: {}", cmd_os, err);
-        }
-
-        // Make sure we close out the thread handle so we don't leak it;
-        // we do this simply by making it owned
-        let _main_thread = unsafe { OwnedHandle::from_raw_handle(pi.hThread) };
-        let proc = unsafe { OwnedHandle::from_raw_handle(pi.hProcess) };
-
-        Ok(Box::new(WinChild { proc }))
+        let child = inner.con.spawn_command(cmd)?;
+        Ok(Box::new(child))
     }
 }

pty/src/win/mod.rs

@@ -1,13 +1,20 @@
 use crate::{Child, ExitStatus};
+use anyhow::Context as _;
 use std::io::{Error as IoError, Result as IoResult};
-use std::os::windows::io::AsRawHandle;
+use std::os::windows::io::{AsRawHandle, RawHandle};
+use std::pin::Pin;
+use std::task::{Context, Poll};
 use winapi::shared::minwindef::DWORD;
 use winapi::um::minwinbase::STILL_ACTIVE;
 use winapi::um::processthreadsapi::*;
 use winapi::um::synchapi::WaitForSingleObject;
 use winapi::um::winbase::INFINITE;
 
+mod awaitable;
 pub mod conpty;
+mod procthreadattr;
+mod psuedocon;
+mod readbuf;
 
 use filedescriptor::OwnedHandle;
 
@@ -16,8 +23,8 @@ pub struct WinChild {
     proc: OwnedHandle,
 }
 
-impl Child for WinChild {
-    fn try_wait(&mut self) -> IoResult<Option<ExitStatus>> {
+impl WinChild {
+    fn is_complete(&mut self) -> IoResult<Option<ExitStatus>> {
         let mut status: DWORD = 0;
         let res = unsafe { GetExitCodeProcess(self.proc.as_raw_handle(), &mut status) };
         if res != 0 {
@@ -31,10 +38,24 @@ impl Child for WinChild {
         }
     }
 
-    fn kill(&mut self) -> IoResult<()> {
-        unsafe {
-            TerminateProcess(self.proc.as_raw_handle(), 1);
+    fn do_kill(&mut self) -> IoResult<()> {
+        let res = unsafe { TerminateProcess(self.proc.as_raw_handle(), 1) };
+        let err = IoError::last_os_error();
+        if res != 0 {
+            Err(err)
+        } else {
+            Ok(())
         }
+    }
+}
+
+impl Child for WinChild {
+    fn try_wait(&mut self) -> IoResult<Option<ExitStatus>> {
+        self.is_complete()
+    }
+
+    fn kill(&mut self) -> IoResult<()> {
+        self.do_kill().ok();
         self.wait()?;
         Ok(())
     }
@@ -55,3 +76,34 @@ impl Child for WinChild {
         }
     }
 }
+
+impl crate::awaitable::Child for WinChild {
+    fn kill(&mut self) -> IoResult<()> {
+        self.do_kill()
+    }
+}
+
+impl std::future::Future for WinChild {
+    type Output = anyhow::Result<ExitStatus>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<anyhow::Result<ExitStatus>> {
+        match self.is_complete() {
+            Ok(Some(status)) => Poll::Ready(Ok(status)),
+            Err(err) => Poll::Ready(Err(err).context("Failed to retrieve process exit status")),
+            Ok(None) => {
+                struct PassRawHandleToWaiterThread(pub RawHandle);
+                unsafe impl Send for PassRawHandleToWaiterThread {}
+                let handle = PassRawHandleToWaiterThread(self.proc.as_raw_handle());
+
+                let waker = cx.waker().clone();
+                std::thread::spawn(move || {
+                    unsafe {
+                        WaitForSingleObject(handle.0, INFINITE);
+                    }
+                    waker.wake();
+                });
+                Poll::Pending
+            }
+        }
+    }
+}

pty/src/win/procthreadattr.rs

@@ -0,0 +1,73 @@
+use crate::win::psuedocon::HPCON;
+use anyhow::{ensure, Error};
+use std::io::Error as IoError;
+use std::mem;
+use std::ptr;
+use winapi::shared::minwindef::DWORD;
+use winapi::um::processthreadsapi::*;
+
+const PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE: usize = 0x00020016;
+
+pub struct ProcThreadAttributeList {
+    data: Vec<u8>,
+}
+
+impl ProcThreadAttributeList {
+    pub fn with_capacity(num_attributes: DWORD) -> Result<Self, Error> {
+        let mut bytes_required: usize = 0;
+        unsafe {
+            InitializeProcThreadAttributeList(
+                ptr::null_mut(),
+                num_attributes,
+                0,
+                &mut bytes_required,
+            )
+        };
+        let mut data = Vec::with_capacity(bytes_required);
+        // We have the right capacity, so force the vec to consider itself
+        // that length.  The contents of those bytes will be maintained
+        // by the win32 apis used in this impl.
+        unsafe { data.set_len(bytes_required) };
+
+        let attr_ptr = data.as_mut_slice().as_mut_ptr() as *mut _;
+        let res = unsafe {
+            InitializeProcThreadAttributeList(attr_ptr, num_attributes, 0, &mut bytes_required)
+        };
+        ensure!(
+            res != 0,
+            "InitializeProcThreadAttributeList failed: {}",
+            IoError::last_os_error()
+        );
+        Ok(Self { data })
+    }
+
+    pub fn as_mut_ptr(&mut self) -> LPPROC_THREAD_ATTRIBUTE_LIST {
+        self.data.as_mut_slice().as_mut_ptr() as *mut _
+    }
+
+    pub fn set_pty(&mut self, con: HPCON) -> Result<(), Error> {
+        let res = unsafe {
+            UpdateProcThreadAttribute(
+                self.as_mut_ptr(),
+                0,
+                PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE,
+                con,
+                mem::size_of::<HPCON>(),
+                ptr::null_mut(),
+                ptr::null_mut(),
+            )
+        };
+        ensure!(
+            res != 0,
+            "UpdateProcThreadAttribute failed: {}",
+            IoError::last_os_error()
+        );
+        Ok(())
+    }
+}
+
+impl Drop for ProcThreadAttributeList {
+    fn drop(&mut self) {
+        unsafe { DeleteProcThreadAttributeList(self.as_mut_ptr()) };
+    }
+}

pty/src/win/psuedocon.rs

@@ -0,0 +1,128 @@
+use super::WinChild;
+use crate::cmdbuilder::CommandBuilder;
+use crate::win::procthreadattr::ProcThreadAttributeList;
+use anyhow::{bail, ensure, Error};
+use filedescriptor::{FileDescriptor, OwnedHandle};
+use lazy_static::lazy_static;
+use shared_library::shared_library;
+use std::ffi::OsString;
+use std::io::Error as IoError;
+use std::mem;
+use std::os::windows::ffi::OsStringExt;
+use std::os::windows::io::{AsRawHandle, FromRawHandle};
+use std::os::windows::raw::HANDLE;
+use std::path::Path;
+use std::ptr;
+use winapi::shared::minwindef::DWORD;
+use winapi::shared::winerror::{HRESULT, S_OK};
+use winapi::um::handleapi::*;
+use winapi::um::processthreadsapi::*;
+use winapi::um::winbase::{
+    CREATE_UNICODE_ENVIRONMENT, EXTENDED_STARTUPINFO_PRESENT, STARTUPINFOEXW,
+};
+use winapi::um::wincon::COORD;
+
+pub type HPCON = HANDLE;
+
+shared_library!(ConPtyFuncs,
+    pub fn CreatePseudoConsole(
+        size: COORD,
+        hInput: HANDLE,
+        hOutput: HANDLE,
+        flags: DWORD,
+        hpc: *mut HPCON
+    ) -> HRESULT,
+    pub fn ResizePseudoConsole(hpc: HPCON, size: COORD) -> HRESULT,
+    pub fn ClosePseudoConsole(hpc: HPCON),
+);
+
+lazy_static! {
+    static ref CONPTY: ConPtyFuncs = ConPtyFuncs::open(Path::new("kernel32.dll")).expect(
+        "this system does not support conpty.  Windows 10 October 2018 or newer is required"
+    );
+}
+
+pub struct PsuedoCon {
+    con: HPCON,
+}
+
+unsafe impl Send for PsuedoCon {}
+unsafe impl Sync for PsuedoCon {}
+
+impl Drop for PsuedoCon {
+    fn drop(&mut self) {
+        unsafe { (CONPTY.ClosePseudoConsole)(self.con) };
+    }
+}
+
+impl PsuedoCon {
+    pub fn new(size: COORD, input: FileDescriptor, output: FileDescriptor) -> Result<Self, Error> {
+        let mut con: HPCON = INVALID_HANDLE_VALUE;
+        let result = unsafe {
+            (CONPTY.CreatePseudoConsole)(
+                size,
+                input.as_raw_handle(),
+                output.as_raw_handle(),
+                0,
+                &mut con,
+            )
+        };
+        ensure!(
+            result == S_OK,
+            "failed to create psuedo console: HRESULT {}",
+            result
+        );
+        Ok(Self { con })
+    }
+
+    pub fn resize(&self, size: COORD) -> Result<(), Error> {
+        let result = unsafe { (CONPTY.ResizePseudoConsole)(self.con, size) };
+        ensure!(
+            result == S_OK,
+            "failed to resize console to {}x{}: HRESULT: {}",
+            size.X,
+            size.Y,
+            result
+        );
+        Ok(())
+    }
+
+    pub fn spawn_command(&self, cmd: CommandBuilder) -> anyhow::Result<WinChild> {
+        let mut si: STARTUPINFOEXW = unsafe { mem::zeroed() };
+        si.StartupInfo.cb = mem::size_of::<STARTUPINFOEXW>() as u32;
+
+        let mut attrs = ProcThreadAttributeList::with_capacity(1)?;
+        attrs.set_pty(self.con)?;
+        si.lpAttributeList = attrs.as_mut_ptr();
+
+        let mut pi: PROCESS_INFORMATION = unsafe { mem::zeroed() };
+
+        let (mut exe, mut cmdline) = cmd.cmdline()?;
+        let cmd_os = OsString::from_wide(&cmdline);
+        let res = unsafe {
+            CreateProcessW(
+                exe.as_mut_slice().as_mut_ptr(),
+                cmdline.as_mut_slice().as_mut_ptr(),
+                ptr::null_mut(),
+                ptr::null_mut(),
+                0,
+                EXTENDED_STARTUPINFO_PRESENT | CREATE_UNICODE_ENVIRONMENT,
+                cmd.environment_block().as_mut_slice().as_mut_ptr() as *mut _,
+                ptr::null_mut(),
+                &mut si.StartupInfo,
+                &mut pi,
+            )
+        };
+        if res == 0 {
+            let err = IoError::last_os_error();
+            bail!("CreateProcessW `{:?}` failed: {}", cmd_os, err);
+        }
+
+        // Make sure we close out the thread handle so we don't leak it;
+        // we do this simply by making it owned
+        let _main_thread = unsafe { OwnedHandle::from_raw_handle(pi.hThread) };
+        let proc = unsafe { OwnedHandle::from_raw_handle(pi.hProcess) };
+
+        Ok(WinChild { proc })
+    }
+}

pty/src/win/readbuf.rs

@@ -0,0 +1,44 @@
+/// A simple read buffer
+pub struct ReadBuffer {
+    data: Vec<u8>,
+    /// The position to read data from
+    pos: usize,
+}
+
+impl ReadBuffer {
+    pub fn new() -> Self {
+        let data = Vec::with_capacity(8192);
+        Self { data, pos: 0 }
+    }
+
+    pub fn append(&mut self, buf: &[u8]) {
+        if self.data.len() + buf.len() > self.data.capacity() {
+            if self.pos == self.data.len() {
+                self.pos = 0;
+            } else if self.pos > 0 {
+                let (front, back) = self.data.split_at_mut(self.pos);
+                let len = back.len();
+                front[0..len].copy_from_slice(back);
+
+                self.pos = len;
+                self.data.resize(len, 0);
+            }
+        }
+        self.data.extend_from_slice(buf);
+    }
+
+    pub fn avail(&self) -> usize {
+        self.data.len() - self.pos
+    }
+
+    pub fn consume(&mut self, buf: &mut [u8]) -> usize {
+        let len = buf.len().min(self.avail());
+        if len == 0 {
+            0
+        } else {
+            buf[0..len].copy_from_slice(&self.data[self.pos..self.pos + len]);
+            self.pos += len;
+            len
+        }
+    }
+}