Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 62 additions & 3 deletions src/runtime/api/bun/spawn/stdio.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
#[cfg(any(target_os = "linux", target_os = "android"))]
use bun_collections::VecExt;
use bun_jsc::{self as jsc, JSGlobalObject, JSValue, JsResult};
use bun_jsc::{self as jsc, JSGlobalObject, JSValue, JsResult, SysErrorJsc as _};
#[cfg(any(target_os = "linux", target_os = "android"))]
use bun_sys::FdExt as _;
#[cfg(windows)]
Expand All @@ -9,13 +9,16 @@ use bun_sys::{self as sys, Fd};

// `bun.jsc.WebCore` lives in this crate (not `bun_jsc`); alias so the body can
// say `webcore::ReadableStream` / `webcore::body::Value`.
use crate::node::types::PathLikeExt as _;
use crate::webcore;
use crate::webcore::blob::store::Data as StoreData;
use crate::webcore::blob::{BlobExt as _, MAX_SIZE};
use crate::webcore::node_types::{PathLike, PathOrFileDescriptor};

// `bun.jsc.Subprocess.StdioKind` is owned by `process.rs` (defined there to
// keep `process` leaf; `subprocess` re-exports it).
use crate::api::bun_process::{self as process, Dup2 as ProcessDup2, StdioKind};
use crate::api::bun_subprocess::{FileWindow, file_window};

