async_hooks: fixup do not reuse HTTPParser

Fix some issues introduced/not fixed via
https://github.com/nodejs/node/pull/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: https://github.com/nodejs/node/issues/27467
Fixes: https://github.com/nodejs/node/issues/26961
Refs: https://github.com/nodejs/node/pull/25094

PR-URL: https://github.com/nodejs/node/pull/27477
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

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_ =
+  async_id_ = execution_async_id == kInvalidAsyncId ? env()->new_async_id()
-    execution_async_id == -1 ? env()->new_async_id() : execution_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)
+  Parser(Environment* env, Local<Object> wrap)
-      : AsyncWrap(env, wrap,
+      : AsyncWrap(env, wrap),
-                  type == HTTP_REQUEST ?
-                      AsyncWrap::PROVIDER_HTTPINCOMINGMESSAGE :
-                      AsyncWrap::PROVIDER_HTTPCLIENTREQUEST),
         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());
+    new Parser(env, args.This());
-    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);
   }
 
 
@@ -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',
+      { type: 'HTTPCLIENTREQUEST',
-        id: 'httpparser:1',
+        id: 'httpclientrequest:1',
         triggerAsyncId: 'tcpserver:1' },
       { type: 'TCPWRAP', id: 'tcp:2', triggerAsyncId: 'tcpserver:1' },
       { type: 'Timeout', id: 'timeout:1', triggerAsyncId: 'tcp:2' },
-      { type: 'HTTPPARSER',
+      { type: 'HTTPINCOMINGMESSAGE',
-        id: 'httpparser:2',
+        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,28 +1,54 @@
 '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 reusedParser = [];
-const asyncHook = createHook({
+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(() => {
@@ -30,10 +56,21 @@ server.listen(PORT, 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];
 

test/async-hooks/verify-graph.js

@@ -98,6 +98,18 @@ module.exports = function verifyGraph(hooks, graph) {
     );
   }
   assert.strictEqual(errors.length, 0);
+
+  // Verify that all expected types are present
+  const expTypes = Object.create(null);
+  for (let i = 0; i < graph.length; i++) {
+    if (expTypes[graph[i].type] == null) expTypes[graph[i].type] = 0;
+    expTypes[graph[i].type]++;
+  }
+
+  for (const type in expTypes) {
+    assert.strictEqual(typeSeen[type], expTypes[type],
+                       `Expecting type '${type}' in graph`);
+  }
 };
 
 //

test/parallel/test-async-hooks-http-parser-destroy.js

@@ -16,7 +16,7 @@ const createdIds = [];
 const destroyedIds = [];
 async_hooks.createHook({
   init: common.mustCallAtLeast((asyncId, type) => {
-    if (type === 'HTTPPARSER') {
+    if (type === 'HTTPINCOMINGMESSAGE' || type === 'HTTPCLIENTREQUEST') {
       createdIds.push(asyncId);
     }
   }, N),
@@ -25,7 +25,7 @@ async_hooks.createHook({
   }
 }).enable();
 
-const server = http.createServer(function(req, res) {
+const server = http.createServer((req, res) => {
   res.end('Hello');
 });
 
@@ -39,6 +39,7 @@ const countdown = new Countdown(N, () => {
     // Give the server sockets time to close (which will also free their
     // associated parser objects) after the server has been closed.
     setTimeout(() => {
+      assert.strictEqual(createdIds.length, 2 * N);
       createdIds.forEach((createdAsyncId) => {
         assert.ok(destroyedIds.indexOf(createdAsyncId) >= 0);
       });

test/parallel/test-http-parser-bad-ref.js

@@ -24,7 +24,8 @@ function flushPool() {
 function demoBug(part1, part2) {
   flushPool();
 
-  const parser = new HTTPParser(HTTPParser.REQUEST);
+  const parser = new HTTPParser();
+  parser.initialize(HTTPParser.REQUEST, {});
 
   parser.headers = [];
   parser.url = '';

test/parallel/test-http-parser-lazy-loaded.js

@@ -7,7 +7,10 @@ const { getOptionValue } = require('internal/options');
 
 // Monkey patch before requiring anything
 class DummyParser {
-  constructor(type) {
+  constructor() {
+    this.test_type = null;
+  }
+  initialize(type) {
     this.test_type = type;
   }
 }
@@ -25,6 +28,7 @@ const { parsers } = require('_http_common');
 
 // Test _http_common was not loaded before monkey patching
 const parser = parsers.alloc();
+parser.initialize(DummyParser.REQUEST, {});
 assert.strictEqual(parser instanceof DummyParser, true);
 assert.strictEqual(parser.test_type, DummyParser.REQUEST);
 

test/parallel/test-http-parser.js

@@ -38,7 +38,8 @@ const kOnMessageComplete = HTTPParser.kOnMessageComplete | 0;
 
 
 function newParser(type) {
-  const parser = new HTTPParser(type);
+  const parser = new HTTPParser();
+  parser.initialize(type, {});
 
   parser.headers = [];
   parser.url = '';
@@ -95,7 +96,7 @@ function expectBody(expected) {
     throw new Error('hello world');
   };
 
-  parser.initialize(HTTPParser.REQUEST, request);
+  parser.initialize(REQUEST, {});
 
   assert.throws(
     () => { parser.execute(request, 0, request.length); },

test/sequential/test-async-wrap-getasyncid.js

@@ -152,7 +152,10 @@ if (common.hasCrypto) { // eslint-disable-line node-core/crypto-check
 
 {
   const { HTTPParser } = require('_http_common');
-  testInitialized(new HTTPParser(HTTPParser.REQUEST), 'HTTPParser');
+  const parser = new HTTPParser();
+  testUninitialized(parser, 'HTTPParser');
+  parser.initialize(HTTPParser.REQUEST, {});
+  testInitialized(parser, 'HTTPParser');
 }
 
 

test/sequential/test-http-regr-gh-2928.js

@@ -7,7 +7,6 @@ const common = require('../common');
 const assert = require('assert');
 const httpCommon = require('_http_common');
 const { HTTPParser } = require('_http_common');
-const { AsyncResource } = require('async_hooks');
 const net = require('net');
 
 const COUNT = httpCommon.parsers.max + 1;
@@ -25,7 +24,7 @@ function execAndClose() {
   process.stdout.write('.');
 
   const parser = parsers.pop();
-  parser.initialize(HTTPParser.RESPONSE, new AsyncResource('ClientRequest'));
+  parser.initialize(HTTPParser.RESPONSE, {});
 
   const socket = net.connect(common.PORT);
   socket.on('error', (e) => {