async_hooks: fixup do not reuse HTTPParser

Fix some issues introduced/not fixed via
#25094:
* Init hook is not emitted for a reused HTTPParser
* HTTPParser was still used as resource in init hook
* type used in init hook was always HTTPINCOMINGMESSAGE even for client
requests
* some tests have not been adapted to new resource names

With this change the async hooks init event is emitted during a call
to Initialize() as the type and resource object is available at this
time. As a result Initialize() must be called now which could be seen
as breaking change even HTTPParser is not part of documented API.

It was needed to put the ClientRequest instance into a wrapper object
instead passing it directly as async resource otherwise
test-domain-multi fails. I think this is because adding an EventEmitter
to a Domain adds a property 'domain' and the presence of this changes
the context propagation in domains.

Besides that tests still refering to resource HTTPParser have been
updated/improved.

Fixes: #27467
Fixes: #26961
Refs: #25094

PR-URL: #27477
Reviewed-By: Anna Henningsen <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>
Reviewed-By: Rich Trott <[email protected]>

Author: Flarna

benchmark/http/bench-parser.js

@@ -24,13 +24,14 @@ function main({ len, n }) {
     bench.start();
     for (var i = 0; i < n; i++) {
       parser.execute(header, 0, header.length);
-      parser.initialize(REQUEST, header);
+      parser.initialize(REQUEST, {});
     }
     bench.end(n);
   }
 
   function newParser(type) {
-    const parser = new HTTPParser(type);
+    const parser = new HTTPParser();
+    parser.initialize(type, {});
 
     parser.headers = [];
 

lib/_http_client.js

@@ -64,6 +64,13 @@ function validateHost(host, name) {
   return host;
 }
 
+class HTTPClientAsyncResource {
+  constructor(type, req) {
+    this.type = type;
+    this.req = req;
+  }
+}
+
 let urlWarningEmitted = false;
 function ClientRequest(input, options, cb) {
   OutgoingMessage.call(this);
@@ -635,7 +642,8 @@ function tickOnSocket(req, socket) {
   const parser = parsers.alloc();
   req.socket = socket;
   req.connection = socket;
-  parser.initialize(HTTPParser.RESPONSE, req);
+  parser.initialize(HTTPParser.RESPONSE,
+                    new HTTPClientAsyncResource('HTTPINCOMINGMESSAGE', req));
   parser.socket = socket;
   parser.outgoing = req;
   req.parser = parser;

lib/_http_common.js

@@ -156,7 +156,7 @@ function parserOnMessageComplete() {
 
 
 const parsers = new FreeList('parsers', 1000, function parsersCb() {
-  const parser = new HTTPParser(HTTPParser.REQUEST);
+  const parser = new HTTPParser();
 
   cleanParser(parser);
 

src/async_wrap.cc

@@ -178,7 +178,7 @@ void AsyncWrap::EmitAfter(Environment* env, double async_id) {
 class PromiseWrap : public AsyncWrap {
  public:
   PromiseWrap(Environment* env, Local<Object> object, bool silent)
-      : AsyncWrap(env, object, PROVIDER_PROMISE, -1, silent) {
+      : AsyncWrap(env, object, PROVIDER_PROMISE, kInvalidAsyncId, silent) {
     MakeWeak();
   }
 
@@ -388,7 +388,7 @@ static void RegisterDestroyHook(const FunctionCallbackInfo<Value>& args) {
 
 void AsyncWrap::GetAsyncId(const FunctionCallbackInfo<Value>& args) {
   AsyncWrap* wrap;
-  args.GetReturnValue().Set(-1);
+  args.GetReturnValue().Set(kInvalidAsyncId);
   ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
   args.GetReturnValue().Set(wrap->get_async_id());
 }
@@ -415,10 +415,15 @@ void AsyncWrap::AsyncReset(const FunctionCallbackInfo<Value>& args) {
   AsyncWrap* wrap;
   ASSIGN_OR_RETURN_UNWRAP(&wrap, args.Holder());
   double execution_async_id =
-      args[0]->IsNumber() ? args[0].As<Number>()->Value() : -1;
+      args[0]->IsNumber() ? args[0].As<Number>()->Value() : kInvalidAsyncId;
   wrap->AsyncReset(execution_async_id);
 }
 
+void AsyncWrap::EmitDestroy() {
+  AsyncWrap::EmitDestroy(env(), async_id_);
+  // Ensure no double destroy is emitted via AsyncReset().
+  async_id_ = kInvalidAsyncId;
+}
 
 void AsyncWrap::QueueDestroyAsyncId(const FunctionCallbackInfo<Value>& args) {
   CHECK(args[0]->IsNumber());
@@ -481,7 +486,7 @@ void AsyncWrap::Initialize(Local<Object> target,
   // kDefaultTriggerAsyncId: Write the id of the resource responsible for a
   //   handle's creation just before calling the new handle's constructor.
   //   After the new handle is constructed kDefaultTriggerAsyncId is set back
-  //   to -1.
+  //   to kInvalidAsyncId.
   FORCE_SET_TARGET_FIELD(target,
                          "async_id_fields",
                          env->async_hooks()->async_id_fields().GetJSArray());
@@ -569,10 +574,16 @@ AsyncWrap::AsyncWrap(Environment* env,
   AsyncReset(execution_async_id, silent);
 }
 
+AsyncWrap::AsyncWrap(Environment* env, v8::Local<v8::Object> object)
+    : BaseObject(env, object),
+      provider_type_(PROVIDER_NONE) {
+  CHECK_GE(object->InternalFieldCount(), 1);
+}
+
 
 AsyncWrap::~AsyncWrap() {
   EmitTraceEventDestroy();
-  EmitDestroy(env(), get_async_id());
+  EmitDestroy();
 }
 
 void AsyncWrap::EmitTraceEventDestroy() {
@@ -612,16 +623,18 @@ void AsyncWrap::AsyncReset(double execution_async_id, bool silent) {
 // the resource is pulled out of the pool and put back into use.
 void AsyncWrap::AsyncReset(Local<Object> resource, double execution_async_id,
                            bool silent) {
-  if (async_id_ != -1) {
+  CHECK_NE(provider_type(), PROVIDER_NONE);
+
+  if (async_id_ != kInvalidAsyncId) {
     // This instance was in use before, we have already emitted an init with
     // its previous async_id and need to emit a matching destroy for that
     // before generating a new async_id.
-    EmitDestroy(env(), async_id_);
+    EmitDestroy();
   }
 
   // Now we can assign a new async_id_ to this instance.
-  async_id_ =
-    execution_async_id == -1 ? env()->new_async_id() : execution_async_id;
+  async_id_ = execution_async_id == kInvalidAsyncId ? env()->new_async_id()
+                                                     : execution_async_id;
   trigger_async_id_ = env()->get_default_trigger_async_id();
 
   switch (provider_type()) {

src/async_wrap.h

@@ -109,12 +109,18 @@ class AsyncWrap : public BaseObject {
   AsyncWrap(Environment* env,
             v8::Local<v8::Object> object,
             ProviderType provider,
-            double execution_async_id = -1);
+            double execution_async_id = kInvalidAsyncId);
+
+  // This constructor creates a reuseable instance where user is responsible
+  // to call set_provider_type() and AsyncReset() before use.
+  AsyncWrap(Environment* env, v8::Local<v8::Object> object);
 
   ~AsyncWrap() override;
 
   AsyncWrap() = delete;
 
+  static constexpr double kInvalidAsyncId = -1;
+
   static v8::Local<v8::FunctionTemplate> GetConstructorTemplate(
       Environment* env);
 
@@ -141,6 +147,8 @@ class AsyncWrap : public BaseObject {
   static void EmitAfter(Environment* env, double async_id);
   static void EmitPromiseResolve(Environment* env, double async_id);
 
+  void EmitDestroy();
+
   void EmitTraceEventBefore();
   static void EmitTraceEventAfter(ProviderType type, double async_id);
   void EmitTraceEventDestroy();
@@ -155,10 +163,11 @@ class AsyncWrap : public BaseObject {
   inline double get_trigger_async_id() const;
 
   void AsyncReset(v8::Local<v8::Object> resource,
-                  double execution_async_id = -1,
+                  double execution_async_id = kInvalidAsyncId,
                   bool silent = false);
 
-  void AsyncReset(double execution_async_id = -1, bool silent = false);
+  void AsyncReset(double execution_async_id = kInvalidAsyncId,
+                  bool silent = false);
 
   // Only call these within a valid HandleScope.
   v8::MaybeLocal<v8::Value> MakeCallback(const v8::Local<v8::Function> cb,
@@ -210,7 +219,7 @@ class AsyncWrap : public BaseObject {
             bool silent);
   ProviderType provider_type_;
   // Because the values may be Reset(), cannot be made const.
-  double async_id_ = -1;
+  double async_id_ = kInvalidAsyncId;
   double trigger_async_id_;
 };
 

src/node_http_parser_impl.h

@@ -154,14 +154,10 @@ struct StringPtr {
 
 class Parser : public AsyncWrap, public StreamListener {
  public:
-  Parser(Environment* env, Local<Object> wrap, parser_type_t type)
-      : AsyncWrap(env, wrap,
-                  type == HTTP_REQUEST ?
-                      AsyncWrap::PROVIDER_HTTPINCOMINGMESSAGE :
-                      AsyncWrap::PROVIDER_HTTPCLIENTREQUEST),
+  Parser(Environment* env, Local<Object> wrap)
+      : AsyncWrap(env, wrap),
         current_buffer_len_(0),
         current_buffer_data_(nullptr) {
-    Init(type);
   }
 
 
@@ -426,11 +422,7 @@ class Parser : public AsyncWrap, public StreamListener {
 
   static void New(const FunctionCallbackInfo<Value>& args) {
     Environment* env = Environment::GetCurrent(args);
-    CHECK(args[0]->IsInt32());
-    parser_type_t type =
-        static_cast<parser_type_t>(args[0].As<Int32>()->Value());
-    CHECK(type == HTTP_REQUEST || type == HTTP_RESPONSE);
-    new Parser(env, args.This(), type);
+    new Parser(env, args.This());
   }
 
 
@@ -443,14 +435,13 @@ class Parser : public AsyncWrap, public StreamListener {
 
 
   static void Free(const FunctionCallbackInfo<Value>& args) {
-    Environment* env = Environment::GetCurrent(args);
     Parser* parser;
     ASSIGN_OR_RETURN_UNWRAP(&parser, args.Holder());
 
     // Since the Parser destructor isn't going to run the destroy() callbacks
     // it needs to be triggered manually.
     parser->EmitTraceEventDestroy();
-    parser->EmitDestroy(env, parser->get_async_id());
+    parser->EmitDestroy();
   }
 
 
@@ -526,6 +517,7 @@ class Parser : public AsyncWrap, public StreamListener {
             : AsyncWrap::PROVIDER_HTTPCLIENTREQUEST);
 
     parser->set_provider_type(provider);
+    parser->AsyncReset(args[1].As<Object>());
     parser->Init(type);
   }
 

test/async-hooks/coverage.md

@@ -9,7 +9,8 @@ Showing which kind of async resource is covered by which test:
 | FSREQCALLBACK        | test-fsreqcallback-{access,readFile}.js    |
 | GETADDRINFOREQWRAP   | test-getaddrinforeqwrap.js                 |
 | GETNAMEINFOREQWRAP   | test-getnameinforeqwrap.js                 |
-| HTTPPARSER           | test-httpparser.{request,response}.js      |
+| HTTPINCOMINGMESSAGE  | test-httpparser.request.js                 |
+| HTTPCLIENTREQUEST    | test-httpparser.response.js                |
 | Immediate            | test-immediate.js                          |
 | JSSTREAM             | TODO (crashes when accessing directly)     |
 | PBKDF2REQUEST        | test-crypto-pbkdf2.js                      |

test/async-hooks/test-graph.http.js

@@ -35,13 +35,13 @@ process.on('exit', function() {
       { type: 'TCPCONNECTWRAP',
         id: 'tcpconnect:1',
         triggerAsyncId: 'tcp:1' },
-      { type: 'HTTPPARSER',
-        id: 'httpparser:1',
+      { type: 'HTTPCLIENTREQUEST',
+        id: 'httpclientrequest:1',
         triggerAsyncId: 'tcpserver:1' },
       { type: 'TCPWRAP', id: 'tcp:2', triggerAsyncId: 'tcpserver:1' },
       { type: 'Timeout', id: 'timeout:1', triggerAsyncId: 'tcp:2' },
-      { type: 'HTTPPARSER',
-        id: 'httpparser:2',
+      { type: 'HTTPINCOMINGMESSAGE',
+        id: 'httpincomingmessage:1',
         triggerAsyncId: 'tcp:2' },
       { type: 'Timeout',
         id: 'timeout:2',

test/async-hooks/test-graph.tls-write.js

@@ -64,12 +64,8 @@ function onexit() {
         id: 'getaddrinforeq:1', triggerAsyncId: 'tls:1' },
       { type: 'TCPCONNECTWRAP',
         id: 'tcpconnect:1', triggerAsyncId: 'tcp:1' },
-      { type: 'WRITEWRAP', id: 'write:1', triggerAsyncId: 'tcpconnect:1' },
       { type: 'TCPWRAP', id: 'tcp:2', triggerAsyncId: 'tcpserver:1' },
       { type: 'TLSWRAP', id: 'tls:2', triggerAsyncId: 'tcpserver:1' },
-      { type: 'WRITEWRAP', id: 'write:2', triggerAsyncId: null },
-      { type: 'WRITEWRAP', id: 'write:3', triggerAsyncId: null },
-      { type: 'WRITEWRAP', id: 'write:4', triggerAsyncId: null },
       { type: 'Immediate', id: 'immediate:1', triggerAsyncId: 'tcp:2' },
       { type: 'Immediate', id: 'immediate:2', triggerAsyncId: 'tcp:1' },
     ]

test/async-hooks/test-httparser-reuse.js

@@ -1,39 +1,76 @@
 'use strict';
 
 const common = require('../common');
-const http = require('http');
 const assert = require('assert');
 const { createHook } = require('async_hooks');
+const http = require('http');
+
+// Verify that resource emitted for an HTTPParser is not reused.
+// Verify that correct create/destroy events are emitted.
+
 const reused = Symbol('reused');
 
-let reusedHTTPParser = false;
-const asyncHook = createHook({
+const reusedParser = [];
+const incomingMessageParser = [];
+const clientRequestParser = [];
+const dupDestroys = [];
+const destroyed = [];
+
+createHook({
   init(asyncId, type, triggerAsyncId, resource) {
+    switch (type) {
+      case 'HTTPINCOMINGMESSAGE':
+        incomingMessageParser.push(asyncId);
+        break;
+      case 'HTTPCLIENTREQUEST':
+        clientRequestParser.push(asyncId);
+        break;
+    }
+
     if (resource[reused]) {
-      reusedHTTPParser = true;
+      reusedParser.push(
+        `resource reused: ${asyncId}, ${triggerAsyncId}, ${type}`
+      );
     }
     resource[reused] = true;
+  },
+  destroy(asyncId) {
+    if (destroyed.includes(asyncId)) {
+      dupDestroys.push(asyncId);
+    } else {
+      destroyed.push(asyncId);
+    }
   }
-});
-asyncHook.enable();
+}).enable();
 
-const server = http.createServer(function(req, res) {
+const server = http.createServer((req, res) => {
   res.end();
 });
 
 const PORT = 3000;
-const url = 'http://127.0.0.1:' + PORT;
+const url = `http://127.0.0.1:${PORT}`;
 
 server.listen(PORT, common.mustCall(() => {
   http.get(url, common.mustCall(() => {
     server.close(common.mustCall(() => {
       server.listen(PORT, common.mustCall(() => {
         http.get(url, common.mustCall(() => {
           server.close(common.mustCall(() => {
-            assert.strictEqual(reusedHTTPParser, false);
+            setTimeout(common.mustCall(verify), 200);
           }));
         }));
       }));
     }));
   }));
 }));
+
+function verify() {
+  assert.strictEqual(reusedParser.length, 0);
+
+  assert.strictEqual(incomingMessageParser.length, 2);
+  assert.strictEqual(clientRequestParser.length, 2);
+
+  assert.strictEqual(dupDestroys.length, 0);
+  incomingMessageParser.forEach((id) => assert.ok(destroyed.includes(id)));
+  clientRequestParser.forEach((id) => assert.ok(destroyed.includes(id)));
+}

test/async-hooks/test-httpparser.request.js

@@ -20,7 +20,8 @@ const request = Buffer.from(
   'GET /hello HTTP/1.1\r\n\r\n'
 );
 
-const parser = new HTTPParser(REQUEST);
+const parser = new HTTPParser();
+parser.initialize(REQUEST, {});
 const as = hooks.activitiesOfTypes('HTTPINCOMINGMESSAGE');
 const httpparser = as[0];
 

test/async-hooks/test-httpparser.response.js

@@ -25,7 +25,8 @@ const request = Buffer.from(
   'pong'
 );
 
-const parser = new HTTPParser(RESPONSE);
+const parser = new HTTPParser();
+parser.initialize(RESPONSE, {});
 const as = hooks.activitiesOfTypes('HTTPCLIENTREQUEST');
 const httpparser = as[0];