use std::future::Future;
use std::io;
use std::os::fd::RawFd;
use std::pin::Pin;
use std::task::{Context, Poll};
use crate::backend::file::{OpenPath, OsFile};
use crate::backend::socket::FdSlot;
use crate::fiber::Holding;
use crate::manifold::file::{FileOutcome, Files, OpenDone, ReadDone, SpliceDone};
use crate::{Driver, backend};
#[derive(Debug)]
pub enum Source {
Fd(RawFd),
Fixed(FdSlot),
}
impl Source {
pub fn fd(fd: RawFd) -> Self {
Self::Fd(fd)
}
pub fn fixed(slot: FdSlot) -> Self {
Self::Fixed(slot)
}
pub fn of(file: &OsFile) -> Self {
Self::Fd(file.fd())
}
}
enum Stage {
Init,
Pending(backend::token::Token),
Done,
}
pub struct Open<'d, 'p, const ID: u8, const N: usize> {
host: Holding<'d, Files<ID, N>>,
driver: *mut Driver,
path: &'p OpenPath,
flags: i32,
fixed: Option<FdSlot>,
stage: Stage,
}
impl<'d, 'p, const ID: u8, const N: usize> Open<'d, 'p, ID, N> {
pub(crate) fn new(
host: Holding<'d, Files<ID, N>>,
driver: &mut Driver,
path: &'p OpenPath,
flags: i32,
fixed: Option<FdSlot>,
) -> Self {
Self {
host,
driver,
path,
flags,
fixed,
stage: Stage::Init,
}
}
}
impl<const ID: u8, const N: usize> Unpin for Open<'_, '_, ID, N> {}
impl<const ID: u8, const N: usize> Drop for Open<'_, '_, ID, N> {
fn drop(&mut self) {
if let Stage::Pending(token) = self.stage {
let driver = unsafe { &mut *self.driver };
self.host.hold().cancel_open(token, driver);
}
}
}
impl<'d, 'p, const ID: u8, const N: usize> Future for Open<'d, 'p, ID, N> {
type Output = io::Result<Source>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut();
let token = match this.stage {
Stage::Done => return Poll::Ready(Err(already_done())),
Stage::Pending(t) => t,
Stage::Init => {
let driver = unsafe { &mut *this.driver };
let begun = match this.fixed {
Some(slot) => this
.host
.hold()
.begin_open_fixed(this.path, this.flags, slot, driver),
None => this.host.hold().begin_open(this.path, this.flags, driver),
};
let Some(t) = begun else {
this.stage = Stage::Done;
return Poll::Ready(Err(io::Error::other("dope::file: open submit failed")));
};
this.stage = Stage::Pending(t);
t
}
};
match this.host.hold().poll_open(token, cx.waker()) {
FileOutcome::Done(done) => {
this.stage = Stage::Done;
Poll::Ready(match done {
OpenDone::Failed(errno) => Err(io::Error::from_raw_os_error(errno)),
OpenDone::Fd(fd) => match this.fixed {
Some(slot) => Ok(Source::Fixed(slot)),
None => Ok(Source::Fd(fd)),
},
})
}
FileOutcome::Pending => Poll::Pending,
}
}
}
pub struct Read<'d, const ID: u8, const N: usize> {
host: Holding<'d, Files<ID, N>>,
driver: *mut Driver,
src: Source,
buf: Option<Vec<u8>>,
offset: u64,
stage: Stage,
}
impl<'d, const ID: u8, const N: usize> Read<'d, ID, N> {
pub(crate) fn new(
host: Holding<'d, Files<ID, N>>,
driver: &mut Driver,
src: Source,
buf: Vec<u8>,
offset: u64,
) -> Self {
Self {
host,
driver,
src,
buf: Some(buf),
offset,
stage: Stage::Init,
}
}
}
impl<const ID: u8, const N: usize> Unpin for Read<'_, ID, N> {}
impl<const ID: u8, const N: usize> Drop for Read<'_, ID, N> {
fn drop(&mut self) {
if let Stage::Pending(token) = self.stage {
let driver = unsafe { &mut *self.driver };
self.host.hold().cancel_read(token, driver);
}
}
}
impl<'d, const ID: u8, const N: usize> Future for Read<'d, ID, N> {
type Output = (Vec<u8>, io::Result<usize>);
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut();
let token = match this.stage {
Stage::Done => return Poll::Ready((Vec::new(), Err(already_done()))),
Stage::Pending(t) => t,
Stage::Init => {
let buf = this
.buf
.take()
.expect("read future polled without a buffer");
if buf.is_empty() {
this.stage = Stage::Done;
return Poll::Ready((buf, Ok(0)));
}
let host = this.host;
let driver = unsafe { &mut *this.driver };
let begun = match this.src {
Source::Fd(fd) => host.hold().begin_read(fd, buf, this.offset, driver),
Source::Fixed(slot) => {
host.hold().begin_read_fixed(slot, buf, this.offset, driver)
}
};
match begun {
Ok(t) => {
this.stage = Stage::Pending(t);
t
}
Err(buf) => {
this.stage = Stage::Done;
return Poll::Ready((
buf,
Err(io::Error::other("dope::file: read submit failed")),
));
}
}
}
};
match this.host.hold().poll_read(token, cx.waker()) {
FileOutcome::Done((buf, done)) => {
this.stage = Stage::Done;
Poll::Ready(match done {
ReadDone::Read(n) => (buf, Ok(n as usize)),
ReadDone::Eof => (buf, Ok(0)),
ReadDone::Failed(errno) => (buf, Err(io::Error::from_raw_os_error(errno))),
})
}
FileOutcome::Pending => Poll::Pending,
}
}
}
pub struct SpliceToPipe<'d, const ID: u8, const N: usize> {
host: Holding<'d, Files<ID, N>>,
driver: *mut Driver,
src: Source,
off_in: i64,
pipe_write_fd: RawFd,
len: u32,
stage: Stage,
}
impl<'d, const ID: u8, const N: usize> SpliceToPipe<'d, ID, N> {
pub(crate) fn new(
host: Holding<'d, Files<ID, N>>,
driver: &mut Driver,
src: Source,
off_in: i64,
pipe_write_fd: RawFd,
len: u32,
) -> Self {
Self {
host,
driver,
src,
off_in,
pipe_write_fd,
len,
stage: Stage::Init,
}
}
}
impl<const ID: u8, const N: usize> Unpin for SpliceToPipe<'_, ID, N> {}
impl<const ID: u8, const N: usize> Drop for SpliceToPipe<'_, ID, N> {
fn drop(&mut self) {
if let Stage::Pending(token) = self.stage {
let driver = unsafe { &mut *self.driver };
self.host.hold().cancel_splice(token, driver);
}
}
}
impl<'d, const ID: u8, const N: usize> Future for SpliceToPipe<'d, ID, N> {
type Output = io::Result<usize>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.get_mut();
let token = match this.stage {
Stage::Done => return Poll::Ready(Err(already_done())),
Stage::Pending(t) => t,
Stage::Init => {
let fd_in = match this.src {
Source::Fd(fd) => fd,
Source::Fixed(_) => {
this.stage = Stage::Done;
return Poll::Ready(Err(io::Error::other(
"dope::file: splice-to-pipe requires a raw fd source",
)));
}
};
let driver = unsafe { &mut *this.driver };
let begun = this.host.hold().begin_splice_to_pipe(
fd_in,
this.off_in,
this.pipe_write_fd,
this.len,
driver,
);
let Some(t) = begun else {
this.stage = Stage::Done;
return Poll::Ready(Err(io::Error::other("dope::file: splice submit failed")));
};
this.stage = Stage::Pending(t);
t
}
};
match this.host.hold().poll_splice(token, cx.waker()) {
FileOutcome::Done(done) => {
this.stage = Stage::Done;
Poll::Ready(match done {
SpliceDone::Moved(n) => Ok(n as usize),
SpliceDone::Eof => Ok(0),
SpliceDone::Failed(errno) => Err(io::Error::from_raw_os_error(errno)),
})
}
FileOutcome::Pending => Poll::Pending,
}
}
}
fn already_done() -> io::Error {
io::Error::other("dope::file: future polled after completion")
}