feat(runtime)!: add shared runtime

.github/CODEOWNERS

@@ -50,6 +50,7 @@ libdd-library-config*/               @DataDog/apm-sdk-capabilities-rust
 libdd-libunwind*/                    @DataDog/libdatadog-profiling
 libdd-log*/                          @DataDog/apm-common-components-core
 libdd-profiling*/                    @DataDog/libdatadog-profiling
+libdd-shared-runtime*/               @DataDog/apm-common-components-core
 libdd-telemetry*/                    @DataDog/apm-common-components-core
 libdd-tinybytes                      @DataDog/apm-common-components-core
 libdd-trace-normalization            @DataDog/serverless @DataDog/libdatadog-apm

Cargo.toml

@@ -41,6 +41,8 @@ members = [
   "spawn_worker",
   "tests/spawn_from_lib",
   "bin_tests",
+  "libdd-shared-runtime",
+  "libdd-shared-runtime-ffi",
   "libdd-data-pipeline",
   "libdd-data-pipeline-ffi",
   "libdd-ddsketch",

libdd-common/src/lib.rs

@@ -38,7 +38,6 @@ pub mod threading;
 #[cfg(not(target_arch = "wasm32"))]
 pub mod timeout;
 pub mod unix_utils;
-pub mod worker;
 
 /// Extension trait for `Mutex` to provide a method that acquires a lock, panicking if the lock is
 /// poisoned.

libdd-data-pipeline-ffi/Cargo.toml

@@ -31,6 +31,7 @@ libdd-trace-utils = { path = "../libdd-trace-utils" }
 [dependencies]
 libdd-capabilities-impl = { version = "0.1.0", path = "../libdd-capabilities-impl" }
 libdd-data-pipeline = { path = "../libdd-data-pipeline" }
+libdd-shared-runtime = { version = "1.0.0", path = "../libdd-shared-runtime" }
 libdd-common-ffi = { path = "../libdd-common-ffi", default-features = false }
 libdd-tinybytes = { path = "../libdd-tinybytes" }
 tracing = { version = "0.1", default-features = false }

libdd-data-pipeline-ffi/cbindgen.toml

@@ -40,4 +40,4 @@ must_use = "DDOG_CHECK_RETURN"
 
 [parse]
 parse_deps = true
-include = ["libdd-common", "libdd-common-ffi", "libdd-data-pipeline"]
+include = ["libdd-common", "libdd-common-ffi", "libdd-shared-runtime", "libdd-data-pipeline"]

libdd-data-pipeline-ffi/src/trace_exporter.rs

@@ -9,15 +9,15 @@ use libdd_common_ffi::{
     CharSlice,
     {slice::AsBytes, slice::ByteSlice},
 };
-
 use libdd_data_pipeline::trace_exporter::{
     TelemetryConfig, TraceExporter as GenericTraceExporter, TraceExporterInputFormat,
     TraceExporterOutputFormat,
 };
 
 type TraceExporter = GenericTraceExporter<NativeCapabilities>;
 
-use std::{ptr::NonNull, time::Duration};
+use libdd_shared_runtime::SharedRuntime;
+use std::{ptr::NonNull, sync::Arc, time::Duration};
 use tracing::debug;
 
 #[inline]
@@ -73,6 +73,7 @@ pub struct TraceExporterConfig {
     process_tags: Option<String>,
     test_session_token: Option<String>,
     connection_timeout: Option<u64>,
+    shared_runtime: Option<Arc<SharedRuntime>>,
     otlp_endpoint: Option<String>,
 }
 
@@ -420,6 +421,36 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_connection_timeout(
     )
 }
 
