another rewrite

8 files changed, 492 insertions, 197 deletions

diff --git a/src/blocking/command.rs b/src/blocking/command.rs
index 6c4a825..9cacebe 100644
--- a/src/blocking/command.rs
+++ b/src/blocking/command.rs
@@ -117,14 +117,13 @@ impl Command {
     }
 
     /// Executes the command as a child process via
-    /// [`std::process::Command::spawn`], and attaches the given `pty` to
-    /// that child. The pty will be attached to all of `stdin`, `stdout`, and
-    /// `stderr` of the child, unless those file descriptors were previously
-    /// overridden through calls to [`stdin`](Self::stdin),
-    /// [`stdout`](Self::stdout), or [`stderr`](Self::stderr). The newly
-    /// created child process will also be made the session leader of a new
-    /// session, and will have the given `pty` instance set as its controlling
-    /// terminal.
+    /// [`std::process::Command::spawn`] on the given pty. The pty will be
+    /// attached to all of `stdin`, `stdout`, and `stderr` of the child,
+    /// unless those file descriptors were previously overridden through calls
+    /// to [`stdin`](Self::stdin), [`stdout`](Self::stdout), or
+    /// [`stderr`](Self::stderr). The newly created child process will also be
+    /// made the session leader of a new session, and will have the given
+    /// pty set as its controlling terminal.
     ///
     /// # Errors
     /// Returns an error if we fail to allocate new file descriptors for
