Make node streams faster (#1502)

* 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>

1 files changed, 21 insertions, 47 deletions

diff --git a/src/bun.js/bindings/JSReadableHelper.cpp b/src/bun.js/bindings/JSReadableHelper.cpp
index 384664ce9..3845d62f5 100644
--- a/src/bun.js/bindings/JSReadableHelper.cpp
+++ b/src/bun.js/bindings/JSReadableHelper.cpp
@@ -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());
 }