perf(tracer): optimize span processors

.gitignore

@@ -18,6 +18,8 @@ ddtrace/internal/_encoding.c
 ddtrace/internal/_rand.c
 ddtrace/internal/_tagset.c
 ddtrace/internal/telemetry/metrics_namespaces.c
+ddtrace/_trace/processor/__init__.c
+ddtrace/internal/peer_service/processor.c
 pygoat/*
 *.so
 *.dylib

ddtrace/_trace/filters.py

@@ -6,7 +6,7 @@
 from ddtrace._trace.span import Span
 
 
-class TraceFilter(TraceProcessor):
+class TraceFilter(TraceProcessor, metaclass=abc.ABCMeta):
     @abc.abstractmethod
     def process_trace(self, trace: List[Span]) -> Optional[List[Span]]:
         """Processes a trace.

ddtrace/_trace/processor/__init__.pxd

@@ -0,0 +1,2 @@
+cdef class TraceProcessor:
+    cpdef object process_trace(self, list trace)

ddtrace/_trace/processor/__init__.pyi

@@ -0,0 +1,53 @@
+from threading import RLock
+from typing import Dict, List, Optional
+
+from ddtrace._trace.span import Span
+from ddtrace.internal.writer.writer import TraceWriter
+
+class TraceProcessor:
+    def process_trace(self, trace: List[Span]) -> Optional[List[Span]]: ...
+
+class SpanProcessor:
+    __processors__: List["SpanProcessor"]
+
+    def on_span_start(self, span: Span) -> None: ...
+    def on_span_finish(self, span: Span) -> None: ...
+    def shutdown(self, timeout: Optional[float]) -> None: ...
+    def register(self) -> None: ...
+    def unregister(self) -> None: ...
+
+class TraceSamplingProcessor(TraceProcessor): ...
+class TopLevelSpanProcessor(SpanProcessor): ...
+class ServiceNameProcessor(TraceProcessor): ...
+class TraceTagsProcessor(TraceProcessor): ...
+
+class SpanAggregator(SpanProcessor):
+    partial_flush_enabled: bool
+    partial_flush_min_spans: int
+    sampling_processor: TraceSamplingProcessor
+    tags_processor: TraceTagsProcessor
+    service_name_processor: ServiceNameProcessor
+    dd_processors: List[TraceProcessor]
+    user_processors: List[TraceProcessor]
+    writer: TraceWriter
+    _traces: Dict[int, List[Span]]
+    _spans_created: Dict[str, int]
+    _spans_finished: Dict[str, int]
+    _lock: RLock
+    _total_spans_finished: int
+
+    def __init__(
+        self,
+        partial_flush_enabled: bool,
+        partial_flush_min_spans: int,
+        dd_processors: Optional[List[TraceProcessor]] = None,
+        user_processors: Optional[List[TraceProcessor]] = None,
+    ) -> None: ...
+    def reset(
+        self,
+        user_processors: Optional[List[TraceProcessor]] = None,
+        compute_stats: Optional[bool] = None,
+        apm_opt_out: Optional[bool] = None,
+        appsec_enabled: Optional[bool] = None,
+        reset_buffer: bool = True,
+    ) -> None: ...