Skip to content

ByteStream: hand off owned buffers from on_pull/on_data and skip the adapter pull view #36736

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
robobun wants to merge 15 commits into
base: main
Choose a base branch
from
Open

ByteStream: hand off owned buffers from on_pull/on_data and skip the adapter pull view #36736

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
3be437a
ByteStream: hand off owned buffers from on_pull/on_data and skip the …
robobun Aug 1, 2026
cb20309
Add test asserting request-body chunks are right-sized; drop dead Byt…
robobun Aug 1, 2026
3a36bdb
[autofix.ci] apply automated fixes
autofix-ci[bot] Aug 1, 2026
ecb6ac6
Trim new code comments to one line each
robobun Aug 1, 2026
3088550
native-readable: skip the pull Buffer.alloc entirely for ReadyOwned s…
robobun Aug 1, 2026
c306505
Drop ByteStream.pending_value and wire pullParkedP reject in the test
robobun Aug 1, 2026
3bfb59f
Initialize kSourceOwnsChunks in constructNativeReadable; assert chunk…
robobun Aug 1, 2026
5600b4d
Test: replace write-pacing sleep with a per-chunk ack from the handler
robobun Aug 1, 2026
546bf74
Test: mark every ack/handler promise observed so a failure surfaces once
robobun Aug 1, 2026
8537aa0
Drop the m_sourceOwnsChunks drain-path disjunct
robobun Aug 1, 2026
2adce47
ci: retrigger
robobun Aug 1, 2026
fe6bc93
Keep the view-copy path for native-readable.ts; Owned handoff only fo…
robobun Aug 2, 2026
41ebfb8
Trim code comments to satisfy comment-cop
robobun Aug 2, 2026
6829d2f
One-line the on_pull dispatch comment
robobun Aug 2, 2026
e179877
on_cancel: gate pending.run on pending.state, not on a stored view
robobun Aug 2, 2026
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
6 changes: 5 additions & 1 deletion src/js/internal/streams/native-readable.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ const kHighWaterMark = Symbol("highWaterMark");
const kPendingRead = Symbol("pendingRead");
const kHasResized = Symbol("hasResized");
const kRemainingChunk = Symbol("remainingChunk");
const kSourceOwnsChunks = Symbol("sourceOwnsChunks");
Comment thread
robobun marked this conversation as resolved.
Outdated