+/// Sets a shared runtime for the TraceExporter to use for background workers.
+///
+/// `handle` must have been initialized with [`ddog_shared_runtime_new`].
+///
+/// When set, the exporter will use the provided runtime instead of creating its own.
+/// This allows multiple exporters (or other components) to share a single runtime.
+/// The config holds a clone of the `Arc` (increments the strong count), so the
+/// original handle remains valid and must still be freed with
+/// [`ddog_shared_runtime_free`].
+#[no_mangle]
+pub unsafe extern "C" fn ddog_trace_exporter_config_set_shared_runtime(
+    config: Option<&mut TraceExporterConfig>,
+    handle: Option<NonNull<SharedRuntime>>,
+) -> Option<Box<ExporterError>> {
+    catch_panic!(
+        match (config, handle) {
+            (Some(config), Some(handle)) => {
+                // SAFETY: handle was produced by Arc::into_raw and the Arc is still alive.
+                // Increment the strong count before reconstructing so the config's Arc
+                // is independent from the caller's handle.
+                Arc::increment_strong_count(handle.as_ptr());
+                config.shared_runtime = Some(Arc::from_raw(handle.as_ptr()));
+                None
+            }
+            _ => gen_error!(ErrorCode::InvalidArgument),
+        },
+        gen_error!(ErrorCode::Panic)
+    )
+}
+
 /// Enables OTLP HTTP/JSON export and sets the endpoint URL.
 ///
 /// When set, traces are sent to this URL in OTLP HTTP/JSON format instead of the Datadog
@@ -502,6 +533,10 @@ pub unsafe extern "C" fn ddog_trace_exporter_new(
                 builder.enable_health_metrics();
             }
 
+            if let Some(runtime) = config.shared_runtime.clone() {
+                builder.set_shared_runtime(runtime);
+            }
+
             if let Some(ref url) = config.otlp_endpoint {
                 builder.set_otlp_endpoint(url);
             }

libdd-data-pipeline/Cargo.toml

@@ -14,6 +14,7 @@ autobenches = false
 [dependencies]
 anyhow = { version = "1.0" }
 arc-swap = "1.7.1"
+async-trait = "0.1"
 http = "1"
 http-body-util = "0.1"
 tracing = { version = "0.1", default-features = false }
@@ -30,7 +31,8 @@ uuid = { version = "1.10.0", features = ["v4"] }
 tokio-util = "0.7.11"
 libdd-capabilities = { path = "../libdd-capabilities", version = "0.1.0" }
 libdd-common = { version = "3.0.2", path = "../libdd-common", default-features = false }
-libdd-telemetry = { version = "4.0.0", path = "../libdd-telemetry", default-features = false, optional = true }
+libdd-shared-runtime = { version = "1.0.0", path = "../libdd-shared-runtime" }
+libdd-telemetry = { version = "4.0.0", path = "../libdd-telemetry", default-features = false, optional = true}
 libdd-trace-protobuf = { version = "3.0.1", path = "../libdd-trace-protobuf" }
 libdd-trace-stats = { version = "2.0.0", path = "../libdd-trace-stats" }
 libdd-trace-utils = { version = "3.0.1", path = "../libdd-trace-utils", default-features = false }

libdd-data-pipeline/examples/send-traces-with-stats.rs

@@ -9,9 +9,11 @@ use libdd_data_pipeline::trace_exporter::{
 use libdd_log::logger::{
     logger_configure_std, logger_set_log_level, LogEventLevel, StdConfig, StdTarget,
 };
+use libdd_shared_runtime::SharedRuntime;
 use libdd_trace_protobuf::pb;
 use std::{
     collections::HashMap,
+    sync::Arc,
     time::{Duration, UNIX_EPOCH},
 };
 
@@ -54,6 +56,8 @@ fn main() {
     .expect("Failed to configure logger");
     logger_set_log_level(LogEventLevel::Debug).expect("Failed to set log level");
 
+    let shared_runtime = Arc::new(SharedRuntime::new().expect("Failed to create runtime"));
+
     let args = Args::parse();
     let telemetry_cfg = TelemetryConfig::default();
     let mut builder = TraceExporter::<NativeCapabilities>::builder();
@@ -68,6 +72,7 @@ fn main() {
         .set_language_version(env!("CARGO_PKG_RUST_VERSION"))
         .set_input_format(TraceExporterInputFormat::V04)
         .set_output_format(TraceExporterOutputFormat::V04)
+        .set_shared_runtime(shared_runtime.clone())
         .enable_telemetry(telemetry_cfg)
         .enable_stats(Duration::from_secs(10));
     let exporter = builder
@@ -89,7 +94,7 @@ fn main() {
     let data = rmp_serde::to_vec_named(&traces).expect("Failed to serialize traces");
 
     exporter.send(data.as_ref()).expect("Failed to send traces");
-    exporter
+    shared_runtime
         .shutdown(None)
-        .expect("Failed to shutdown exporter");
+        .expect("Failed to shutdown runtime");
 }