@@ -134,10 +133,9 @@ impl Command {
     /// session leader or set its controlling terminal.
     pub fn spawn(
         &mut self,
-        pty: &crate::blocking::Pty,
+        pts: &crate::blocking::Pts,
     ) -> crate::Result<std::process::Child> {
-        let pts = pty.pts();
-        let (stdin, stdout, stderr) = crate::sys::setup_subprocess(pts)?;
+        let (stdin, stdout, stderr) = pts.0.setup_subprocess()?;
 
         if !self.stdin {
             self.inner.stdin(stdin);
@@ -149,7 +147,7 @@ impl Command {
             self.inner.stderr(stderr);
         }
 
-        let mut session_leader = crate::sys::session_leader(pts);
+        let mut session_leader = pts.0.session_leader();
         // Safety: setsid() is an async-signal-safe function and ioctl() is a
         // raw syscall (which is inherently async-signal-safe).
         if let Some(mut custom) = self.pre_exec.take() {
@@ -190,6 +188,7 @@ impl Command {
         self
     }
 
+    /// See [`std::os::unix::process::CommandExt::arg0`]
     pub fn arg0<S>(&mut self, arg: S) -> &mut Self
     where
         S: AsRef<std::ffi::OsStr>,
diff --git a/src/blocking/mod.rs b/src/blocking/mod.rs
index 17a0733..9f4e284 100644
--- a/src/blocking/mod.rs
+++ b/src/blocking/mod.rs
@@ -4,4 +4,4 @@
 mod command;
 pub use command::Command;
 mod pty;
-pub use pty::Pty;
+pub use pty::{Pts, Pty};
diff --git a/src/blocking/pty.rs b/src/blocking/pty.rs
index 6fa3382..e2c5bde 100644
--- a/src/blocking/pty.rs
+++ b/src/blocking/pty.rs
@@ -1,8 +1,5 @@
 /// An allocated pty
-pub struct Pty {
-    pt: std::fs::File,
-    pts: std::fs::File,
-}
+pub struct Pty(crate::sys::Pty);
 
 impl Pty {
     /// Allocate and return a new pty.
@@ -10,12 +7,7 @@ impl Pty {
     /// # Errors
     /// Returns an error if the pty failed to be allocated.
     pub fn new() -> crate::Result<Self> {
-        let (pt, ptsname) = crate::sys::create_pt()?;
-        let pts = std::fs::OpenOptions::new()
-            .read(true)
-            .write(true)
-            .open(&ptsname)?;
-        Ok(Self { pt, pts })
+        Ok(Self(crate::sys::Pty::open()?))
     }
 
     /// Change the terminal size associated with the pty.
@@ -23,51 +15,60 @@ impl Pty {
     /// # Errors
     /// Returns an error if we were unable to set the terminal size.
     pub fn resize(&self, size: crate::Size) -> crate::Result<()> {
-        Ok(crate::sys::set_term_size(self, size)?)
+        self.0.set_term_size(size)
     }
 
-    pub(crate) fn pts(&self) -> &std::fs::File {
-        &self.pts
+    /// Opens a file descriptor for the other end of the pty, which should be
+    /// attached to the child process running in it. See
+    /// [`Command::spawn`](crate::blocking::Command::spawn).
+    ///
+    /// # Errors
+    /// Returns an error if the device node to open could not be determined,
+    /// or if the device node could not be opened.
+    pub fn pts(&self) -> crate::Result<Pts> {
+        Ok(Pts(self.0.pts()?))
     }
 }
 
-impl std::ops::Deref for Pty {
-    type Target = std::fs::File;
-
-    fn deref(&self) -> &Self::Target {
-        &self.pt
+impl std::os::unix::io::AsRawFd for Pty {
+    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
+        self.0 .0.as_raw_fd()
     }
 }
 
-impl std::ops::DerefMut for Pty {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.pt
+impl std::io::Read for Pty {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        self.0 .0.read(buf)
     }
 }
 
-impl std::os::unix::io::AsRawFd for Pty {
-    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
-        self.pt.as_raw_fd()
+impl std::io::Write for Pty {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        self.0 .0.write(buf)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.0 .0.flush()
     }
 }
 
-// there is a Read impl for &std::fs::File, but without this explicit impl,
-// rust finds the Read impl for std::fs::File first, and then complains that
-// it requires &mut self, because method resolution/autoderef doesn't take
-// mutability into account
 impl std::io::Read for &Pty {
     fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
-        (&self.pt).read(buf)
+        (&self.0 .0).read(buf)
     }
 }
 
-// same as above
 impl std::io::Write for &Pty {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        (&self.pt).write(buf)
+        (&self.0 .0).write(buf)
     }
 
     fn flush(&mut self) -> std::io::Result<()> {
-        (&self.pt).flush()
+        (&self.0 .0).flush()
     }
 }
+
+/// The child end of the pty
+///
+/// See [`Pty::pts`] and [`Command::spawn`](crate::blocking::Command::spawn)
+pub struct Pts(pub(crate) crate::sys::Pts);
diff --git a/src/command.rs b/src/command.rs
index 8627a35..7ff9858 100644
--- a/src/command.rs
+++ b/src/command.rs
@@ -1,8 +1,6 @@
-use async_process::unix::CommandExt as _;
-
-/// Wrapper around [`async_process::Command`]
+/// Wrapper around [`tokio::process::Command`]
 pub struct Command {
-    inner: async_process::Command,
+    inner: tokio::process::Command,
     stdin: bool,
     stdout: bool,
     stderr: bool,
@@ -13,10 +11,10 @@ pub struct Command {
 }
 
 impl Command {
-    /// See [`async_process::Command::new`]
+    /// See [`tokio::process::Command::new`]
     pub fn new<S: AsRef<std::ffi::OsStr>>(program: S) -> Self {
         Self {
-            inner: async_process::Command::new(program),
+            inner: tokio::process::Command::new(program),
             stdin: false,
             stdout: false,
             stderr: false,
@@ -25,13 +23,13 @@ impl Command {
         }
     }
 
-    /// See [`async_process::Command::arg`]
+    /// See [`tokio::process::Command::arg`]
     pub fn arg<S: AsRef<std::ffi::OsStr>>(&mut self, arg: S) -> &mut Self {
         self.inner.arg(arg);
         self
     }
 
-    /// See [`async_process::Command::args`]
+    /// See [`tokio::process::Command::args`]
     pub fn args<I, S>(&mut self, args: I) -> &mut Self
     where
         I: IntoIterator<Item = S>,
@@ -41,7 +39,7 @@ impl Command {
         self
     }
 
-    /// See [`async_process::Command::env`]
+    /// See [`tokio::process::Command::env`]
     pub fn env<K, V>(&mut self, key: K, val: V) -> &mut Self
     where
         K: AsRef<std::ffi::OsStr>,
@@ -51,7 +49,7 @@ impl Command {
         self
     }
 
-    /// See [`async_process::Command::envs`]
+    /// See [`tokio::process::Command::envs`]
     pub fn envs<I, K, V>(&mut self, vars: I) -> &mut Self
     where
         I: IntoIterator<Item = (K, V)>,
@@ -62,7 +60,7 @@ impl Command {
         self
     }
 
-    /// See [`async_process::Command::env_remove`]
+    /// See [`tokio::process::Command::env_remove`]
     pub fn env_remove<K: AsRef<std::ffi::OsStr>>(
         &mut self,
         key: K,
@@ -71,13 +69,13 @@ impl Command {
         self
     }
 
-    /// See [`async_process::Command::env_clear`]
+    /// See [`tokio::process::Command::env_clear`]
     pub fn env_clear(&mut self) -> &mut Self {
         self.inner.env_clear();
         self
     }
 
-    /// See [`async_process::Command::current_dir`]
+    /// See [`tokio::process::Command::current_dir`]
     pub fn current_dir<P: AsRef<std::path::Path>>(
         &mut self,
         dir: P,
@@ -86,7 +84,7 @@ impl Command {
         self
     }
 
-    /// See [`async_process::Command::stdin`]
+    /// See [`tokio::process::Command::stdin`]
     pub fn stdin<T: Into<std::process::Stdio>>(
         &mut self,
         cfg: T,
@@ -96,7 +94,7 @@ impl Command {
         self
     }
 
-    /// See [`async_process::Command::stdout`]
+    /// See [`tokio::process::Command::stdout`]
     pub fn stdout<T: Into<std::process::Stdio>>(
         &mut self,
         cfg: T,
@@ -106,7 +104,7 @@ impl Command {
         self
     }
 
-    /// See [`async_process::Command::stderr`]
+    /// See [`tokio::process::Command::stderr`]
     pub fn stderr<T: Into<std::process::Stdio>>(
         &mut self,
         cfg: T,
@@ -117,27 +115,25 @@ impl Command {
     }
 
     /// Executes the command as a child process via
-    /// [`async_process::Command::spawn`], and attaches the given `pty` to
-    /// that child. The pty will be attached to all of `stdin`, `stdout`, and
-    /// `stderr` of the child, unless those file descriptors were previously
-    /// overridden through calls to [`stdin`](Self::stdin),
-    /// [`stdout`](Self::stdout), or [`stderr`](Self::stderr). The newly
-    /// created child process will also be made the session leader of a new
-    /// session, and will have the given `pty` instance set as its controlling
-    /// terminal.
+    /// [`tokio::process::Command::spawn`] on the given pty. The pty will be
+    /// attached to all of `stdin`, `stdout`, and `stderr` of the child,
+    /// unless those file descriptors were previously overridden through calls
+    /// to [`stdin`](Self::stdin), [`stdout`](Self::stdout), or
+    /// [`stderr`](Self::stderr). The newly created child process will also be
+    /// made the session leader of a new session, and will have the given
+    /// pty set as its controlling terminal.
     ///
     /// # Errors
     /// Returns an error if we fail to allocate new file descriptors for
     /// attaching the pty to the child process, or if we fail to spawn the
     /// child process (see the documentation for
-    /// [`async_process::Command::spawn`]), or if we fail to make the child a
+    /// [`tokio::process::Command::spawn`]), or if we fail to make the child a
     /// session leader or set its controlling terminal.
     pub fn spawn(
         &mut self,
-        pty: &crate::Pty,
-    ) -> crate::Result<async_process::Child> {
-        let pts = pty.pts();
-        let (stdin, stdout, stderr) = crate::sys::setup_subprocess(pts)?;
+        pts: &crate::Pts,
+    ) -> crate::Result<tokio::process::Child> {
+        let (stdin, stdout, stderr) = pts.0.setup_subprocess()?;
 
         if !self.stdin {
             self.inner.stdin(stdin);
@@ -149,7 +145,7 @@ impl Command {
             self.inner.stderr(stderr);
         }
 
-        let mut session_leader = crate::sys::session_leader(pts);
+        let mut session_leader = pts.0.session_leader();
         // Safety: setsid() is an async-signal-safe function and ioctl() is a
         // raw syscall (which is inherently async-signal-safe).
         if let Some(mut custom) = self.pre_exec.take() {
@@ -168,19 +164,19 @@ impl Command {
         Ok(self.inner.spawn()?)
     }
 
-    /// See [`async_process::unix::CommandExt::uid`]
+    /// See [`tokio::process::Command::uid`]
     pub fn uid(&mut self, id: u32) -> &mut Self {
         self.inner.uid(id);
         self
     }
 
-    /// See [`async_process::unix::CommandExt::gid`]
+    /// See [`tokio::process::Command::gid`]
     pub fn gid(&mut self, id: u32) -> &mut Self {
         self.inner.gid(id);
         self
     }
 
-    /// See [`async_process::unix::CommandExt::pre_exec`]
+    /// See [`tokio::process::Command::pre_exec`]
     #[allow(clippy::missing_safety_doc)]
     pub unsafe fn pre_exec<F>(&mut self, f: F) -> &mut Self
     where
@@ -190,7 +186,7 @@ impl Command {
         self
     }
 
-    /// See [`async_process::unix::CommandExt::arg0`]
+    /// See [`tokio::process::Command::arg0`]
     pub fn arg0<S>(&mut self, arg: S) -> &mut Self
     where
         S: AsRef<std::ffi::OsStr>,
diff --git a/src/error.rs b/src/error.rs
index 9d4c31e..db9c647 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -5,6 +5,9 @@ pub enum Error {
     Io(std::io::Error),
     /// error came from nix::Error
     Nix(nix::Error),
+    /// unsplit was called on halves of two different ptys
+    #[cfg(feature = "async")]
+    Unsplit(crate::OwnedReadPty, crate::OwnedWritePty),
 }
 
 impl std::fmt::Display for Error {
@@ -12,6 +15,10 @@ impl std::fmt::Display for Error {
         match self {
             Self::Io(e) => write!(f, "{}", e),
             Self::Nix(e) => write!(f, "{}", e),
+            #[cfg(feature = "async")]
+            Self::Unsplit(..) => {
+                write!(f, "unsplit called on halves of two different ptys")
+            }
         }
     }
 }
@@ -33,6 +40,8 @@ impl std::error::Error for Error {
         match self {
             Self::Io(e) => Some(e),
             Self::Nix(e) => Some(e),
+            #[cfg(feature = "async")]
+            Self::Unsplit(..) => None,
         }
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index a4bb412..ce8dc9e 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,5 +1,5 @@
-//! This crate is a wrapper around [`std::process::Command`] or
-//! [`async_process::Command`] which provides the ability to allocate a pty
+//! This crate is a wrapper around [`tokio::process::Command`] or
+//! [`std::process::Command`] which provides the ability to allocate a pty
 //! and spawn new processes attached to that pty, with the pty as their
 //! controlling terminal. This allows for manipulation of interactive
 //! programs.
@@ -10,25 +10,23 @@
 //! let mut pty = pty_process::Pty::new().unwrap();
 //! pty.resize(pty_process::Size::new(24, 80)).unwrap();
 //! let mut cmd = pty_process::Command::new("nethack");
-//! let child = cmd.spawn(&pty).unwrap();
+//! let child = cmd.spawn(&pty.pts().unwrap()).unwrap();
 //! ```
 //!
-//! The returned `child` is a normal instance of [`async_process::Child`] (or
+//! The returned `child` is a normal instance of [`tokio::process::Child`] (or
 //! [`std::process::Child`] for the [`blocking`](crate::blocking) variant),
 //! with its `stdin`/`stdout`/`stderr` file descriptors pointing at the given
-//! pty. The `pty` instance implements [`std::io::Read`] and
-//! [`std::io::Write`] (or [`futures_io::AsyncRead`] and
-//! [`futures_io::AsyncWrite`] for the [`blocking`] variant), and can be used
-//! to communicate with the child process. The child process will also be made
-//! a session leader of a new session, and the controlling terminal of that
-//! session will be set to the given pty.
+//! pty. The `pty` instance implements [`tokio::io::AsyncRead`] and
+//! [`tokio::io::AsyncWrite`] (or [`std::io::Read`] and [`std::io::Write`] for
+//! the [`blocking`] variant), and can be used to communicate with the child
+//! process. The child process will also be made a session leader of a new
+//! session, and the controlling terminal of that session will be set to the
+//! given pty.
 //!
 //! # Features
 //!
 //! By default, only the [`blocking`](crate::blocking) APIs are available. To
-//! include the asynchronous APIs, you must enable the `async` feature. See
-//! the `examples` directory in the repository for examples of how to use this
-//! crate with the various different asynchronous frameworks.
+//! include the asynchronous APIs, you must enable the `async` feature.
 
 #![warn(clippy::cargo)]
 #![warn(clippy::pedantic)]
@@ -59,4 +57,4 @@ pub use command::Command;
 #[cfg(feature = "async")]
 mod pty;
 #[cfg(feature = "async")]
-pub use pty::Pty;
+pub use pty::{OwnedReadPty, OwnedWritePty, Pts, Pty, ReadPty, WritePty};
diff --git a/src/pty.rs b/src/pty.rs
index 1d7f113..87fb311 100644
--- a/src/pty.rs
+++ b/src/pty.rs
@@ -1,8 +1,11 @@
+#![allow(clippy::module_name_repetitions)]
+
+use std::io::{Read as _, Write as _};
+
+type AsyncPty = tokio::io::unix::AsyncFd<crate::sys::Pty>;
+
 /// An allocated pty
-pub struct Pty {
-    pt: async_io::Async<std::fs::File>,
-    pts: std::fs::File,
-}
+pub struct Pty(AsyncPty);
 
 impl Pty {
     /// Allocate and return a new pty.
@@ -11,13 +14,9 @@ impl Pty {
     /// Returns an error if the pty failed to be allocated, or if we were
     /// unable to put it into non-blocking mode.
     pub fn new() -> crate::Result<Self> {
-        let (pt, ptsname) = crate::sys::create_pt()?;
-        let pt = async_io::Async::new(pt)?;
-        let pts = std::fs::OpenOptions::new()
-            .read(true)
-            .write(true)
-            .open(&ptsname)?;
-        Ok(Self { pt, pts })
+        let pty = crate::sys::Pty::open()?;
+        pty.set_nonblocking()?;
+        Ok(Self(tokio::io::unix::AsyncFd::new(pty)?))
     }
 
     /// Change the terminal size associated with the pty.
@@ -25,69 +24,318 @@ impl Pty {
     /// # Errors
     /// Returns an error if we were unable to set the terminal size.
     pub fn resize(&self, size: crate::Size) -> crate::Result<()> {
-        Ok(crate::sys::set_term_size(self, size)?)
+        self.0.get_ref().set_term_size(size)
     }
 
-    pub(crate) fn pts(&self) -> &std::fs::File {
-        &self.pts
+    /// Opens a file descriptor for the other end of the pty, which should be
+    /// attached to the child process running in it. See
+    /// [`Command::spawn`](crate::Command::spawn).
+    ///
+    /// # Errors
+    /// Returns an error if the device node to open could not be determined,
+    /// or if the device node could not be opened.
+    pub fn pts(&self) -> crate::Result<Pts> {
+        Ok(Pts(self.0.get_ref().pts()?))
     }
-}
-
-impl std::ops::Deref for Pty {
-    type Target = async_io::Async<std::fs::File>;
 
-    fn deref(&self) -> &Self::Target {
-        &self.pt
+    /// Splits a `Pty` into a read half and a write half, which can be used to
+    /// read from and write to the pty concurrently. Does not allocate, but
+    /// the returned halves cannot be moved to independent tasks.
+    pub fn split(&mut self) -> (ReadPty<'_>, WritePty<'_>) {
+        (ReadPty(&self.0), WritePty(&self.0))
     }
-}
 
-impl std::ops::DerefMut for Pty {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.pt
+    /// Splits a `Pty` into a read half and a write half, which can be used to
+    /// read from and write to the pty concurrently. This method requires an
+    /// allocation, but the returned halves can be moved to independent tasks.
+    /// The original `Pty` instance can be recovered via the
+    /// [`OwnedReadPty::unsplit`] method.
+    #[must_use]
+    pub fn into_split(self) -> (OwnedReadPty, OwnedWritePty) {
+        let Self(pt) = self;
+        let read_pt = std::sync::Arc::new(pt);
+        let write_pt = std::sync::Arc::clone(&read_pt);
+        (OwnedReadPty(read_pt), OwnedWritePty(write_pt))
     }
 }
 
 impl std::os::unix::io::AsRawFd for Pty {
     fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
-        self.pt.as_raw_fd()
+        self.0.as_raw_fd()
     }
 }
 
-// there is an AsyncRead impl for &Async<std::fs::File>, but without this
-// explicit impl, rust finds the AsyncRead impl for Async<std::fs::File>
-// first, and then complains that it requires &mut self, because method
-// resolution/autoderef doesn't take mutability into account
-impl futures_io::AsyncRead for &Pty {
+impl tokio::io::AsyncRead for Pty {
     fn poll_read(
         self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
-        buf: &mut [u8],
-    ) -> std::task::Poll<Result<usize, std::io::Error>> {
-        std::pin::Pin::new(&mut &self.pt).poll_read(cx, buf)
+        buf: &mut tokio::io::ReadBuf,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        loop {
+            let mut guard = match self.0.poll_read_ready(cx) {
+                std::task::Poll::Ready(guard) => guard,
+                std::task::Poll::Pending => return std::task::Poll::Pending,
+            }?;
+            let mut b = [0u8; 4096];
+            match guard.try_io(|inner| (&inner.get_ref().0).read(&mut b)) {
+                Ok(Ok(bytes)) => {
+                    // XXX this is safe, but not particularly efficient
+                    buf.clear();
+                    buf.initialize_unfilled_to(bytes);
+                    buf.set_filled(bytes);
+                    buf.filled_mut().copy_from_slice(&b[..bytes]);
+                    return std::task::Poll::Ready(Ok(()));
+                }
+                Ok(Err(e)) => return std::task::Poll::Ready(Err(e)),
+                Err(_would_block) => continue,
+            }
+        }
     }
 }
 
-// same as above
-impl futures_io::AsyncWrite for &Pty {
+impl tokio::io::AsyncWrite for Pty {
     fn poll_write(
         self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
         buf: &[u8],
-    ) -> std::task::Poll<Result<usize, std::io::Error>> {
-        std::pin::Pin::new(&mut &self.pt).poll_write(cx, buf)
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        loop {
+            let mut guard = match self.0.poll_write_ready(cx) {
+                std::task::Poll::Ready(guard) => guard,
+                std::task::Poll::Pending => return std::task::Poll::Pending,
+            }?;
+            match guard.try_io(|inner| (&inner.get_ref().0).write(buf)) {
+                Ok(result) => return std::task::Poll::Ready(result),
+                Err(_would_block) => continue,
+            }
+        }
     }
 
     fn poll_flush(
         self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        loop {
+            let mut guard = match self.0.poll_write_ready(cx) {
+                std::task::Poll::Ready(guard) => guard,
+                std::task::Poll::Pending => return std::task::Poll::Pending,
+            }?;
+            match guard.try_io(|inner| (&inner.get_ref().0).flush()) {
+                Ok(_) => return std::task::Poll::Ready(Ok(())),
+                Err(_would_block) => continue,
+            }
+        }
+    }
+
+    fn poll_shutdown(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Result<(), std::io::Error>> {
-        std::pin::Pin::new(&mut &self.pt).poll_flush(cx)
+        std::task::Poll::Ready(Ok(()))
     }
+}
+
+/// The child end of the pty
+///
+/// See [`Pty::pts`] and [`Command::spawn`](crate::Command::spawn)
+pub struct Pts(pub(crate) crate::sys::Pts);
+
+/// Borrowed read half of a [`Pty`]
+pub struct ReadPty<'a>(&'a AsyncPty);
 
-    fn poll_close(
+impl<'a> tokio::io::AsyncRead for ReadPty<'a> {
+    fn poll_read(
         self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        loop {
+            let mut guard = match self.0.poll_read_ready(cx) {
+                std::task::Poll::Ready(guard) => guard,
+                std::task::Poll::Pending => return std::task::Poll::Pending,
+            }?;
+            let mut b = [0u8; 4096];
+            match guard.try_io(|inner| (&inner.get_ref().0).read(&mut b)) {
+                Ok(Ok(bytes)) => {
+                    // XXX this is safe, but not particularly efficient
+                    buf.clear();
+                    buf.initialize_unfilled_to(bytes);
+                    buf.set_filled(bytes);
+                    buf.filled_mut().copy_from_slice(&b[..bytes]);
+                    return std::task::Poll::Ready(Ok(()));
+                }
+                Ok(Err(e)) => return std::task::Poll::Ready(Err(e)),
+                Err(_would_block) => continue,
+            }
+        }
+    }
+}
+
+/// Borrowed write half of a [`Pty`]
+pub struct WritePty<'a>(&'a AsyncPty);
+
+impl<'a> WritePty<'a> {
+    /// Change the terminal size associated with the pty.
+    ///
+    /// # Errors
+    /// Returns an error if we were unable to set the terminal size.
+    pub fn resize(&self, size: crate::Size) -> crate::Result<()> {
+        self.0.get_ref().set_term_size(size)
+    }
+}
+
+impl<'a> tokio::io::AsyncWrite for WritePty<'a> {
+    fn poll_write(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        loop {
+            let mut guard = match self.0.poll_write_ready(cx) {
+                std::task::Poll::Ready(guard) => guard,
+                std::task::Poll::Pending => return std::task::Poll::Pending,
+            }?;
+            match guard.try_io(|inner| (&inner.get_ref().0).write(buf)) {
+                Ok(result) => return std::task::Poll::Ready(result),
+                Err(_would_block) => continue,
+            }
+        }
+    }
+
+    fn poll_flush(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        loop {
+            let mut guard = match self.0.poll_write_ready(cx) {
+                std::task::Poll::Ready(guard) => guard,
+                std::task::Poll::Pending => return std::task::Poll::Pending,
+            }?;
+            match guard.try_io(|inner| (&inner.get_ref().0).flush()) {
+                Ok(_) => return std::task::Poll::Ready(Ok(())),
+                Err(_would_block) => continue,
+            }
+        }
+    }
+
+    fn poll_shutdown(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+}
+
+/// Owned read half of a [`Pty`]
+#[derive(Debug)]
+pub struct OwnedReadPty(std::sync::Arc<AsyncPty>);
+
+impl OwnedReadPty {
+    /// Attempt to join the two halves of a `Pty` back into a single instance.
+    /// The two halves must have originated from calling
+    /// [`into_split`](Pty::into_split) on a single instance.
+    ///
+    /// # Errors
+    /// Returns an error if the two halves came from different [`Pty`]
+    /// instances. The mismatched halves are returned as part of the error.
+    pub fn unsplit(self, write_half: OwnedWritePty) -> crate::Result<Pty> {
+        let Self(read_pt) = self;
+        let OwnedWritePty(write_pt) = write_half;
+        if std::sync::Arc::ptr_eq(&read_pt, &write_pt) {
+            drop(write_pt);
+            Ok(Pty(std::sync::Arc::try_unwrap(read_pt)
+                // it shouldn't be possible for more than two references to
+                // the same pty to exist
+                .unwrap_or_else(|_| unreachable!())))
+        } else {
+            Err(crate::Error::Unsplit(
+                Self(read_pt),
+                OwnedWritePty(write_pt),
+            ))
+        }
+    }
+}
+
+impl tokio::io::AsyncRead for OwnedReadPty {
+    fn poll_read(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        loop {
+            let mut guard = match self.0.poll_read_ready(cx) {
+                std::task::Poll::Ready(guard) => guard,
+                std::task::Poll::Pending => return std::task::Poll::Pending,
+            }?;
+            let mut b = [0u8; 4096];
+            match guard.try_io(|inner| (&inner.get_ref().0).read(&mut b)) {
+                Ok(Ok(bytes)) => {
+                    // XXX this is safe, but not particularly efficient
+                    buf.clear();
+                    buf.initialize_unfilled_to(bytes);
+                    buf.set_filled(bytes);
+                    buf.filled_mut().copy_from_slice(&b[..bytes]);
+                    return std::task::Poll::Ready(Ok(()));
+                }
+                Ok(Err(e)) => return std::task::Poll::Ready(Err(e)),
+                Err(_would_block) => continue,
+            }
+        }
+    }
+}
+
+/// Owned write half of a [`Pty`]
+#[derive(Debug)]
+pub struct OwnedWritePty(std::sync::Arc<AsyncPty>);
+
+impl OwnedWritePty {
+    /// Change the terminal size associated with the pty.
+    ///
+    /// # Errors
+    /// Returns an error if we were unable to set the terminal size.
+    pub fn resize(&self, size: crate::Size) -> crate::Result<()> {
+        self.0.get_ref().set_term_size(size)
+    }
+}
+
+impl tokio::io::AsyncWrite for OwnedWritePty {
+    fn poll_write(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        loop {
+            let mut guard = match self.0.poll_write_ready(cx) {
+                std::task::Poll::Ready(guard) => guard,
+                std::task::Poll::Pending => return std::task::Poll::Pending,
+            }?;
+            match guard.try_io(|inner| (&inner.get_ref().0).write(buf)) {
+                Ok(result) => return std::task::Poll::Ready(result),
+                Err(_would_block) => continue,
+            }
+        }
+    }
+
+    fn poll_flush(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        loop {
+            let mut guard = match self.0.poll_write_ready(cx) {
+                std::task::Poll::Ready(guard) => guard,
+                std::task::Poll::Pending => return std::task::Poll::Pending,
+            }?;
+            match guard.try_io(|inner| (&inner.get_ref().0).flush()) {
+                Ok(_) => return std::task::Poll::Ready(Ok(())),
+                Err(_would_block) => continue,
+            }
+        }
+    }
+
+    fn poll_shutdown(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Result<(), std::io::Error>> {
-        std::pin::Pin::new(&mut &self.pt).poll_close(cx)
+        std::task::Poll::Ready(Ok(()))
     }
 }
diff --git a/src/sys.rs b/src/sys.rs
index 0d897bc..244a43d 100644
--- a/src/sys.rs
+++ b/src/sys.rs
@@ -1,83 +1,127 @@
-use std::os::unix::io::{FromRawFd as _, IntoRawFd as _};
+use std::os::unix::io::{AsRawFd as _, FromRawFd as _, IntoRawFd as _};
 
-pub fn create_pt() -> nix::Result<(std::fs::File, std::path::PathBuf)> {
-    let pt = nix::pty::posix_openpt(
-        nix::fcntl::OFlag::O_RDWR
-            | nix::fcntl::OFlag::O_NOCTTY
-            | nix::fcntl::OFlag::O_CLOEXEC,
-    )?;
-    nix::pty::grantpt(&pt)?;
-    nix::pty::unlockpt(&pt)?;
+#[derive(Debug)]
+pub struct Pty(pub nix::pty::PtyMaster);
 
-    let ptsname = nix::pty::ptsname_r(&pt)?.into();
+impl Pty {
+    pub fn open() -> crate::Result<Self> {
+        let pt = nix::pty::posix_openpt(
+            nix::fcntl::OFlag::O_RDWR
+                | nix::fcntl::OFlag::O_NOCTTY
+                | nix::fcntl::OFlag::O_CLOEXEC,
+        )?;
+        nix::pty::grantpt(&pt)?;
+        nix::pty::unlockpt(&pt)?;
 
-    let pt_fd = pt.into_raw_fd();
+        Ok(Self(pt))
+    }
+
+    pub fn set_term_size(&self, size: crate::Size) -> crate::Result<()> {
+        let size = size.into();
+        let fd = self.0.as_raw_fd();
 
-    // Safety: posix_openpt (or the previous functions operating on the
-    // result) would have returned an Err (causing us to return early) if the
-    // file descriptor was invalid. additionally, into_raw_fd gives up
-    // ownership over the file descriptor, allowing the newly created File
-    // object to take full ownership.
-    let pt = unsafe { std::fs::File::from_raw_fd(pt_fd) };
+        // Safety: nix::pty::PtyMaster is required to contain a valid file
+        // descriptor and size is guaranteed to be initialized because it's a
+        // normal rust value, and nix::pty::Winsize is a repr(C) struct with
+        // the same layout as `struct winsize` from sys/ioctl.h.
+        Ok(unsafe {
+            set_term_size_unsafe(fd, std::ptr::NonNull::from(&size).as_ptr())
+        }
+        .map(|_| ())?)
+    }
+
+    pub fn pts(&self) -> crate::Result<Pts> {
+        Ok(Pts(std::fs::OpenOptions::new()
+            .read(true)
+            .write(true)
+            .open(nix::pty::ptsname_r(&self.0)?)?
+            .into_raw_fd()))
+    }
+
+    #[cfg(feature = "async")]
+    pub fn set_nonblocking(&self) -> nix::Result<()> {
+        let bits = nix::fcntl::fcntl(
+            self.0.as_raw_fd(),
+            nix::fcntl::FcntlArg::F_GETFL,
+        )?;
+        // Safety: bits was just returned from a F_GETFL call. ideally i would
+        // just be able to use from_bits here, but it fails for some reason?
+        let mut opts =
+            unsafe { nix::fcntl::OFlag::from_bits_unchecked(bits) };
+        opts |= nix::fcntl::OFlag::O_NONBLOCK;
+        nix::fcntl::fcntl(
+            self.0.as_raw_fd(),
+            nix::fcntl::FcntlArg::F_SETFL(opts),
+        )?;
+
+        Ok(())
+    }
+}
 
-    Ok((pt, ptsname))
+impl std::os::unix::io::AsRawFd for Pty {
+    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
+        self.0.as_raw_fd()
+    }
 }
 
-pub fn set_term_size(
-    fh: &impl std::os::unix::io::AsRawFd,
-    size: crate::Size,
-) -> nix::Result<()> {
-    let size = size.into();
-    let fd = fh.as_raw_fd();
-
-    // Safety: std::fs::File is required to contain a valid file descriptor
-    // and size is guaranteed to be initialized because it's a normal rust
-    // value, and nix::pty::Winsize is a repr(C) struct with the same layout
-    // as `struct winsize` from sys/ioctl.h.
-    unsafe {
-        set_term_size_unsafe(fd, std::ptr::NonNull::from(&size).as_ptr())
+pub struct Pts(std::os::unix::io::RawFd);
+
+impl Pts {
+    pub fn setup_subprocess(
+        &self,
+    ) -> nix::Result<(
+        std::process::Stdio,
+        std::process::Stdio,
+        std::process::Stdio,
+    )> {
+        let pts_fd = self.0.as_raw_fd();
+
+        let stdin = nix::fcntl::fcntl(
+            pts_fd,
+            nix::fcntl::FcntlArg::F_DUPFD_CLOEXEC(0),
+        )?;
+        let stdout = nix::fcntl::fcntl(
+            pts_fd,
+            nix::fcntl::FcntlArg::F_DUPFD_CLOEXEC(0),
+        )?;
+        let stderr = nix::fcntl::fcntl(
+            pts_fd,
+            nix::fcntl::FcntlArg::F_DUPFD_CLOEXEC(0),
+        )?;
+
+        // Safety: these file descriptors were all just returned from dup, so
+        // they must be valid
+        Ok((
+            unsafe { std::process::Stdio::from_raw_fd(stdin) },
+            unsafe { std::process::Stdio::from_raw_fd(stdout) },
+            unsafe { std::process::Stdio::from_raw_fd(stderr) },
+        ))
+    }
+
+    pub fn session_leader(&self) -> impl FnMut() -> std::io::Result<()> {
+        let pts_fd = self.0.as_raw_fd();
+        move || {
+            nix::unistd::setsid()?;
+            set_controlling_terminal(pts_fd)?;
+            Ok(())
+        }
     }
-    .map(|_| ())
 }
 
-pub fn setup_subprocess(
-    pts: &impl std::os::unix::io::AsRawFd,
-) -> nix::Result<(
-    std::process::Stdio,
-    std::process::Stdio,
-    std::process::Stdio,
-)> {
-    let pts_fd = pts.as_raw_fd();
-
-    let stdin =
-        nix::fcntl::fcntl(pts_fd, nix::fcntl::FcntlArg::F_DUPFD_CLOEXEC(0))?;
-    let stdout =
-        nix::fcntl::fcntl(pts_fd, nix::fcntl::FcntlArg::F_DUPFD_CLOEXEC(0))?;
-    let stderr =
-        nix::fcntl::fcntl(pts_fd, nix::fcntl::FcntlArg::F_DUPFD_CLOEXEC(0))?;
-
-    // Safety: these file descriptors were all just returned from dup, so they
-    // must be valid
-    Ok((
-        unsafe { std::process::Stdio::from_raw_fd(stdin) },
-        unsafe { std::process::Stdio::from_raw_fd(stdout) },
-        unsafe { std::process::Stdio::from_raw_fd(stderr) },
-    ))
+impl Drop for Pts {
+    fn drop(&mut self) {
+        let _ = nix::unistd::close(self.0);
+    }
 }
 
-pub fn session_leader(
-    pts: &impl std::os::unix::io::AsRawFd,
-) -> impl FnMut() -> std::io::Result<()> {
-    let pts_fd = pts.as_raw_fd();
-    move || {
-        nix::unistd::setsid()?;
-        set_controlling_terminal(pts_fd)?;
-        Ok(())
+impl std::os::unix::io::AsRawFd for Pts {
+    fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
+        self.0
     }
 }
 
 fn set_controlling_terminal(fd: std::os::unix::io::RawFd) -> nix::Result<()> {
-    // Safety: std::fs::File is required to contain a valid file descriptor
+    // Safety: Pts is required to contain a valid file descriptor
     unsafe { set_controlling_terminal_unsafe(fd, std::ptr::null()) }
         .map(|_| ())
 }