mirrors/bun.sh
1
0
Fork 0
mirror of https://github.com/oven-sh/bun synced 2026-02-12 11:59:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make node streams faster (#1502)

Browse Source 
* Make node streams faster

* Fix for macOS, improve performance, handle ref and unref

Co-authored-by: Jarred Sumner <709451+Jarred-Sumner@users.noreply.github.com>
This commit is contained in:
Jarred Sumner
2022-11-13 19:14:44 -08:00
committed by GitHub
parent 58b67347e6
commit b18e4064a2
23 changed files with 824 additions and 394 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
src/bun.js/bindings/JSReadableHelper.cpp
View File

@@ -11,7 +11,7 @@
#include "JSDOMAttribute.h"
#include "headers.h"
#include "JSDOMConvertEnumeration.h"
#include "JavaScriptCore/StrongInlines.h"
namespace WebCore {
using namespace JSC;
@@ -56,13 +56,15 @@ static bool callRead(JSValue stream, JSFunction* read, JSC::MarkedArgumentBuffer
return !ret.isUndefinedOrNull();
}
static JSC_DECLARE_HOST_FUNCTION(jsReadable_maybeReadMore_);
JSC_DEFINE_HOST_FUNCTION(jsReadable_maybeReadMore_, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
JSC_DEFINE_HOST_FUNCTION(jsReadable_maybeReadMore, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
{
JSReadableHelper_EXTRACT_STREAM_STATE
auto read
= stream->get(lexicalGlobalObject, Identifier::fromString(vm, "read"_s));
auto clientData
= WebCore::clientData(vm);
auto readIdentifier = clientData->builtinNames().readPublicName();
auto read = stream->get(lexicalGlobalObject, readIdentifier);
auto callData = JSC::getCallData(read);
if (callData.type == CallData::Type::None) {
throwException(lexicalGlobalObject, throwScope, createNotAFunctionError(lexicalGlobalObject, read));
@@ -85,24 +87,14 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_maybeReadMore_, (JSGlobalObject * lexicalGlo
RELEASE_AND_RETURN(throwScope, JSValue::encode(jsUndefined()));
}
JSC_DEFINE_HOST_FUNCTION(jsReadable_maybeReadMore, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
{
JSReadableHelper_EXTRACT_STREAM_STATE
// make this static?
JSFunction* maybeReadMore_
= JSC::JSFunction::create(vm, lexicalGlobalObject, 0, "maybeReadMore_"_s, jsReadable_maybeReadMore_, ImplementationVisibility::Public);
lexicalGlobalObject->queueMicrotask(maybeReadMore_, JSValue(stream), JSValue(state), JSValue {}, JSValue {});
RELEASE_AND_RETURN(throwScope, JSValue::encode(jsUndefined()));
}
void flow(JSGlobalObject* lexicalGlobalObject, JSObject* streamObj, JSReadableState* state)
{
VM& vm = lexicalGlobalObject->vm();
auto throwScope = DECLARE_THROW_SCOPE(vm);
auto read = streamObj->get(lexicalGlobalObject, Identifier::fromString(vm, "read"_s));
auto clientData = WebCore::clientData(vm);
auto readIdentifier = clientData->builtinNames().readPublicName();
auto read = streamObj->get(lexicalGlobalObject, readIdentifier);
auto callData = JSC::getCallData(read);
if (callData.type == CallData::Type::None) {
@@ -118,8 +110,7 @@ void flow(JSGlobalObject* lexicalGlobalObject, JSObject* streamObj, JSReadableSt
}
}
static JSC_DECLARE_HOST_FUNCTION(jsReadable_resume_);
JSC_DEFINE_HOST_FUNCTION(jsReadable_resume_, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
JSC_DEFINE_HOST_FUNCTION(jsReadable_resume, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
{
JSReadableHelper_EXTRACT_STREAM_STATE
@@ -132,18 +123,20 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_resume_, (JSGlobalObject * lexicalGlobalObje
}
auto& emitter = jsEmitterWrap->wrapped();
auto clientData = WebCore::clientData(vm);
auto readIdentifier = clientData->builtinNames().readPublicName();
if (!state->getBool(JSReadableState::reading)) {
// stream.read(0)
MarkedArgumentBuffer args;
args.append(jsNumber(0));
callRead(stream, jsCast<JSFunction*>(stream->get(lexicalGlobalObject, Identifier::fromString(vm, "read"_s))), WTFMove(args), vm, lexicalGlobalObject, emitter);
callRead(stream, jsCast<JSFunction*>(stream->get(lexicalGlobalObject, readIdentifier)), WTFMove(args), vm, lexicalGlobalObject, emitter);
}
state->setBool(JSReadableState::resumeScheduled, true);
// stream.emit('resume')
auto eventType = Identifier::fromString(vm, "resume"_s);
auto eventType = clientData->builtinNames().resumePublicName();
MarkedArgumentBuffer args;
emitter.emitForBindings(eventType, args);
@@ -152,7 +145,7 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_resume_, (JSGlobalObject * lexicalGlobalObje
if (state->m_flowing > 0 && !state->getBool(JSReadableState::reading)) {
// stream.read(0)
auto read = stream->get(lexicalGlobalObject, Identifier::fromString(vm, "read"_s));
auto read = stream->get(lexicalGlobalObject, readIdentifier);
auto callData = JSC::getCallData(read);
if (callData.type == CallData::Type::None) {
throwException(lexicalGlobalObject, throwScope, createNotAFunctionError(lexicalGlobalObject, read));
@@ -165,31 +158,16 @@ JSC_DEFINE_HOST_FUNCTION(jsReadable_resume_, (JSGlobalObject * lexicalGlobalObje
RELEASE_AND_RETURN(throwScope, JSValue::encode(jsUndefined()));
}
JSC_DEFINE_HOST_FUNCTION(jsReadable_resume, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
{
JSReadableHelper_EXTRACT_STREAM_STATE
if (!state->getBool(JSReadableState::resumeScheduled))
{
state->setBool(JSReadableState::resumeScheduled, true);
// make this static?
JSFunction* resume_ = JSC::JSFunction::create(
vm, lexicalGlobalObject, 0, "resume_"_s, jsReadable_resume_, ImplementationVisibility::Public);
lexicalGlobalObject->queueMicrotask(resume_, JSValue(stream), JSValue(state), JSValue {}, JSValue {});
}
RELEASE_AND_RETURN(throwScope, JSValue::encode(jsUndefined()));
}
EncodedJSValue emitReadable_(JSGlobalObject* lexicalGlobalObject, JSObject* stream, JSReadableState* state)
{
VM& vm = lexicalGlobalObject->vm();
auto throwScope = DECLARE_THROW_SCOPE(vm);
JSValue errored = state->m_errored.get();
if (!state->getBool(JSReadableState::destroyed) && !errored.toBoolean(lexicalGlobalObject) && (state->m_length || state->getBool(JSReadableState::ended))) {
// stream.emit('readable')
auto eventType = Identifier::fromString(vm, "readable"_s);
auto clientData = WebCore::clientData(vm);
auto eventType = clientData->builtinNames().readablePublicName();
MarkedArgumentBuffer args;
auto emitter = jsDynamicCast<JSEventEmitter*>(stream);
if (!emitter) {
@@ -206,7 +184,6 @@ EncodedJSValue emitReadable_(JSGlobalObject* lexicalGlobalObject, JSObject* stre
return JSValue::encode(jsUndefined());
}
JSC_DECLARE_HOST_FUNCTION(jsReadable_emitReadable_);
JSC_DEFINE_HOST_FUNCTION(jsReadable_emitReadable_, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame))
{
JSReadableHelper_EXTRACT_STREAM_STATE
@@ -223,11 +200,8 @@ EncodedJSValue emitReadable(JSGlobalObject* lexicalGlobalObject, JSObject* strea
state->setBool(JSReadableState::needReadable, false);
if (!state->getBool(JSReadableState::emittedReadable)) {
state->setBool(JSReadableState::emittedReadable, true);
// make this static?
JSFunction* emitReadable_ = JSC::JSFunction::create(
vm, lexicalGlobalObject, 0, "emitReadable_"_s, jsReadable_emitReadable_, ImplementationVisibility::Public);
lexicalGlobalObject->queueMicrotask(emitReadable_, JSValue(stream), JSValue(state), JSValue {}, JSValue {});
Zig::GlobalObject* globalObject = reinterpret_cast<Zig::GlobalObject*>(lexicalGlobalObject);
globalObject->queueMicrotask(JSValue(globalObject->emitReadableNextTickFunction()), JSValue(stream), JSValue(state), JSValue {}, JSValue {});
}
return JSValue::encode(jsUndefined());
}