perf: add threading benchmark scenario (#2789)

* perf: add threading benchmark scenario

* fix unused variable

* use NoopWriter and simulate some work in the child spans

* fix type hinting

* fix missing type import

* add to main docs

* Add workflow to spelling list

* Update benchmarks/threading/README.rst

Co-authored-by: Gabriele N. Tornetta <[email protected]>

Co-authored-by: Gabriele N. Tornetta <[email protected]>

Author: brettlangdon

benchmarks/README.rst

@@ -69,3 +69,9 @@ Example::
 
   scripts/perf-run-scenario span ddtrace==0.50.0 ddtrace==0.51.0 ./artifacts/
   scripts/perf-run-scenario span Datadog/dd-trace-py@master Datadog/dd-trace-py@my-feature ./artifacts/
+
+
+Scenarios
+^^^^^^^^^
+
+.. include:: ../benchmarks/threading/README.rst

benchmarks/threading/README.rst

@@ -0,0 +1,10 @@
+threading
+~~~~~~~~~
+
+This benchmark test is used to simulate the creation, encoding, and flushing of traces in threaded environments.
+
+It uses a ``concurrent.futures.ThreadPool`` to manage the total number of workers.
+
+The only modification to the tracing workflow that has been made is using a ``NoopWriter`` which does not start a
+background thread and drops traces on ``writer.write``. This means we skip encoding, queuing, and flushing payloads
+to the agent, but we will still use the span processors.

benchmarks/threading/config.yaml

@@ -0,0 +1,13 @@
+1-thread: &baseline
+  nthreads: 1
+  ntraces: 1000
+  nspans: 10
+10-threads:
+  <<: *baseline
+  nthreads: 10
+50-threads:
+  <<: *baseline
+  nthreads: 50
+100-threads:
+  <<: *baseline
+  nthreads: 100

benchmarks/threading/scenario.py

@@ -0,0 +1,58 @@
+import concurrent.futures
+import random
+from typing import Callable
+from typing import Generator
+from typing import List
+from typing import Optional
+
+import bm
+
+from ddtrace.internal.writer import TraceWriter
+from ddtrace.span import Span
+from ddtrace.tracer import Tracer
+
+
+class NoopWriter(TraceWriter):
+    def recreate(self):
+        # type: () -> TraceWriter
+        return NoopWriter()
+
+    def stop(self, timeout=None):
+        # type: (Optional[float]) -> None
+        pass
+
+    def write(self, spans=None):
+        # type: (Optional[List[Span]]) -> None
+        pass
+
+
+@bm.register
+class Threading(bm.Scenario):
+    nthreads = bm.var(type=int)
+    ntraces = bm.var(type=int)
+    nspans = bm.var(type=int)
+
+    def create_trace(self, tracer):
+        # type: (Tracer) -> None
+        with tracer.trace("root"):
+            for _ in range(self.nspans - 1):
+                with tracer.trace("child"):
+                    # Simulate work in each child
+                    random.random()
+
+    def run(self):
+        # type: () -> Generator[Callable[[int], None], None, None]
+        from ddtrace import tracer
+
+        # configure global tracer to drop traces rather
+        tracer.configure(writer=NoopWriter())
+
+        def _(loops):
+            # type: (int) -> None
+            for _ in range(loops):
+                with concurrent.futures.ThreadPoolExecutor(max_workers=self.nthreads) as executor:
+                    tasks = {executor.submit(self.create_trace, tracer) for i in range(self.ntraces)}
+                    for task in concurrent.futures.as_completed(tasks):
+                        task.result()
+
+        yield _

docs/spelling_wordlist.txt

@@ -145,5 +145,6 @@ uvicorn
 versioned
 vertica
 whitelist
+workflow
 wsgi
 xfail

scripts/perf-build-push

@@ -11,7 +11,7 @@ shift
 
 IMAGES=()
 
-for SCENARIO in "encoder" "span" "tracer" "django_simple" "flask_simple"; do
+for SCENARIO in "encoder" "span" "tracer" "django_simple" "flask_simple" "threading"; do
     TAG="${REPOSITORY}/perf-${SCENARIO}"
     scripts/perf-build-scenario "${SCENARIO}" "${TAG}"
     IMAGE="${TAG}:latest"