Add max concurrency when processing object props

Author: watson

integration-tests/debugger/snapshot-time-budget.spec.js

@@ -3,37 +3,86 @@
 const assert = require('node:assert')
 const { setup } = require('./utils')
 
-describe('Dynamic Instrumentation', function () {
-  // Force a very small time budget in ms to trigger partial snapshots
-  const t = setup({
-    dependencies: ['fastify'],
-    env: { DD_DYNAMIC_INSTRUMENTATION_CAPTURE_TIMEOUT_MS: '1' }
-  })
+const tolerance = 5
 
+describe('Dynamic Instrumentation', function () {
   describe('input messages', function () {
     describe('with snapshot under tight time budget', function () {
-      beforeEach(t.triggerBreakpoint)
-
-      it('should include partial snapshot marked with notCapturedReason: timeout', function (done) {
-        t.agent.on('debugger-input', ({ payload: [{ debugger: { snapshot: { captures } } }] }) => {
-          const { locals } = captures.lines[t.breakpoint.line]
-          assert.strictEqual(
-            containsTimeBudget(locals),
-            true,
-            'expected at least one field/element to be marked with notCapturedReason: "timeout"'
-          )
-          done()
+      context('1ms time budget', function () {
+        // Force a very small time budget in ms to trigger partial snapshots
+        const target = 1
+        const t = setup({
+          dependencies: ['fastify'],
+          env: { DD_DYNAMIC_INSTRUMENTATION_CAPTURE_TIMEOUT_MS: String(target) }
         })
 
-        t.agent.addRemoteConfig(t.generateRemoteConfig({
-          captureSnapshot: true,
-          capture: { maxReferenceDepth: 5 }
-        }))
+        it(
+          'should include partial snapshot marked with notCapturedReason: timeout',
+          test({ t, maxPausedTime: target + tolerance, breakpointIndex: 0, maxReferenceDepth: 5 })
+        )
+      })
+
+      context('default time budget', function () {
+        const target = 10 // default time budget in ms
+        const t = setup({ dependencies: ['fastify'] })
+
+        // TODO: Make this pass
+        // eslint-disable-next-line mocha/no-pending-tests
+        it.skip(
+          'should keep budget when state includes an object with 1 million properties',
+          test({ t, maxPausedTime: target + tolerance, breakpointIndex: 1, maxReferenceDepth: 1 })
+        )
+
+        // TODO: Make this pass
+        // eslint-disable-next-line mocha/no-pending-tests
+        it.skip(
+          'should keep budget when state includes an array of 1 million primitives',
+          test({ t, maxPausedTime: target + tolerance, breakpointIndex: 2, maxReferenceDepth: 1 })
+        )
+
+        // TODO: Make this pass
+        // eslint-disable-next-line mocha/no-pending-tests
+        it.skip(
+          'should keep budget when state includes an array of 1 million objects',
+          test({ t, maxPausedTime: target + tolerance, breakpointIndex: 3, maxReferenceDepth: 1 })
+        )
       })
     })
   })
 })
 
+function test ({ t, maxPausedTime, breakpointIndex, maxReferenceDepth }) {
+  const breakpoint = t.breakpoints[breakpointIndex]
+
+  return async function () {
+    const payloadReceived = new Promise((/** @type {(value?: unknown) => void} */ resolve) => {
+      t.agent.on('debugger-input', ({ payload: [{ debugger: { snapshot: { captures } } }] }) => {
+        const { locals } = captures.lines[breakpoint.line]
+        assert.strictEqual(
+          containsTimeBudget(locals),
+          true,
+          'expected at least one field/element to be marked with notCapturedReason: "timeout"'
+        )
+        resolve()
+      })
+    })
+
+    t.agent.addRemoteConfig(breakpoint.generateRemoteConfig({
+      captureSnapshot: true,
+      capture: { maxReferenceDepth }
+    }))
+
+    const { data } = await breakpoint.triggerBreakpoint()
+
+    assert.ok(
+      data.paused <= maxPausedTime,
+      `expected thread to be paused <=${maxPausedTime}ms, but was paused for ~${data.paused}ms`
+    )
+
+    await payloadReceived
+  }
+}
+
 function containsTimeBudget (node) {
   if (node == null || typeof node !== 'object') return false
   if (node.notCapturedReason === 'timeout') return true

integration-tests/debugger/snapshot.spec.js

@@ -209,12 +209,12 @@ describe('Dynamic Instrumentation', function () {
         t.agent.on('debugger-input', ({ payload: [{ debugger: { snapshot: { captures } } }] }) => {
           const { locals } = captures.lines[t.breakpoint.line]
 
-          assert.deepStrictEqual(Object.keys(locals), [
+          assert.deepStrictEqual(Object.keys(locals).sort(), [
             // Up to 3 properties from the closure scope
             'fastify', 'getUndefined',
             // Up to 3 properties from the local scope
             'request', 'nil', 'undef'
-          ])
+          ].sort())
 
           assert.strictEqual(locals.request.type, 'Request')
           assert.strictEqual(Object.keys(locals.request.fields).length, maxFieldCount)

integration-tests/debugger/target-app/snapshot-time-budget.js

@@ -6,13 +6,35 @@ const Fastify = require('fastify')
 
 const fastify = Fastify({ logger: { level: 'error' } })
 
-fastify.get('/:name', function handler (request) {
+fastify.get('/deeply-nested-large-object', function handler () {
   // The size of `obj` generated is carefully tuned to never result in a snapshot larger than the 1MB size limit, while
   // still being large enough to trigger the time budget limit. However, the size of `fastify` and `request` is not
   // stable across Node.js and Fastify versions, so the generated object might need to be adjusted.
-  const obj = generateObject(5, 6) // eslint-disable-line no-unused-vars
+  const obj = generateObject(5, 12) // eslint-disable-line no-unused-vars
+  const start = process.hrtime.bigint()
+  const diff = process.hrtime.bigint() - start // BREAKPOINT: /deeply-nested-large-object
+  return { paused: Number(diff) / 1_000_000 }
+})
+
+fastify.get('/large-object-of-primitives', function handler () {
+  const obj = generateObject(0, 1_000_000) // eslint-disable-line no-unused-vars
+  const start = process.hrtime.bigint()
+  const diff = process.hrtime.bigint() - start // BREAKPOINT: /large-object-of-primitives
+  return { paused: Number(diff) / 1_000_000 }
+})
+
+fastify.get('/large-array-of-primitives', function handler () {
+  const arr = Array.from({ length: 1_000_000 }, (_, i) => i) // eslint-disable-line no-unused-vars
+  const start = process.hrtime.bigint()
+  const diff = process.hrtime.bigint() - start // BREAKPOINT: /large-array-of-primitives
+  return { paused: Number(diff) / 1_000_000 }
+})
 
-  return { hello: request.params.name } // BREAKPOINT: /foo
+fastify.get('/large-array-of-objects', function handler () {
+  const arr = Array.from({ length: 1_000_000 }, (_, i) => ({ i })) // eslint-disable-line no-unused-vars
+  const start = process.hrtime.bigint()
+  const diff = process.hrtime.bigint() - start // BREAKPOINT: /large-array-of-objects
+  return { paused: Number(diff) / 1_000_000 }
 })
 
 fastify.listen({ port: process.env.APP_PORT || 0 }, (err) => {

integration-tests/debugger/utils.js

@@ -53,14 +53,16 @@ function setup ({ env, testApp, testAppSource, dependencies, silent, stdioHandle
   }
 
   // Trigger the breakpoint once probe is successfully installed
-  function triggerBreakpoint (url) {
+  async function triggerBreakpoint (url) {
     let triggered = false
-    t.agent.on('debugger-diagnostics', ({ payload }) => {
-      payload.forEach((event) => {
-        if (!triggered && event.debugger.diagnostics.status === 'INSTALLED') {
-          triggered = true
-          t.axios.get(url)
-        }
+    return new Promise((resolve, reject) => {
+      t.agent.on('debugger-diagnostics', ({ payload }) => {
+        payload.forEach((event) => {
+          if (!triggered && event.debugger.diagnostics.status === 'INSTALLED') {
+            triggered = true
+            t.axios.get(url).then(resolve).catch(reject)
+          }
+        })
       })
     })
   }

packages/dd-trace/src/debugger/devtools_client/index.js

@@ -167,8 +167,13 @@ session.on('Debugger.paused', async ({ params }) => {
   // TODO: Create unique states for each affected probe based on that probes unique `capture` settings (DEBUG-2863)
   const processLocalState = numberOfProbesWithSnapshots !== 0 && await getLocalStateForCallFrame(
     params.callFrames[0],
-    start + BigInt(config.dynamicInstrumentation.captureTimeoutMs ?? 10) * 1_000_000n,
-    { maxReferenceDepth, maxCollectionSize, maxFieldCount, maxLength }
+    {
+      maxReferenceDepth,
+      maxCollectionSize,
+      maxFieldCount,
+      maxLength,
+      deadlineNs: start + BigInt(config.dynamicInstrumentation.captureTimeoutMs ?? 10) * 1_000_000n
+    }
   )
 
   await session.post('Debugger.resume')

packages/dd-trace/src/debugger/devtools_client/snapshot/collector.js

@@ -10,6 +10,28 @@ module.exports = {
   getRuntimeObject: getObject // TODO: Called once per stack frame, but doesn't retain the `deadlineReached` flag.
 }
 
+/**
+ * @typedef {Object} GetObjectOptions
+ * @property {Object} maxReferenceDepth - The maximum depth of the object to traverse
+ * @property {number} maxCollectionSize - The maximum size of a collection to include in the snapshot
+ * @property {number} maxFieldCount - The maximum number of properties on an object to include in the snapshot
+ * @property {bigint} deadlineNs - The deadline in nanoseconds compared to `process.hrtime.bigint()`
+ * @property {boolean} [deadlineReached=false] - Whether the deadline has been reached. Should not be set by the
+ *   caller, but is used to signal the deadline overrun to the caller.
+ */
+
+/**
+ * Get the properties of an object using the Chrome DevTools Protocol.
+ *
+ * @param {string} objectId - The ID of the object to get the properties of
+ * @param {GetObjectOptions} opts - The options for the snapshot. Also used to track the deadline and communicate the
+ *   deadline overrun to the caller using the `deadlineReached` flag.
+ * @param {number} [depth=0] - The depth of the object. Only used internally by this module to track the current depth
+ *   and should not be set by the caller.
+ * @param {boolean} [collection=false] - Whether the object is a collection. Only used internally by this module to
+ *   track the current object type and should not be set by the caller.
+ * @returns {Promise<Object[]>} The properties of the object
+ */
 async function getObject (objectId, opts, depth = 0, collection = false) {
   const { result, privateProperties } = await session.post('Runtime.getProperties', {
     objectId,
@@ -43,30 +65,48 @@ async function traverseGetPropertiesResult (props, opts, depth) {
 
   if (depth >= opts.maxReferenceDepth) return props
 
-  const promises = []
+  const work = []
 
   for (const prop of props) {
     if (prop.value === undefined) continue
-    if (overBudget(opts)) {
-      prop.value[timeBudgetSym] = true
-      continue
-    }
     const { value: { type, objectId, subtype } } = prop
     if (type === 'object') {
       if (objectId === undefined) continue // if `subtype` is "null"
       if (LEAF_SUBTYPES.has(subtype)) continue // don't waste time with these subtypes
-      promises.push(getObjectProperties(subtype, objectId, opts, depth).then((properties) => {
-        prop.value.properties = properties
-      }))
+      work.push([
+        prop.value,
+        () => getObjectProperties(subtype, objectId, opts, depth).then((properties) => {
+          prop.value.properties = properties
+        })
+      ])
     } else if (type === 'function') {
-      promises.push(getFunctionProperties(objectId, opts, depth + 1).then((properties) => {
-        prop.value.properties = properties
-      }))
+      work.push([
+        prop.value,
+        () => getFunctionProperties(objectId, opts, depth + 1).then((properties) => {
+          prop.value.properties = properties
+        })
+      ])
     }
   }
 
-  if (promises.length) {
-    await Promise.all(promises)
+  if (work.length) {
+    // Iterate over the work in chunks of 2. The closer to 1, the less we'll overshoot the deadline, but the longer it
+    // takes to complete. `2` seems to be the best compromise.
+    // Anecdotally, on my machine, with no deadline, a concurrency of `1` takes twice as long as a concurrency of `2`.
+    // From thereon, there's no real measureable savings with a higher concurrency.
+    for (let i = 0; i < work.length; i += 2) {
+      if (overBudget(opts)) {
+        for (let j = i; j < work.length; j++) {
+          work[j][0][timeBudgetSym] = true
+        }
+        break
+      }
+      // eslint-disable-next-line no-await-in-loop
+      await Promise.all([
+        work[i][1](),
+        work[i + 1]?.[1]()
+      ])
+    }
   }
 
   return props

packages/dd-trace/src/debugger/devtools_client/snapshot/index.js

@@ -4,6 +4,8 @@ const { getRuntimeObject } = require('./collector')
 const { processRawState } = require('./processor')
 const log = require('../log')
 
+const BIGINT_MAX = (1n << 256n) - 1n
+
 const DEFAULT_MAX_REFERENCE_DEPTH = 3
 const DEFAULT_MAX_COLLECTION_SIZE = 100
 const DEFAULT_MAX_FIELD_COUNT = 20
@@ -17,27 +19,44 @@ function returnError () {
   return new Error('Error getting local state')
 }
 
-async function getLocalStateForCallFrame (
-  callFrame,
-  deadlineNs,
-  {
-    maxReferenceDepth = DEFAULT_MAX_REFERENCE_DEPTH,
-    maxCollectionSize = DEFAULT_MAX_COLLECTION_SIZE,
-    maxFieldCount = DEFAULT_MAX_FIELD_COUNT,
-    maxLength = DEFAULT_MAX_LENGTH
-  } = {}
-) {
+/**
+ * @typedef {Object} GetLocalStateForCallFrameOptions
+ * @property {number} [maxReferenceDepth=DEFAULT_MAX_REFERENCE_DEPTH] - The maximum depth of the object to traverse
+ * @property {number} [maxCollectionSize=DEFAULT_MAX_COLLECTION_SIZE] - The maximum size of a collection to include
+ *   in the snapshot
+ * @property {number} [maxFieldCount=DEFAULT_MAX_FIELD_COUNT] - The maximum number of properties on an object to
+ *   include in the snapshot
+ * @property {number} [maxLength=DEFAULT_MAX_LENGTH] - The maximum length of a string to include in the snapshot
+ * @property {bigint} [deadlineNs=BIGINT_MAX] - The deadline in nanoseconds compared to
+ *   `process.hrtime.bigint()`. If the deadline is reached, the snapshot will be truncated.
+ * @property {boolean} [deadlineReached=false] - Whether the deadline has been reached. Should not be set by the
+ *   caller, but is used to signal the deadline overrun to the caller.
+ */
+
+/**
+ * Get the local state for a call frame.
+ *
+ * @param {import('inspector').Debugger.CallFrame} callFrame - The call frame to get the local state for
+ * @param {GetLocalStateForCallFrameOptions} [opts={}] - The options for the snapshot
+ * @returns {Promise<Object>} The local state for the call frame
+ */
+async function getLocalStateForCallFrame (callFrame, opts = {}) {
+  opts.maxReferenceDepth ??= DEFAULT_MAX_REFERENCE_DEPTH
+  opts.maxCollectionSize ??= DEFAULT_MAX_COLLECTION_SIZE
+  opts.maxFieldCount ??= DEFAULT_MAX_FIELD_COUNT
+  opts.maxLength ??= DEFAULT_MAX_LENGTH
+  opts.deadlineNs ??= BIGINT_MAX
+  opts.deadlineReached ??= false
   const rawState = []
   let processedState = null
 
   try {
-    await Promise.all(callFrame.scopeChain.map(async (scope) => {
-      if (scope.type === 'global') return // The global scope is too noisy
-      rawState.push(...await getRuntimeObject(
-        scope.object.objectId,
-        { maxReferenceDepth, maxCollectionSize, maxFieldCount, deadlineNs }
-      ))
-    }))
+    for (const scope of callFrame.scopeChain) {
+      if (opts.deadlineReached === true) break // TODO: Variables in scope are silently dropped: Not the best UX
+      if (scope.type === 'global') continue // The global scope is too noisy
+      // eslint-disable-next-line no-await-in-loop
+      rawState.push(...await getRuntimeObject(scope.object.objectId, opts))
+    }
   } catch (err) {
     // TODO: We might be able to get part of the scope chain.
     // Consider if we could set errors just for the part of the scope chain that throws during collection.
@@ -47,7 +66,7 @@ async function getLocalStateForCallFrame (
 
   // Delay calling `processRawState` so the caller gets a chance to resume the main thread before processing `rawState`
   return () => {
-    processedState = processedState ?? processRawState(rawState, maxLength)
+    processedState = processedState ?? processRawState(rawState, opts.maxLength)
     return processedState
   }
 }

packages/dd-trace/test/debugger/devtools_client/snapshot/complex-types.spec.js

@@ -34,7 +34,6 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
 
         resolve((await getLocalStateForCallFrame(
           params.callFrames[0],
-          Infinity,
           { maxFieldCount: Number.MAX_SAFE_INTEGER })
         )())
       })

packages/dd-trace/test/debugger/devtools_client/snapshot/error-handling.spec.js

@@ -10,7 +10,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
   describe('error handling', function () {
     it('should generate a notCapturedReason if an error is thrown during inital collection', async function () {
       const invalidCallFrameThatTriggersAnException = {}
-      const processLocalState = await getLocalStateForCallFrame(invalidCallFrameThatTriggersAnException, Infinity)
+      const processLocalState = await getLocalStateForCallFrame(invalidCallFrameThatTriggersAnException)
       const result = processLocalState()
       assert.ok(result instanceof Error)
       assert.strictEqual(result.message, 'Error getting local state')

packages/dd-trace/test/debugger/devtools_client/snapshot/utils.js

@@ -99,7 +99,7 @@ function assertOnBreakpoint (done, snapshotConfig, callback) {
   session.once('Debugger.paused', ({ params }) => {
     assert.strictEqual(params.hitBreakpoints.length, 1)
 
-    getLocalStateForCallFrame(params.callFrames[0], Infinity, snapshotConfig).then((process) => {
+    getLocalStateForCallFrame(params.callFrames[0], snapshotConfig).then((process) => {
       callback(process())
       done()
     }).catch(done)