const MIN_BUFFER_SIZE = 512;
let dynamicallyAdjustChunkSize = (_?) => (
Expand Down Expand Up @@ -142,6 +143,9 @@ function read(this: NativeReadable, maxToRead: number) {
if (typeof result === "number" && result > 1) {
this[kHasResized] = true;
this[kHighWaterMark] = Math.min(this[kHighWaterMark], result);
} else if (typeof result === "number" && result < 0) {
this[kHasResized] = true;
this[kSourceOwnsChunks] = true;
}
Comment thread
robobun marked this conversation as resolved.
if ($isTypedArrayView(result) && result.byteLength > 0) {
pushAndCheck(this, result);
Expand All @@ -153,7 +157,7 @@ function read(this: NativeReadable, maxToRead: number) {
pushAndCheck(this, drainResult);
}
}
const chunk = getRemainingChunk(this, maxToRead);
const chunk = this[kSourceOwnsChunks] ? undefined : getRemainingChunk(this, maxToRead);
var result = ptr.pull(chunk, this[kCloseState]);
$assert(result !== undefined);
$debug(
Expand Down
20 changes: 14 additions & 6 deletions src/jsc/bindings/webcore/streams/BunStreamSource.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -417,6 +417,8 @@ static JSC::JSUint8Array* uint8Subarray(JSGlobalObject* globalObject, JSC::JSUin
static JSC::JSUint8Array* nativeGetInternalBuffer(JSC::VM& vm, JSGlobalObject* globalObject, JSNativeStreamSourceAdapter* adapter)
{
auto scope = DECLARE_THROW_SCOPE(vm);
if (adapter->m_sourceOwnsChunks)
return nullptr;
const size_t chunkSize = adapter->m_chunkSize;
if (JSObject* pending = adapter->pendingView()) {
auto* view = uncheckedDowncast<JSC::JSUint8Array>(pending);
Expand Down Expand Up @@ -478,8 +480,6 @@ static JSValue nativeDecodePullResult(JSC::VM& vm, JSGlobalObject* globalObject,
return jsUndefined();
}
if (auto* chunk = dynamicDowncast<JSC::JSArrayBufferView>(result)) {
if (!isClosed)
nativeAdjustChunkSize(adapter, chunk->byteLength());
if (chunk->byteLength() > 0) {
if (adapter->m_textMode) {
nativeEnqueueTextChunk(globalObject, controller, adapter->m_textState, chunk->span(), /* flush */ false);
Expand Down Expand Up @@ -556,7 +556,13 @@ void materializeNativeSource(JSGlobalObject* globalObject, JSReadableStream* str
auto* adapter = WebCore::JSNativeStreamSourceAdapter::create(vm, runtime->nativeStreamSourceAdapterStructure(domGlobalObject));
adapter->setHandle(vm, handle);
adapter->m_textMode = stream->m_nativeTextMode;
adapter->m_chunkSize = std::max(static_cast<size_t>(chunkSize), autoAllocateChunkSize);
if (chunkSize < 0) {
adapter->m_sourceOwnsChunks = true;
adapter->m_hasResized = true;
adapter->m_chunkSize = 0;
} else {
adapter->m_chunkSize = std::max(static_cast<size_t>(chunkSize), autoAllocateChunkSize);
}
auto* closer = JSC::constructEmptyArray(globalObject, nullptr, 1);
RETURN_IF_EXCEPTION(scope, );
closer->putDirectIndex(globalObject, 0, jsBoolean(false));
Expand Down Expand Up @@ -625,7 +631,8 @@ static JSPromise* nativeSourcePullImpl(JSC::VM& vm, JSGlobalObject* globalObject
closer->putDirectIndex(globalObject, 0, jsBoolean(false));
RETURN_IF_EXCEPTION(scope, nullptr);

if (JSObject* pendingObject = adapter->pendingView()) {
JSObject* pendingObject = adapter->pendingView();
if (pendingObject || adapter->m_sourceOwnsChunks) {
Comment thread
robobun marked this conversation as resolved.
Outdated
MarkedArgumentBuffer noArgs;
JSValue drained = invokeMethod(vm, globalObject, handle, builtinNames(vm).drainPublicName(), noArgs);
RETURN_IF_EXCEPTION(scope, nullptr);
Expand All @@ -634,7 +641,8 @@ static JSPromise* nativeSourcePullImpl(JSC::VM& vm, JSGlobalObject* globalObject
if (isTruthy) {
bool isClosed = nativeCloserFlag(vm, globalObject, adapter);
RETURN_IF_EXCEPTION(scope, nullptr);
JSValue newView = nativeDecodePullResult(vm, globalObject, adapter, controller, drained, uncheckedDowncast<JSC::JSUint8Array>(pendingObject), isClosed);
JSC::JSUint8Array* pendingView = pendingObject ? uncheckedDowncast<JSC::JSUint8Array>(pendingObject) : nullptr;
JSValue newView = nativeDecodePullResult(vm, globalObject, adapter, controller, drained, pendingView, isClosed);
RETURN_IF_EXCEPTION(scope, nullptr);
nativeStorePendingView(vm, adapter, newView);
return nullptr;
Expand All @@ -645,7 +653,7 @@ static JSPromise* nativeSourcePullImpl(JSC::VM& vm, JSGlobalObject* globalObject
RETURN_IF_EXCEPTION(scope, nullptr);

MarkedArgumentBuffer pullArgs;
pullArgs.append(view);
pullArgs.append(view ? JSValue(view) : jsUndefined());
pullArgs.append(closer);
ASSERT(!pullArgs.hasOverflowed());
JSValue result = invokeMethod(vm, globalObject, handle, builtinNames(vm).pullPublicName(), pullArgs);
Expand Down
2 changes: 2 additions & 0 deletions src/jsc/bindings/webcore/streams/BunStreamSource.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,8 @@ class JSNativeStreamSourceAdapter final : public JSC::JSInternalFieldObjectImpl<
bool m_closed : 1 { false };
// Body.textStream(): each pulled byte span is UTF-8-decoded before enqueue.
bool m_textMode : 1 { false };
// start() returned <0 (Start::ReadyOwned): never allocate PendingView.
bool m_sourceOwnsChunks : 1 { false };
Bun::WebStreams::StreamingUTF8DecodeState m_textState;

private:
Expand Down
2 changes: 1 addition & 1 deletion src/runtime/server/RequestContext.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4063,7 +4063,7 @@ where
let _ = bytes.on_data(WebCore::streams::Result::Temporary(borrowed));

// What `on_data` buffered; `on_stream_drained` resumes once it empties.
let buffered = bytes.buffer.get().len().saturating_sub(bytes.offset.get());
let buffered = bytes.buffer.get().len();
if bytes.buffer_action.get().is_some()
|| (bytes.sink.get().is_some() && !bytes.sink_paused.get())
{
Expand Down
2 changes: 0 additions & 2 deletions src/runtime/webcore/Body.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -884,7 +884,6 @@ impl Value {

match drain_result {
DrainResult::EstimatedSize(estimated_size) => {
reader.context.high_water_mark = estimated_size as blob::SizeType;
reader
.context
.size_hint
Expand Down Expand Up @@ -1509,7 +1508,6 @@ impl Value {

match drain_result {
DrainResult::EstimatedSize(estimated_size) => {
reader.context.high_water_mark = estimated_size as blob::SizeType;
reader
.context
.size_hint
Expand Down
Loading
Loading