// `SpawnOptions.Stdio` is platform-dependent: process.rs exposes `PosixStdio` /
// `WindowsStdio`; alias the active one as `SpawnOptionsStdio` so the body stays
Expand Down Expand Up @@ -60,6 +63,8 @@ pub(crate) enum Stdio {
Dup2(Dup2),
Path(PathLike<'static>),
Blob(webcore::blob::Any),
/// A sliced `Bun.file()` at stdin: the parent sends the window in chunks. `None` after `Writable::init` gave it to the writer.
FileWindow(Option<Box<FileWindow>>),
#[cfg(any(target_os = "linux", target_os = "android"))]
Memfd(Fd),
Pipe,
Expand Down Expand Up @@ -270,7 +275,9 @@ impl Stdio {
out: d.out,
to: d.to,
}),
Self::Capture(_) | Self::Pipe | Self::ReadableStream(_) => buffer(),
Self::Capture(_) | Self::Pipe | Self::ReadableStream(_) | Self::FileWindow(_) => {
buffer()
}
#[cfg(not(windows))]
Self::SocketFd => SpawnOptionsStdio::SocketFd,
// Windows extra-stdio is a libuv pipe handle (no raw-fd ownership
Expand All @@ -292,7 +299,11 @@ impl Stdio {

pub(crate) fn is_piped(&self) -> bool {
match self {
Self::Capture(_) | Self::Blob(_) | Self::Pipe | Self::ReadableStream(_) => true,
Self::Capture(_)
| Self::Blob(_)
| Self::FileWindow(_)
| Self::Pipe
| Self::ReadableStream(_) => true,
Self::Ipc => cfg!(windows),
_ => false,
}
Expand Down Expand Up @@ -573,6 +584,19 @@ impl Stdio {
if blob.needs_to_read_file() {
if let Some(store) = blob.store() {
if let StoreData::File(ref file) = store.data {
// stdout and stderr write to the file, so a window means nothing there.
let window = match &blob {
webcore::blob::Any::Blob(blob) if i != 1 && i != 2 => blob.file_window(),
_ => None,
};
if window.is_some() && i != 0 {
// Only stdin has a writer in the parent, and the child would read the whole file.
return Err(global.throw_invalid_arguments(format_args!(
"A sliced Bun.file() cannot be used for stdio[{i}] yet"
)));
}
let window =
window.map(|(offset, size)| (offset, (size != MAX_SIZE).then_some(size)));
match file.pathlike {
PathOrFileDescriptor::Fd(store_fd) => {
if Some(store_fd) == fd {
Expand Down Expand Up @@ -602,12 +626,30 @@ impl Stdio {
}
}

if let Some((offset, length)) = window {
let opened = FileWindow::open_fd(store_fd, offset, length);
if self.set_file_window(global, opened)? {
return Ok(());
}
}

*self = Stdio::Fd(store_fd);
}

return Ok(());
}
PathOrFileDescriptor::Path(ref path) => {
if let Some((offset, length)) = window {
let mut buf = bun_paths::path_buffer_pool::get();
// The error names the path that the caller gave, not its NT form.
let opened =
FileWindow::open_path(path.slice_z(&mut buf), offset, length)
.map_err(|err| err.with_path(path.slice()));
if self.set_file_window(global, opened)? {
return Ok(());
}
}

*self = Stdio::Path(path.clone());
return Ok(());
}
Expand Down Expand Up @@ -641,6 +683,23 @@ impl Stdio {
}
}

impl Stdio {
/// Returns `false` for a file that is not regular: it has no window, so the child gets the file.
fn set_file_window(
&mut self,
global: &JSGlobalObject,
opened: sys::Result<file_window::Opened>,
) -> JsResult<bool> {
*self = match opened {
Ok(file_window::Opened::Window(window)) => Stdio::FileWindow(Some(window)),
Ok(file_window::Opened::Empty) => Stdio::Ignore,
Ok(file_window::Opened::NotRegular) => return Ok(false),
Err(err) => return Err(global.throw_value(err.to_js(global))),
};
Ok(true)
}
}

#[cfg(any(target_os = "linux", target_os = "android"))]
impl Stdio {
/// Move the memfd out (ownership passes to the caller); `self` becomes `Ignore`.
Expand Down
3 changes: 2 additions & 1 deletion src/runtime/api/bun/subprocess.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1498,7 +1498,8 @@ impl Subprocess<'_> {
}
}

pub(crate) use bun_spawn::subprocess::{Source, SourceData};
pub(crate) use bun_spawn::file_window;
pub(crate) use bun_spawn::subprocess::{FileWindow, Source, SourceData};

// JSC-tier payloads wrap as `Source::Any(Box<dyn SourceData>)` — the lower-tier
// `bun_spawn` crate cannot name `webcore`/`jsc`, so the vtable travels with the
Expand Down
2 changes: 2 additions & 0 deletions src/runtime/api/bun/subprocess/Readable.rs
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,8 @@ impl Readable {
Readable::Pipe(PipeReader::create(event_loop, process, result, max_size))
}
Stdio::Blob(..) => panic!("TODO: implement Blob support in Stdio readable"),
// Made for stdin only in Stdio::extract_blob(); stdout/stderr never see this.
Stdio::FileWindow(..) => unreachable!("FileWindow at stdout/stderr"),
Stdio::Capture(..) => panic!("TODO: implement capture support in Stdio readable"),
// ReadableStream is handled separately
Stdio::ReadableStream(..) => Readable::Ignore,
Expand Down
16 changes: 15 additions & 1 deletion src/runtime/api/bun/subprocess/Writable.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ use crate::webcore::streams::SourceHandle;
#[cfg(windows)]
use bun_io::pipe_writer::BaseWindowsPipeWriter as _;

use super::{Flags, StaticPipeWriter, StdioResult, Subprocess, js};
use super::{Flags, Source, StaticPipeWriter, StdioResult, Subprocess, js};

pub(crate) enum Writable<'a> {
Pipe(RefPtr<FileSink>),
Expand Down Expand Up @@ -209,6 +209,14 @@ impl<'a> Writable<'a> {
super::source_from_blob(blob),
)));
}
Stdio::FileWindow(window) => {
return Ok(Writable::Buffer(StaticPipeWriter::create(
evtloop,
subprocess as *mut Subprocess<'a>,
result,
Source::FileWindow(window.take().unwrap()),
)));
}
Stdio::Fd(fd) => {
return Ok(Writable::Fd(*fd));
}
Expand Down Expand Up @@ -306,6 +314,12 @@ impl<'a> Writable<'a> {
super::source_from_blob(blob),
)))
}
Stdio::FileWindow(window) => Ok(Writable::Buffer(StaticPipeWriter::create(
evtloop,
std::ptr::from_mut::<Subprocess<'a>>(subprocess),
result,
Source::FileWindow(window.take().unwrap()),
))),
#[cfg(any(target_os = "linux", target_os = "android"))]
Stdio::Memfd(_) => {
// Transfer ownership: `Stdio`'s Drop would close the memfd, so
Expand Down
16 changes: 16 additions & 0 deletions src/runtime/shell/subproc.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1057,6 +1057,14 @@ impl Writable {
JscSubprocess::source_from_blob(blob),
)));
}
Stdio::FileWindow(window) => {
return Ok(Writable::Buffer(StaticPipeWriter::create(
event_loop,
subprocess,
result,
JscSubprocess::Source::FileWindow(window.take().unwrap()),
)));
}
Stdio::Fd(fd) => {
return Ok(Writable::Fd(*fd));
}
Expand Down Expand Up @@ -1110,6 +1118,12 @@ impl Writable {
JscSubprocess::source_from_blob(blob),
)))
}
Stdio::FileWindow(window) => Ok(Writable::Buffer(StaticPipeWriter::create(
event_loop,
subprocess,
result,
JscSubprocess::Source::FileWindow(window.take().unwrap()),
))),
#[cfg(any(target_os = "linux", target_os = "android"))]
Stdio::Memfd(memfd) => {
debug_assert!(memfd.is_valid());
Expand Down Expand Up @@ -1271,6 +1285,7 @@ impl Readable {
// blobs are immutable, so we should only ever get the case
// where the user passed in a Blob with an fd
Stdio::Blob(_) => Readable::Ignore,
Stdio::FileWindow(_) => Readable::Ignore,
Stdio::Pipe => Readable::Pipe(PipeReader::create(
event_loop,
process,
Expand Down Expand Up @@ -1305,6 +1320,7 @@ impl Readable {
// blobs are immutable, so we should only ever get the case
// where the user passed in a Blob with an fd
Stdio::Blob(_) => Readable::Ignore,
Stdio::FileWindow(_) => Readable::Ignore,
#[cfg(any(target_os = "linux", target_os = "android"))]
Stdio::Memfd(memfd) => {
let fd = *memfd;
Expand Down
133 changes: 133 additions & 0 deletions src/spawn/file_window.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
//! A window of a regular file that a `StaticPipeWriter` sends one chunk at a time.

use bun_core::ZStr;
use bun_sys::{self as sys, Fd, File, O};

pub struct FileWindow {
file: File,
/// Where the next chunk starts.
offset: u64,
/// Bytes of the window that no chunk has read. `None`: the window ends at EOF.
remaining: Option<u64>,
/// One buffer for every chunk. `chunk[..len]` holds the bytes of the chunk that is read.
chunk: Box<[u8]>,
len: usize,
/// A positional read moves the file pointer, and a duplicate shares it with the caller's descriptor.
#[cfg(windows)]
restore_position: bool,
}

pub enum Opened {
/// The first chunk is read.
Window(Box<FileWindow>),
/// The window holds no bytes.
Empty,
/// Only a regular file has a byte at an offset.
NotRegular,
}

impl FileWindow {
/// The most bytes of a window that the parent holds.
const CHUNK_SIZE: u64 = 64 * 1024;

/// `length` is `None` for a window that ends at EOF.
pub fn open_path(path: &ZStr, offset: u64, length: Option<u64>) -> sys::Result<Opened> {
// An open of a FIFO takes its writer out of `open(2)`, so the type is read first.
let stat = sys::stat(path)?;
if !sys::is_regular_file(stat.st_mode as _) {
return Ok(Opened::NotRegular);
}
if Self::is_empty(offset, length, stat.st_size.max(0) as u64) {
return Ok(Opened::Empty);
}
// The path can name a FIFO or a terminal by now. The open must not wait for a writer.
let flags = O::RDONLY | O::CLOEXEC | O::NONBLOCK | O::NOCTTY;
Self::with_first_chunk(File::open(path, flags, 0)?, offset, length, false)
}

/// `length` is `None` for a window that ends at EOF.
pub fn open_fd(fd: Fd, offset: u64, length: Option<u64>) -> sys::Result<Opened> {
let stat = sys::fstat(fd)?;
if !sys::is_regular_file(stat.st_mode as _) {
return Ok(Opened::NotRegular);
}
if Self::is_empty(offset, length, stat.st_size.max(0) as u64) {
return Ok(Opened::Empty);
}
// The caller can close `fd` before the last chunk is read.
let file = File::from_fd(sys::dup(fd)?);
Self::with_first_chunk(file, offset, length, true)
}

/// A size of 0 is not a length: procfs reports it for a file that has bytes.
fn is_empty(offset: u64, length: Option<u64>, size: u64) -> bool {
length == Some(0) || (size != 0 && offset >= size)
}

fn with_first_chunk(
file: File,
offset: u64,
length: Option<u64>,
is_duplicate: bool,
) -> sys::Result<Opened> {
#[cfg(not(windows))]
let _ = is_duplicate;
let capacity = length.unwrap_or(u64::MAX).min(Self::CHUNK_SIZE) as usize;
let mut window = Box::new(FileWindow {
file,
offset,
remaining: length,
chunk: vec![0; capacity].into_boxed_slice(),
len: 0,
#[cfg(windows)]
restore_position: is_duplicate,
});
Ok(match window.read_chunk()? {
0 => Opened::Empty,
_ => Opened::Window(window),
})
}

/// Returns the length of the new chunk: 0 at the end of the window or of the file.
fn read_chunk(&mut self) -> sys::Result<usize> {
self.len = 0;
let want = self
.remaining
.unwrap_or(u64::MAX)
.min(self.chunk.len() as u64) as usize;
if want == 0 {
return Ok(0);
}
#[cfg(windows)]
let position = if self.restore_position {
Some(self.file.get_pos()?)
} else {
None
};
let read = self.file.pread_all(&mut self.chunk[..want], self.offset);
#[cfg(windows)]
if let Some(position) = position {
self.file.seek_to(position)?;
}
let len = read?;
self.len = len;
self.offset += len as u64;
if let Some(remaining) = &mut self.remaining {
*remaining -= len as u64;
}
Ok(len)
}

pub(crate) fn slice(&self) -> &[u8] {
&self.chunk[..self.len]
}

/// Replaces the chunk with the next one. `Ok(false)`: the window or the file has no more bytes.
pub(crate) fn refill(&mut self) -> sys::Result<bool> {
Ok(self.read_chunk()? != 0)
}

pub(crate) fn memory_cost(&self) -> usize {
core::mem::size_of::<Self>() + self.chunk.len()
}
}
Loading
Loading