use crate::fd_table::{
allocate_file_description_id, FdResult, FileDescription, ProcessFdTable, SharedFileDescription,
FILETYPE_PIPE, O_NONBLOCK, O_RDONLY, O_RDWR, O_WRONLY,
};
use crate::poll::{PollEvents, PollNotifier, POLLERR, POLLHUP, POLLIN, POLLOUT};
use std::collections::{BTreeMap, VecDeque};
use std::error::Error;
use std::fmt;
use std::sync::{Arc, Condvar, Mutex, MutexGuard};
use std::time::Duration;
use web_time::Instant;
pub const MAX_PIPE_BUFFER_BYTES: usize = 65_536;
pub const PIPE_BUF_BYTES: usize = 4_096;
pub type PipeResult<T> = Result<T, PipeError>;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PipeError {
code: &'static str,
message: String,
}
impl PipeError {
pub fn code(&self) -> &'static str {
self.code
}
fn bad_file_descriptor(message: impl Into<String>) -> Self {
Self {
code: "EBADF",
message: message.into(),
}
}
fn broken_pipe(message: impl Into<String>) -> Self {
Self {
code: "EPIPE",
message: message.into(),
}
}
fn would_block(message: impl Into<String>) -> Self {
Self {
code: "EAGAIN",
message: message.into(),
}
}
fn no_reader(message: impl Into<String>) -> Self {
Self {
code: "ENXIO",
message: message.into(),
}
}
}
impl fmt::Display for PipeError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}: {}", self.code, self.message)
}
}
impl Error for PipeError {}
#[derive(Debug, Clone)]
pub struct PipeEnd {
pub description: SharedFileDescription,
pub filetype: u8,
}
#[derive(Debug, Clone)]
pub struct PipePair {
pub read: PipeEnd,
pub write: PipeEnd,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct PipeRef {
pipe_id: u64,
end: PipeSide,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PipeSide {
Read,
Write,
ReadWrite,
}
#[derive(Debug, Default)]
struct PendingRead {
length: usize,
result: Option<Option<Vec<u8>>>,
}
#[derive(Debug)]
struct PipeState {
buffer: VecDeque<Vec<u8>>,
readers: usize,
writers: usize,
waiting_reads: VecDeque<u64>,
mode: u32,
uid: u32,
gid: u32,
named_key: Option<(u64, u64)>,
}
impl Default for PipeState {
fn default() -> Self {
Self {
buffer: VecDeque::new(),
readers: 0,
writers: 0,
waiting_reads: VecDeque::new(),
mode: 0o600,
uid: 0,
gid: 0,
named_key: None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PipeMetadata {
pub mode: u32,
pub uid: u32,
pub gid: u32,
}
#[derive(Debug)]
struct PipeManagerState {
pipes: BTreeMap<u64, PipeState>,
desc_to_pipe: BTreeMap<u64, PipeRef>,
named_pipes: BTreeMap<(u64, u64), u64>,
waiters: BTreeMap<u64, PendingRead>,
next_pipe_id: u64,
next_waiter_id: u64,
}
impl Default for PipeManagerState {
fn default() -> Self {
Self {
pipes: BTreeMap::new(),
desc_to_pipe: BTreeMap::new(),
named_pipes: BTreeMap::new(),
waiters: BTreeMap::new(),
next_pipe_id: 1,
next_waiter_id: 1,
}
}
}
#[derive(Debug)]
struct PipeManagerInner {
state: Mutex<PipeManagerState>,
waiters: Condvar,
}
#[derive(Debug, Clone)]
pub struct PipeManager {
inner: Arc<PipeManagerInner>,
notifier: Option<PollNotifier>,
}
impl Default for PipeManager {
fn default() -> Self {
Self {
inner: Arc::new(PipeManagerInner {
state: Mutex::new(PipeManagerState::default()),
waiters: Condvar::new(),
}),
notifier: None,
}
}
}
impl PipeManager {
pub fn is_write_to_read_pair(
&self,
write_description_id: u64,
read_description_id: u64,
) -> bool {
let state = lock_or_recover(&self.inner.state);
match (
state.desc_to_pipe.get(&write_description_id),
state.desc_to_pipe.get(&read_description_id),
) {
(Some(write), Some(read)) => {
write.pipe_id == read.pipe_id
&& write.end == PipeSide::Write
&& read.end == PipeSide::Read
}
_ => false,
}
}
pub fn new() -> Self {
Self::default()
}
pub(crate) fn with_notifier(notifier: PollNotifier) -> Self {
Self {
notifier: Some(notifier),
..Self::default()
}
}
pub fn create_pipe(&self) -> PipePair {
let mut state = lock_or_recover(&self.inner.state);
let pipe_id = state.next_pipe_id;
state.next_pipe_id += 1;
let read_id = allocate_file_description_id();
let write_id = allocate_file_description_id();
state.pipes.insert(
pipe_id,
PipeState {
readers: 1,
writers: 1,
..PipeState::default()
},
);
state.desc_to_pipe.insert(
read_id,
PipeRef {
pipe_id,
end: PipeSide::Read,
},
);
state.desc_to_pipe.insert(
write_id,
PipeRef {
pipe_id,
end: PipeSide::Write,
},
);
drop(state);
PipePair {
read: PipeEnd {
description: Arc::new(FileDescription::with_ref_count(
read_id,
format!("pipe:{pipe_id}:read"),
O_RDONLY,
0,
)),
filetype: FILETYPE_PIPE,
},
write: PipeEnd {
description: Arc::new(FileDescription::with_ref_count(
write_id,
format!("pipe:{pipe_id}:write"),
O_WRONLY,
0,
)),
filetype: FILETYPE_PIPE,
},
}
}
pub fn open_named_pipe(
&self,
key: (u64, u64),
path: &str,
flags: u32,
timeout: Option<Duration>,
) -> PipeResult<PipeEnd> {
let access_mode = flags & 0b11;
if !matches!(access_mode, O_RDONLY | O_WRONLY | O_RDWR) {
return Err(PipeError::bad_file_descriptor("invalid FIFO access mode"));
}
let mut state = lock_or_recover(&self.inner.state);
let pipe_id = match state.named_pipes.get(&key).copied() {
Some(pipe_id) => pipe_id,
None => {
let pipe_id = state.next_pipe_id;
state.next_pipe_id += 1;
state.named_pipes.insert(key, pipe_id);
state.pipes.insert(
pipe_id,
PipeState {
named_key: Some(key),
..PipeState::default()
},
);
pipe_id
}
};
if access_mode == O_WRONLY
&& flags & O_NONBLOCK != 0
&& state
.pipes
.get(&pipe_id)
.is_none_or(|pipe| pipe.readers == 0)
{
return Err(PipeError::no_reader(format!("FIFO has no reader: {path}")));
}
let description_id = allocate_file_description_id();
let side = match access_mode {
O_RDONLY => PipeSide::Read,
O_WRONLY => PipeSide::Write,
O_RDWR => PipeSide::ReadWrite,
_ => unreachable!(),
};
state
.desc_to_pipe
.insert(description_id, PipeRef { pipe_id, end: side });
let pipe = state
.pipes
.get_mut(&pipe_id)
.expect("named pipe must exist after allocation");
match side {
PipeSide::Read => pipe.readers += 1,
PipeSide::Write => pipe.writers += 1,
PipeSide::ReadWrite => {
pipe.readers += 1;
pipe.writers += 1;
}
}
self.notify_waiters_and_pollers();
let should_wait = flags & O_NONBLOCK == 0 && access_mode != O_RDWR;
if should_wait {
let ready = |state: &PipeManagerState| {
state.pipes.get(&pipe_id).is_some_and(|pipe| match side {
PipeSide::Read => pipe.writers > 0,
PipeSide::Write => pipe.readers > 0,
PipeSide::ReadWrite => true,
})
};
if let Some(timeout) = timeout {
let (next, result) = self
.inner
.waiters
.wait_timeout_while(state, timeout, |state| !ready(state))
.unwrap_or_else(|poisoned| poisoned.into_inner());
state = next;
if result.timed_out() && !ready(&state) {
drop(state);
self.close(description_id);
return Err(PipeError::would_block(format!(
"FIFO open timed out: {path}"
)));
}
} else {
state = self
.inner
.waiters
.wait_while(state, |state| !ready(state))
.unwrap_or_else(|poisoned| poisoned.into_inner());
}
}
drop(state);
Ok(PipeEnd {
description: Arc::new(FileDescription::with_ref_count(
description_id,
path,
flags,
0,
)),
filetype: FILETYPE_PIPE,
})
}
pub fn poll(&self, description_id: u64, requested: PollEvents) -> PipeResult<PollEvents> {
let state = lock_or_recover(&self.inner.state);
let pipe_ref = state
.desc_to_pipe
.get(&description_id)
.copied()
.ok_or_else(|| PipeError::bad_file_descriptor("not a pipe end"))?;
let pipe = state
.pipes
.get(&pipe_ref.pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
let mut events = PollEvents::empty();
match pipe_ref.end {
PipeSide::Read => {
if requested.intersects(POLLIN) && !pipe.buffer.is_empty() {
events |= POLLIN;
}
if pipe.writers == 0 {
events |= POLLHUP;
}
}
PipeSide::Write => {
if pipe.readers == 0 {
events |= POLLERR;
} else if requested.intersects(POLLOUT)
&& (available_capacity(pipe) > 0 || !pipe.waiting_reads.is_empty())
{
events |= POLLOUT;
}
}
PipeSide::ReadWrite => {
if requested.intersects(POLLIN) && !pipe.buffer.is_empty() {
events |= POLLIN;
}
if requested.intersects(POLLOUT)
&& (available_capacity(pipe) > 0 || !pipe.waiting_reads.is_empty())
{
events |= POLLOUT;
}
}
}
Ok(events)
}
pub fn write(&self, description_id: u64, data: impl AsRef<[u8]>) -> PipeResult<usize> {
self.write_with_mode(description_id, data, true)
}
pub fn write_blocking(&self, description_id: u64, data: impl AsRef<[u8]>) -> PipeResult<usize> {
self.write_with_mode(description_id, data, false)
}
pub fn write_with_mode(
&self,
description_id: u64,
data: impl AsRef<[u8]>,
nonblocking: bool,
) -> PipeResult<usize> {
let payload = data.as_ref();
let mut state = lock_or_recover(&self.inner.state);
let pipe_ref = state
.desc_to_pipe
.get(&description_id)
.copied()
.ok_or_else(|| PipeError::bad_file_descriptor("not a pipe write end"))?;
if !matches!(pipe_ref.end, PipeSide::Write | PipeSide::ReadWrite) {
return Err(PipeError::bad_file_descriptor("not a pipe write end"));
}
loop {
let waiter_id = {
let pipe = state
.pipes
.get_mut(&pipe_ref.pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
if pipe.readers == 0 {
return Err(PipeError::broken_pipe("read end closed"));
}
pipe.waiting_reads.pop_front()
};
if let Some(waiter_id) = waiter_id {
let waiter_length = match state.waiters.get(&waiter_id) {
Some(waiter) => waiter.length,
None => continue,
};
let delivered_len = waiter_length.min(payload.len());
let delivered = payload[..delivered_len].to_vec();
let remainder = &payload[delivered_len..];
if !remainder.is_empty() {
let pipe = state
.pipes
.get_mut(&pipe_ref.pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
pipe.buffer.push_back(remainder.to_vec());
}
if let Some(waiter) = state.waiters.get_mut(&waiter_id) {
waiter.result = Some(Some(delivered));
self.notify_waiters_and_pollers();
return Ok(payload.len());
}
continue;
}
let current_buffer_size = {
let pipe = state
.pipes
.get(&pipe_ref.pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
buffer_size(&pipe.buffer)
};
let available = MAX_PIPE_BUFFER_BYTES.saturating_sub(current_buffer_size);
if payload.len() <= PIPE_BUF_BYTES {
if available >= payload.len() {
let pipe = state
.pipes
.get_mut(&pipe_ref.pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
pipe.buffer.push_back(payload.to_vec());
self.notify_waiters_and_pollers();
return Ok(payload.len());
}
} else if available > 0 {
let chunk_len = available.min(payload.len());
let pipe = state
.pipes
.get_mut(&pipe_ref.pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
pipe.buffer.push_back(payload[..chunk_len].to_vec());
self.notify_waiters_and_pollers();
return Ok(chunk_len);
}
if nonblocking {
return Err(PipeError::would_block("pipe buffer full"));
}
state = wait_or_recover(&self.inner.waiters, state);
}
}
pub fn read(&self, description_id: u64, length: usize) -> PipeResult<Option<Vec<u8>>> {
self.read_with_timeout(description_id, length, None)
}
pub fn read_with_timeout(
&self,
description_id: u64,
length: usize,
timeout: Option<Duration>,
) -> PipeResult<Option<Vec<u8>>> {
let mut state = lock_or_recover(&self.inner.state);
let pipe_ref = state
.desc_to_pipe
.get(&description_id)
.copied()
.ok_or_else(|| PipeError::bad_file_descriptor("not a pipe read end"))?;
if !matches!(pipe_ref.end, PipeSide::Read | PipeSide::ReadWrite) {
return Err(PipeError::bad_file_descriptor("not a pipe read end"));
}
let mut waiter_id = None;
let deadline = timeout.map(|duration| Instant::now() + duration);
loop {
if let Some(id) = waiter_id {
if let Some(waiter) = state.waiters.get_mut(&id) {
if let Some(result) = waiter.result.take() {
state.waiters.remove(&id);
return Ok(result);
}
}
}
{
let pipe = state
.pipes
.get_mut(&pipe_ref.pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
if !pipe.buffer.is_empty() {
let result = drain_buffer(&mut pipe.buffer, length);
self.notify_waiters_and_pollers();
return Ok(Some(result));
}
if pipe.writers == 0 {
if let Some(id) = waiter_id {
state.waiters.remove(&id);
}
return Ok(None);
}
}
if waiter_id.is_none() && deadline.is_some_and(|deadline| Instant::now() >= deadline) {
return Err(PipeError::would_block("pipe read timed out"));
}
let id = if let Some(id) = waiter_id {
id
} else {
let next = state.next_waiter_id;
state.next_waiter_id += 1;
state.waiters.insert(
next,
PendingRead {
length,
result: None,
},
);
let Some(pipe) = state.pipes.get_mut(&pipe_ref.pipe_id) else {
state.waiters.remove(&next);
return Err(PipeError::bad_file_descriptor("pipe not found"));
};
pipe.waiting_reads.push_back(next);
self.notify_waiters_and_pollers();
waiter_id = Some(next);
next
};
let Some(deadline) = deadline else {
state = wait_or_recover(&self.inner.waiters, state);
if !state.waiters.contains_key(&id) {
waiter_id = None;
}
continue;
};
let now = Instant::now();
if now >= deadline {
if let Some(id) = waiter_id.take() {
state.waiters.remove(&id);
if let Some(pipe) = state.pipes.get_mut(&pipe_ref.pipe_id) {
pipe.waiting_reads.retain(|queued| *queued != id);
}
self.notify_waiters_and_pollers();
}
return Err(PipeError::would_block("pipe read timed out"));
}
let remaining = deadline.saturating_duration_since(now);
let (next_state, wait_result) =
wait_timeout_or_recover(&self.inner.waiters, state, remaining);
state = next_state;
if !state.waiters.contains_key(&id) {
waiter_id = None;
}
if wait_result.timed_out() {
if let Some(id) = waiter_id.take() {
state.waiters.remove(&id);
if let Some(pipe) = state.pipes.get_mut(&pipe_ref.pipe_id) {
pipe.waiting_reads.retain(|queued| *queued != id);
}
self.notify_waiters_and_pollers();
}
return Err(PipeError::would_block("pipe read timed out"));
}
}
}
pub fn close(&self, description_id: u64) {
let mut state = lock_or_recover(&self.inner.state);
let Some(pipe_ref) = state.desc_to_pipe.remove(&description_id) else {
return;
};
let (waiter_ids, remove_pipe, should_notify) =
if let Some(pipe) = state.pipes.get_mut(&pipe_ref.pipe_id) {
match pipe_ref.end {
PipeSide::Read => {
pipe.readers = pipe.readers.saturating_sub(1);
(Vec::new(), pipe.readers == 0 && pipe.writers == 0, true)
}
PipeSide::Write => {
pipe.writers = pipe.writers.saturating_sub(1);
let waiter_ids = if pipe.writers == 0 {
pipe.waiting_reads.drain(..).collect::<Vec<_>>()
} else {
Vec::new()
};
(waiter_ids, pipe.readers == 0 && pipe.writers == 0, true)
}
PipeSide::ReadWrite => {
pipe.readers = pipe.readers.saturating_sub(1);
pipe.writers = pipe.writers.saturating_sub(1);
let waiter_ids = if pipe.writers == 0 {
pipe.waiting_reads.drain(..).collect::<Vec<_>>()
} else {
Vec::new()
};
(waiter_ids, pipe.readers == 0 && pipe.writers == 0, true)
}
}
} else {
(Vec::new(), false, false)
};
for waiter_id in waiter_ids {
if let Some(waiter) = state.waiters.get_mut(&waiter_id) {
waiter.result = Some(None);
}
}
if remove_pipe {
if let Some(pipe) = state.pipes.remove(&pipe_ref.pipe_id) {
if let Some(key) = pipe.named_key {
state.named_pipes.remove(&key);
}
}
}
if should_notify {
self.notify_waiters_and_pollers();
}
}
pub fn is_pipe(&self, description_id: u64) -> bool {
lock_or_recover(&self.inner.state)
.desc_to_pipe
.contains_key(&description_id)
}
pub fn pipe_id_for(&self, description_id: u64) -> Option<u64> {
lock_or_recover(&self.inner.state)
.desc_to_pipe
.get(&description_id)
.map(|pipe_ref| pipe_ref.pipe_id)
}
pub fn metadata(&self, description_id: u64) -> Option<PipeMetadata> {
let state = lock_or_recover(&self.inner.state);
let pipe_id = state.desc_to_pipe.get(&description_id)?.pipe_id;
let pipe = state.pipes.get(&pipe_id)?;
Some(PipeMetadata {
mode: pipe.mode,
uid: pipe.uid,
gid: pipe.gid,
})
}
pub fn set_owner(&self, description_id: u64, uid: u32, gid: u32) -> PipeResult<()> {
let mut state = lock_or_recover(&self.inner.state);
let pipe_id = state
.desc_to_pipe
.get(&description_id)
.ok_or_else(|| PipeError::bad_file_descriptor("not a pipe end"))?
.pipe_id;
let pipe = state
.pipes
.get_mut(&pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
pipe.uid = uid;
pipe.gid = gid;
Ok(())
}
pub fn chmod(&self, description_id: u64, mode: u32) -> PipeResult<()> {
let mut state = lock_or_recover(&self.inner.state);
let pipe_id = state
.desc_to_pipe
.get(&description_id)
.ok_or_else(|| PipeError::bad_file_descriptor("not a pipe end"))?
.pipe_id;
let pipe = state
.pipes
.get_mut(&pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
pipe.mode = mode & 0o7777;
Ok(())
}
pub fn pipe_count(&self) -> usize {
lock_or_recover(&self.inner.state).pipes.len()
}
pub fn has_named_pipe(&self, key: (u64, u64)) -> bool {
lock_or_recover(&self.inner.state)
.named_pipes
.contains_key(&key)
}
pub fn named_pipe_peer_ready(&self, description_id: u64) -> PipeResult<Option<bool>> {
let state = lock_or_recover(&self.inner.state);
let pipe_ref = state
.desc_to_pipe
.get(&description_id)
.copied()
.ok_or_else(|| PipeError::bad_file_descriptor("not a pipe end"))?;
let pipe = state
.pipes
.get(&pipe_ref.pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
if pipe.named_key.is_none() {
return Ok(None);
}
Ok(Some(match pipe_ref.end {
PipeSide::Read => pipe.writers > 0,
PipeSide::Write => pipe.readers > 0,
PipeSide::ReadWrite => true,
}))
}
pub fn buffered_bytes(&self) -> usize {
lock_or_recover(&self.inner.state)
.pipes
.values()
.map(|pipe| buffer_size(&pipe.buffer))
.sum()
}
pub fn waiting_reader_count(&self, description_id: u64) -> PipeResult<usize> {
let state = lock_or_recover(&self.inner.state);
let pipe_ref = state
.desc_to_pipe
.get(&description_id)
.copied()
.ok_or_else(|| PipeError::bad_file_descriptor("not a pipe end"))?;
let pipe = state
.pipes
.get(&pipe_ref.pipe_id)
.ok_or_else(|| PipeError::bad_file_descriptor("pipe not found"))?;
Ok(pipe.waiting_reads.len())
}
pub fn pending_read_waiter_count(&self) -> usize {
lock_or_recover(&self.inner.state).waiters.len()
}
pub fn create_pipe_fds(&self, fd_table: &mut ProcessFdTable) -> FdResult<(u32, u32)> {
let pipe = self.create_pipe();
let read_fd =
fd_table.open_with(Arc::clone(&pipe.read.description), FILETYPE_PIPE, None)?;
match fd_table.open_with(Arc::clone(&pipe.write.description), FILETYPE_PIPE, None) {
Ok(write_fd) => Ok((read_fd, write_fd)),
Err(error) => {
fd_table.close(read_fd);
self.close(pipe.read.description.id());
self.close(pipe.write.description.id());
Err(error)
}
}
}
fn notify_waiters_and_pollers(&self) {
self.inner.waiters.notify_all();
if let Some(notifier) = &self.notifier {
notifier.notify();
}
}
}
fn buffer_size(buffer: &VecDeque<Vec<u8>>) -> usize {
buffer.iter().map(Vec::len).sum()
}
fn available_capacity(pipe: &PipeState) -> usize {
MAX_PIPE_BUFFER_BYTES.saturating_sub(buffer_size(&pipe.buffer))
}
fn drain_buffer(buffer: &mut VecDeque<Vec<u8>>, length: usize) -> Vec<u8> {
let mut chunks = Vec::new();
let mut remaining = length;
while remaining > 0 {
let Some(chunk) = buffer.pop_front() else {
break;
};
if chunk.len() <= remaining {
remaining -= chunk.len();
chunks.push(chunk);
} else {
let (head, tail) = chunk.split_at(remaining);
chunks.push(head.to_vec());
buffer.push_front(tail.to_vec());
remaining = 0;
}
}
if chunks.len() == 1 {
return chunks.pop().expect("single chunk should exist");
}
let total = chunks.iter().map(Vec::len).sum();
let mut result = Vec::with_capacity(total);
for chunk in chunks {
result.extend_from_slice(&chunk);
}
result
}
fn lock_or_recover<'a, T>(mutex: &'a Mutex<T>) -> MutexGuard<'a, T> {
match mutex.lock() {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
fn wait_or_recover<'a, T>(condvar: &Condvar, guard: MutexGuard<'a, T>) -> MutexGuard<'a, T> {
match condvar.wait(guard) {
Ok(guard) => guard,
Err(poisoned) => poisoned.into_inner(),
}
}
fn wait_timeout_or_recover<'a, T>(
condvar: &Condvar,
guard: MutexGuard<'a, T>,
timeout: Duration,
) -> (MutexGuard<'a, T>, std::sync::WaitTimeoutResult) {
match condvar.wait_timeout(guard, timeout) {
Ok(result) => result,
Err(poisoned) => poisoned.into_inner(),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn zero_timeout_empty_read_does_not_publish_false_readiness() {
let notifier = PollNotifier::default();
let manager = PipeManager::with_notifier(notifier.clone());
let pipe = manager.create_pipe();
let observed = notifier.snapshot();
let error = manager
.read_with_timeout(pipe.read.description.id(), 1, Some(Duration::ZERO))
.expect_err("empty nonblocking read must return EAGAIN");
assert_eq!(error.code(), "EAGAIN");
assert_eq!(notifier.snapshot(), observed);
let state = lock_or_recover(&manager.inner.state);
assert!(state.waiters.is_empty());
assert_eq!(state.next_waiter_id, 1);
}
}