Add private Windows ETW thread spawn tracer prototype

Introduce threadpoolctl._thread_tracer as an experimental private package
with a ctypes-based WindowsThreadSpawnTracer that counts kernel Thread
Start events for a target PID via the NT Kernel Logger.

Includes pure-Python event payload parsing, mocked ETW lifecycle tests that
run on all platforms, and keeps the tracer out of the public threadpoolctl API.

Co-authored-by: Olivier Grisel <olivier.grisel@ensta.org>

Author: cursoragent

threadpoolctl/_thread_tracer/__init__.py

@@ -0,0 +1,19 @@
+"""Private experimental thread spawn tracer (not part of the public API)."""
+
+import sys
+
+from threadpoolctl._thread_tracer._types import ThreadSpawnStats, ThreadTracerError
+from threadpoolctl._thread_tracer._windows_etw import WindowsThreadSpawnTracer
+
+
+def windows_tracer_available():
+    """Return whether the Windows ETW tracer can run on this platform."""
+    return sys.platform == "win32"
+
+
+__all__ = [
+    "ThreadSpawnStats",
+    "ThreadTracerError",
+    "WindowsThreadSpawnTracer",
+    "windows_tracer_available",
+]

threadpoolctl/_thread_tracer/_parsing.py

@@ -0,0 +1,68 @@
+"""Pure-Python helpers to classify kernel thread ETW events."""
+
+import ctypes as ct
+import struct
+
+# Kernel thread event opcodes (EVENT_TRACE_TYPE_* for the Thread task).
+EVENT_TRACE_TYPE_START = 1
+EVENT_TRACE_TYPE_END = 2
+EVENT_TRACE_TYPE_DCSTART = 3
+EVENT_TRACE_TYPE_DCEND = 4
+
+_MIN_USER_DATA_LENGTH = 8
+
+
+def parse_kernel_thread_payload(user_data, user_data_length):
+    """Extract ``(process_id, thread_id)`` from a kernel Thread event payload.
+
+    The first two fields are stable across Thread event schema versions. Pointer
+    fields that follow are ignored because only the identifiers are needed.
+    """
+    if user_data is None or user_data_length < _MIN_USER_DATA_LENGTH:
+        return None
+
+    if isinstance(user_data, (bytes, bytearray)):
+        raw = user_data[:user_data_length]
+    else:
+        address = user_data
+        if not isinstance(address, int):
+            address = ct.cast(user_data, ct.c_void_p).value or 0
+        if not address:
+            return None
+        raw = ct.string_at(address, user_data_length)
+
+    if len(raw) < _MIN_USER_DATA_LENGTH:
+        return None
+    process_id, thread_id = struct.unpack_from("<II", raw, 0)
+    return process_id, thread_id
+
+
+def classify_kernel_thread_event(opcode, user_data, user_data_length, target_pid):
+    """Return a spawn classification for a kernel Thread event.
+
+    Returns one of:
+    - ``"spawn"`` for a thread start event in ``target_pid``
+    - ``"existing"`` for a DCStart rundown event in ``target_pid``
+    - ``None`` if the event should be ignored
+    """
+    if opcode not in (
+        EVENT_TRACE_TYPE_START,
+        EVENT_TRACE_TYPE_END,
+        EVENT_TRACE_TYPE_DCSTART,
+        EVENT_TRACE_TYPE_DCEND,
+    ):
+        return None
+
+    parsed = parse_kernel_thread_payload(user_data, user_data_length)
+    if parsed is None:
+        return None
+
+    process_id, _thread_id = parsed
+    if process_id != target_pid:
+        return None
+
+    if opcode == EVENT_TRACE_TYPE_START:
+        return "spawn"
+    if opcode == EVENT_TRACE_TYPE_DCSTART:
+        return "existing"
+    return None

threadpoolctl/_thread_tracer/_types.py

@@ -0,0 +1,32 @@
+"""Private types for the experimental thread spawn tracer."""
+
+
+class ThreadSpawnStats(object):
+    """Statistics collected during a thread tracing window."""
+
+    __slots__ = ("spawn_count", "existing_thread_count", "thread_ids")
+
+    def __init__(self, spawn_count=0, existing_thread_count=0, thread_ids=None):
+        self.spawn_count = spawn_count
+        self.existing_thread_count = existing_thread_count
+        self.thread_ids = frozenset(thread_ids or ())
+
+    def __eq__(self, other):
+        if not isinstance(other, ThreadSpawnStats):
+            return NotImplemented
+        return (
+            self.spawn_count == other.spawn_count
+            and self.existing_thread_count == other.existing_thread_count
+            and self.thread_ids == other.thread_ids
+        )
+
+    def __repr__(self):
+        return (
+            "ThreadSpawnStats(spawn_count={0.spawn_count}, "
+            "existing_thread_count={0.existing_thread_count}, "
+            "thread_ids={0.thread_ids!r})"
+        ).format(self)
+
+
+class ThreadTracerError(OSError):
+    """Raised when the platform tracer cannot start or stop cleanly."""