refactor(profiling): add memalloc race regression test (#13026)

Add a regression test for races in the memory allocation profiler. The
test is marked skip for now, for a few reasons:

- It doesn't trigger the crash in a deterministic amount of time, so
  it's not really reasonable for CI/local dev loop as-is
- It probably benefits more from having the thread sanitizer enabled,
  which we don't currently do for the memalloc extension

I'm adding the test so that we have an actual reproducer of the problem
that we can easily run ourselves available to any dd-trace-py
developers, and have it actually committed somewhere people can find it.
It's currently only really useful for local development. I plan to
tweak/optimize some of the synchronization code to reduce memalloc
overhead, and we need a reliable reproducer of the crashes the
synchronization was meant to fix in order to be confident we don't
reintroduce them.

The test reproduces the crash fixed by #11460, as well as the exception
fixed by #12075. Both issues stem from the same problem: at one point,
memalloc had no synchronization beyond the GIL protecting its internal
state. It turns out that calling back into C Python APIs, as we do when
collecting tracebacks, can in some cases lead to the GIL being released.
So we need additional synchronization for state modification that
straddles C Python API calls. We previously only reliably saw this in a
demo program but weren't able to reproduce it locally. Now that I
understand the crash much better, I was able to create a standalone
reproducer. The key elements are: allocate a lot, trigger GC a lot
(including from memalloc traceback collection), and release the GIL
during GC.

Important note: this only reliably crashes on Python 3.11. The very
specific path to releasing the GIL that we hit was modified in 3.12 and
later (see python/cpython#97922). We will
probably support 3.11 for a while longer, so it's still worth having
this test.

Author: nsrip-dd

tests/profiling_v2/collector/test_memalloc.py

@@ -301,3 +301,75 @@ def test_heap_profiler_sampling_accuracy(sample_interval):
         # probably too generous for low sampling intervals but at least won't be
         # flaky.
         assert abs(rel - actual_rel) < 0.10
+
+
+@pytest.mark.skip(reason="too slow, indeterministic")
+@pytest.mark.subprocess(
+    env=dict(
+        # Turn off other profilers so that we're just testing memalloc
+        DD_PROFILING_STACK_ENABLED="false",
+        DD_PROFILING_LOCK_ENABLED="false",
+        # Upload a lot, since rotating out memalloc profiler state can race with profiling
+        DD_PROFILING_UPLOAD_INTERVAL="1",
+    ),
+)
+def test_memealloc_data_race_regression():
+    import gc
+    import threading
+    import time
+
+    from ddtrace.profiling import Profiler
+
+    gc.enable()
+    # This threshold is controls when garbage collection is triggered. The
+    # threshold is on the count of live allocations, which is checked when doing
+    # a new allocation. This test is ultimately trying to get the allocation of
+    # frame objects during the memory profiler's traceback function to trigger
+    # garbage collection. We want a lower threshold to improve the odds that
+    # this happens.
+    gc.set_threshold(100)
+
+    class Thing:
+        def __init__(self):
+            # Self reference so this gets deallocated in GC vs via refcount
+            self.ref = self
+
+        def __del__(self):
+            # Force GIL yield,  so if/when memalloc triggers GC, this is
+            # deallocated, releasing GIL while memalloc is sampling and allowing
+            # something else to run and possibly modify memalloc's internal
+            # state concurrently
+            time.sleep(0)
+
+    def do_alloc():
+        def f():
+            return Thing()
+
+        return f
+
+    def lotsa_allocs(ev):
+        while not ev.is_set():
+            f = do_alloc()
+            f()
+            time.sleep(0.01)
+
+    p = Profiler()
+    p.start()
+
+    threads = []
+    ev = threading.Event()
+    for i in range(4):
+        t = threading.Thread(target=lotsa_allocs, args=(ev,))
+        t.start()
+        threads.append(t)
+
+    # Arbitrary sleep. This typically crashes in about a minute.
+    # But for local development, either let it run way longer or
+    # figure out sanitizer instrumentation
+    time.sleep(120)
+
+    p.stop()
+
+    ev.set()
+    for t in threads:
+        t.join()