Skip to content
Open
Show file tree
Hide file tree
Changes from 4 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
21 changes: 21 additions & 0 deletions src/io/PipeReader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -536,11 +536,26 @@ impl PosixBufferedReader {
}

match poll.register_with_fd(lp.cast(), FilePollKind::Readable, poll.fd()) {
sys::Result::Err(err) if crate::pipes::is_unpollable(&err) => {
self.demote_to_unpollable();
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
Ok(())
}
sys::Result::Err(err) => Err(err),
sys::Result::Ok(()) => Ok(()),
}
}

/// Drops the poll the kernel refused and reads the fd synchronously, like
/// a regular file. No poll callback will come, so a caller that waits for
/// one must issue the read itself: see [`Self::is_pollable`].
Comment thread
robobun marked this conversation as resolved.
Resolved
fn demote_to_unpollable(&mut self) {
self.flags.remove(PosixFlags::POLLABLE);
let fd = self.handle.get_fd();
if let PollOrFd::Poll(poll) = mem::replace(&mut self.handle, PollOrFd::Fd(fd)) {
poll.deinit_force_unregister();
}
}
Comment thread
robobun marked this conversation as resolved.
Resolved

pub fn start(&mut self, fd: Fd, is_pollable: bool) -> sys::Result<()> {
if !is_pollable {
self.buffer().clear();
Expand Down Expand Up @@ -575,6 +590,12 @@ impl PosixBufferedReader {
self.limit = ReadLimit(len);
}

/// False after `start(fd, true)` when the kernel refused the poll. Reads
/// then run synchronously, and nothing drives them until the parent reads.
Comment thread
robobun marked this conversation as resolved.
Resolved
pub fn is_pollable(&self) -> bool {
self.flags.contains(PosixFlags::POLLABLE)
}

// Exists for consistently with Windows.
pub fn has_pending_read(&self) -> bool {
// `is_watching()` (registered && !needs-rearm) rather than
Expand Down
10 changes: 10 additions & 0 deletions src/io/pipes.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,16 @@ use bun_sys::FdExt;
use crate::FilePollFlag;
use crate::{FilePollRef, Owner};

/// The kernel refused to watch the fd because it has no poll support: a
/// regular file, or a character device such as `/dev/null`, `/dev/zero` or
/// `/dev/urandom`. `epoll_ctl` reports it as `EPERM`, kqueue as `EINVAL`.
/// A read or write on such an fd never waits, so the caller continues
/// without a poll.
Comment thread
robobun marked this conversation as resolved.
Resolved
#[cfg(not(windows))]
pub fn is_unpollable(err: &bun_sys::Error) -> bool {
Comment thread
robobun marked this conversation as resolved.
Resolved
matches!(err.get_errno(), bun_sys::E::EPERM | bun_sys::E::EINVAL)
Comment thread
robobun marked this conversation as resolved.
Resolved
}

pub enum PollOrFd {
Poll(FilePollRef),
Fd(Fd),
Expand Down
5 changes: 4 additions & 1 deletion src/runtime/server/FileResponseStream.rs
Original file line number Diff line number Diff line change
Expand Up @@ -566,7 +566,10 @@ impl FileResponseStream {
self.insert_state(State::RESPONSE_DONE);
self.detach_resp();
let resp = self.resp.get();
resp.end_without_body(resp.should_close_connection());
// The reader hit EOF before any write. `end` frames the empty body
// (`content-length: 0`, or the last chunk) so a keep-alive client
// does not wait for more.
Comment thread
robobun marked this conversation as resolved.
Resolved
resp.end(b"", resp.should_close_connection());
self.deliver(resp, StreamEnd::Complete);
// This end runs uncorked (reader callbacks), so no cork or parser
// gate will run the close check; do it here, after `on_complete`
Expand Down
3 changes: 3 additions & 0 deletions src/runtime/server/RequestContext.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2005,6 +2005,9 @@ where
offset: sendfile.offset as u64,
length: if is_regular {
Some(sendfile.remain as u64)
} else if original_size != crate::webcore::blob::MAX_SIZE {
// A `.slice()` bounds a device or pipe body. Without one the reader runs to EOF.
Some(original_size as u64)
} else {
None
},
Expand Down
7 changes: 7 additions & 0 deletions src/runtime/shell/IOReader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,13 @@ impl IOReader {
let fd = self.state().fd;
if let Err(e) = r.start(fd, true) {
self.on_reader_error(&e);
} else if !r.is_pollable() {
// The kernel refused the poll (a regular file, /dev/null):
// no callback will come, so read the fd now.
// SAFETY: the reader cell is live for `self`'s lifetime and
// `r` is not used past this point; `read` is the raw
// re-entrancy-safe entry (its dispatch runs shell code).
unsafe { bun_io::BufferedReader::read(self.reader.get()) };
Comment thread
robobun marked this conversation as resolved.
Resolved
}
}
return Yield::suspended();
Expand Down
20 changes: 4 additions & 16 deletions src/runtime/shell/IOWriter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -356,32 +356,20 @@ impl IOWriter {
if let Err(e) = s.writer.start(s.fd, s.flags.pollable) {
#[cfg(not(windows))]
{
// We get this if we pass in a file descriptor that is not
// pollable, for example a special character device like
// /dev/null. If so, restart with polling disabled.
// A regular file or a character device such as /dev/null:
// restart with polling disabled.
Comment thread
robobun marked this conversation as resolved.
Resolved
//
// It's also possible on Linux for EINVAL to be returned
// when registering multiple writable/readable polls for the
// same file descriptor. The shell code here makes sure to
// _not_ run into that case, but it is possible.
if e.get_errno() == E::EINVAL {
crate::shell_log!("IOWriter(fd={}) got EINVAL", s.fd);
if bun_io::pipes::is_unpollable(&e) {
crate::shell_log!("IOWriter(fd={}) got {}", s.fd, e.get_errno());
s.flags.pollable = false;
s.flags.nonblock = false;
s.flags.is_socket = false;
return self.__start();
}
#[cfg(any(target_os = "linux", target_os = "android"))]
{
// On linux regular files are not pollable and return EPERM,
// so restart if that's the case with polling disabled.
if e.get_errno() == E::EPERM {
s.flags.pollable = false;
s.flags.nonblock = false;
s.flags.is_socket = false;
return self.__start();
}
}
}
#[cfg(windows)]
{
Expand Down
13 changes: 13 additions & 0 deletions src/runtime/webcore/FileReader.rs
Original file line number Diff line number Diff line change
Expand Up @@ -411,6 +411,19 @@ impl FileReader {
}
return streams::Start::Err(e);
}
#[cfg(unix)]
{
use bun_io::pipe_reader::PosixFlags;
// The kernel refused to poll the fd (a character device such as
// `/dev/zero`), so the reader is synchronous from here on and
// the ref would root an abandoned stream forever.
Comment thread
robobun marked this conversation as resolved.
Resolved
if need_io_ref && !self.reader().flags.contains(PosixFlags::POLLABLE) {
self.waiting_for_on_reader_done.set(false);
let parent = self.parent();
// SAFETY: see `parent()`; JS finalizer still holds a ref so this cannot free it.
let _ = unsafe { Source::decrement_count(parent) };
}
}
} else {
#[cfg(unix)]
{
Expand Down
59 changes: 58 additions & 1 deletion test/js/bun/http/bun-serve-file.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import type { Server } from "bun";
import { afterAll, beforeAll, describe, expect, it, mock, test } from "bun:test";
import { bunEnv, bunExe, isASAN, isWindows, rmScope, rss, tempDir, tempDirWithFiles } from "harness";
import { bunEnv, bunExe, isASAN, isLinux, isWindows, rmScope, rss, tempDir, tempDirWithFiles } from "harness";
import { mkfifo } from "mkfifo";
import { closeSync, openSync, unlinkSync, writeSync } from "node:fs";
import { join } from "node:path";
Expand Down Expand Up @@ -1225,6 +1225,63 @@ test.skipIf(isWindows)("Response(Bun.file(FIFO)) frames the body as chunked, not
}
});

// The kernel refuses to poll these character devices (epoll_ctl EPERM). The
// server must read them synchronously, not fail the response.
describe.skipIf(isWindows)("Response(Bun.file(<character device>))", () => {
it("/dev/null ends as an empty body a keep-alive client can frame", async () => {
await using server = Bun.serve({
port: 0,
hostname: "127.0.0.1",
fetch() {
return new Response(Bun.file("/dev/null"));
},
});
const res = await fetch(`http://127.0.0.1:${server.port}/`);
expect({
status: res.status,
contentLength: res.headers.get("content-length"),
body: (await res.arrayBuffer()).byteLength,
}).toEqual({ status: 200, contentLength: "0", body: 0 });
});

it("/dev/urandom slice delivers the slice", async () => {
await using server = Bun.serve({
port: 0,
hostname: "127.0.0.1",
fetch() {
return new Response(Bun.file("/dev/urandom").slice(0, 200_000));
},
});
const res = await fetch(`http://127.0.0.1:${server.port}/`);
expect(res.status).toBe(200);
expect((await res.arrayBuffer()).byteLength).toBe(200_000);
});
});

// A FIFO whose writer closes without a byte ends the reader at EOF before the
// first write. The response must still frame its empty body. Linux only: the
// server does not see a FIFO hangup on macOS yet.
test.skipIf(!isLinux)("Response(Bun.file(FIFO)) with no data frames an empty body", async () => {
using dir = tempDir("serve-fifo-empty", {});
const fifoPath = join(String(dir), "body.fifo");
mkfifo(fifoPath);

await using server = Bun.serve({
port: 0,
hostname: "127.0.0.1",
fetch() {
// The writer opens the FIFO, which lets the server's open proceed, and closes it at once.
Bun.spawn({ cmd: ["sh", "-c", `: > "${fifoPath}"`], env: bunEnv });
Comment thread
robobun marked this conversation as resolved.
Resolved
return new Response(Bun.file(fifoPath));
},
});
const res = await fetch(`http://127.0.0.1:${server.port}/`);
expect({
status: res.status,
body: (await res.arrayBuffer()).byteLength,
}).toEqual({ status: 200, body: 0 });
});

// A file route serves the window of the Bun.file() slice it was built from,
// given either the slice or its unread stream (which is turned back into the
// slice). FileRoute used to clamp the window to the file size without taking
Expand Down
47 changes: 47 additions & 0 deletions test/js/web/streams/streams.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -2055,6 +2055,53 @@ describe.skipIf(isWindows)("Bun.file().stream() surfaces read() errors", () => {
});
});

// epoll and kqueue refuse a character device with no poll support. The reader
// must read such a device synchronously, not report the refusal as an error.
describe.skipIf(isWindows)("Bun.file().stream() on a character device", () => {
it.each(["/dev/zero", "/dev/urandom"])("%s streams the sliced range", async path => {
let total = 0;
for await (const chunk of Bun.file(path)
.slice(0, 1 << 20)
.stream()) {
total += chunk.length;
}
expect(total).toBe(1 << 20);
});

it("/dev/null streams to a clean end", async () => {
const chunks = [];
for await (const chunk of Bun.file("/dev/null").stream()) chunks.push(chunk);
expect(chunks).toHaveLength(0);
});

it("Response(Bun.file(device).slice()).body delivers the slice", async () => {
let total = 0;
for await (const chunk of new Response(Bun.file("/dev/zero").slice(0, 1024)).body) {
total += chunk.length;
}
expect(total).toBe(1024);
});

it("a stream left mid-read does not keep the process alive", async () => {
await using proc = Bun.spawn({
cmd: [
bunExe(),
"-e",
`
const reader = Bun.file("/dev/urandom").stream().getReader();
const { value } = await reader.read();
console.log("read", value.length > 0);
`,
],
env: bunEnv,
stderr: "pipe",
});
const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
expect(stdout).toBe("read true\n");
expect(exitCode).toBe(0);
Comment thread
robobun marked this conversation as resolved.
Resolved
});
});

it("fs.createReadStream(filename) should be able to break inside async loop", async () => {
for (let i = 0; i < 10; i++) {
const fileStream = createReadStream(join(import.meta.dir, "..", "fetch", "fixture.png"));
Expand Down
Loading