feat(native-spans)!: change buffer foundation

libdd-trace-utils/Cargo.toml

@@ -45,6 +45,7 @@ libdd-tinybytes = { version = "1.1.1", path = "../libdd-tinybytes", features = [
     "serialization",
 ] }
 indexmap = "2.11"
+rustc-hash = "2"
 
 # Compression feature
 flate2 = { version = "1.0", optional = true }
@@ -86,6 +87,7 @@ test-utils = [
     "cargo-platform",
     "urlencoding",
 ]
+change-buffer = []
 compression = ["zstd", "flate2"]
 # FIPS mode uses the FIPS-compliant cryptographic provider (Unix only)
 fips = ["libdd-common/fips", "libdd-capabilities-impl/fips"]

libdd-trace-utils/build.rs

@@ -0,0 +1,10 @@
+fn main() {
+    let commit = std::process::Command::new("git")
+        .args(["rev-parse", "--short", "HEAD"])
+        .output()
+        .ok()
+        .and_then(|o| String::from_utf8(o.stdout).ok())
+        .map(|s| s.trim().to_string())
+        .unwrap_or_else(|| "unknown".to_string());
+    println!("cargo:rustc-env=GIT_COMMIT={commit}");
+}

libdd-trace-utils/src/change_buffer/buffer.rs

@@ -0,0 +1,195 @@
+use crate::change_buffer::utils::*;
+use crate::change_buffer::{ChangeBufferError, Result};
+
+/// A handle to a change buffer shared with another runtime. The memory is shared, meaning that
+/// cloning is cheap (copying the pointer and length).
+#[derive(Clone, Copy)]
+pub struct ChangeBuffer {
+    ptr: *mut u8,
+    len: usize,
+}
+
+impl ChangeBuffer {
+    /// # Safety
+    ///
+    /// The underlying raw memory must be valid for reads and writes.
+    ///
+    /// The underlying raw memory must not be freed until after this struct's
+    /// lifetime. Having the calling code manage the memory makes it simpler to
+    /// integrate with managed runtimes.
+    pub unsafe fn from_raw_parts(ptr: *const u8, len: usize) -> Self {
+        Self {
+            ptr: ptr as *mut u8,
+            len,
+        }
+    }
+
+    /// # Safety
+    ///
+    /// Same safety conditions as [std::slice::from_raw_parts].
+    unsafe fn as_slice(&self) -> &[u8] {
+        unsafe { std::slice::from_raw_parts(self.ptr, self.len) }
+    }
+
+    /// # Safety
+    ///
+    /// Same safety conditions as [std::slice::from_raw_parts_mut].
+    unsafe fn as_mut_slice(&mut self) -> &mut [u8] {
+        unsafe { std::slice::from_raw_parts_mut(self.ptr, self.len) }
+    }
+
+    /// Read a value of type `T` starting at offset `index`.
+    pub fn read<T: Copy + FromBytes>(&self, index: &mut usize) -> Result<T> {
+        let size = std::mem::size_of::<T>();
+        // Safety: the allocation of `self.ptr` is required to be valid for read and writes at
+        // construction time, and to remain alive for the lifetime of `self`. We do not materialize
+        // other references during the lifetime of `slice`.
+        let slice = unsafe { self.as_slice() };
+        let out_of_bounds_err = || ChangeBufferError::ReadOutOfBounds {
+            offset: *index,
+            value_len: size,
+            buffer_len: self.len,
+        };
+        let Some(end) = index.checked_add(size) else {
+            return Err(out_of_bounds_err());
+        };
+
+        let bytes = slice
+            .get(*index..*index + size)
+            .ok_or_else(out_of_bounds_err)?;
+        *index += size;
+        Ok(T::from_bytes(bytes))
+    }
+
+    /// Read a value without bounds checking.
+    ///
+    /// # Safety
+    ///
+    /// Caller must ensure `*index + size_of::<T>() <= self.len` and that `index + size_of<T>::() <
+    /// usize::MAX`.
+    #[inline(always)]
+    unsafe fn read_unchecked<T: Copy + FromBytes>(&self, index: &mut usize) -> T {
+        let size = std::mem::size_of::<T>();
+        // Safety: the allocation of `self.ptr` is guaranteed to be valid for read and writes at
+        // construction time. We do not materialize other references during the lifetime of `slice`.
+        let slice = unsafe { self.as_slice() };
+        let bytes = slice.get_unchecked(*index..*index + size);
+        *index += size;
+        T::from_bytes(bytes)
+    }
+
+    /// Write a raw `u32` in the buffer.
+    pub fn write_u32(&mut self, offset: usize, value: u32) -> Result<()> {
+        let len = self.len;
+        // Safety: the allocation of `self.ptr` is guaranteed to be valid for read and writes at
+        // construction time. We do not materialize other references during the lifetime of `slice`.
+        let slice = unsafe { self.as_mut_slice() };
+        let out_of_bounds_err = || ChangeBufferError::WriteOutOfBounds {
+            offset,
+            value_len: offset + 4,
+            buffer_len: len,
+        };
+        let Some(end) = offset.checked_add(4) else {
+            return Err(out_of_bounds_err());
+        };
+
+        let target = slice.get_mut(offset..end).ok_or_else(out_of_bounds_err)?;
+        let bytes = value.to_le_bytes();
+        target.copy_from_slice(&bytes);
+        Ok(())
+    }
+
+    /// Clear the op count, which is stored in the first 4 bytes of the buffer. This effectively
+    /// reset the buffer (semantically), but without actually zeroing the rest.
+    pub fn clear_count(&mut self) -> Result<()> {
+        self.write_u32(0, 0)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::ChangeBuffer;
+    use crate::change_buffer::Result;
+
+    #[test]
+    fn buffer_creation_and_slices() {
+        let mut buf = unsafe {
+            let buf: [u8; 256] = std::mem::zeroed();
+            ChangeBuffer::from_raw_parts(buf.as_ptr(), 256)
+        };
+        {
+            // Safety: slice is the only reference to the buffer in its scope.
+            let slice = unsafe { buf.as_mut_slice() };
+            assert_eq!(256, slice.len());
+            slice[1] = 42;
+        }
+        // Safety: slice is the only reference to the buffer in its scope.
+        let slice = unsafe { buf.as_slice() };
+        assert_eq!(256, slice.len());
+        assert_eq!(42, slice[1]);
+    }
+
+    #[test]
+    fn read_and_write() -> Result<()> {
+        let example =
+            b"This is an example string, long enough to get 16 bytes out of it, without issue.";
+        let mut ex_buf = example.to_vec();
+        let mut buf = unsafe { ChangeBuffer::from_raw_parts(ex_buf.as_mut_ptr(), ex_buf.len()) };
+        let mut index = 8;
+        assert_eq!(8101238474429984353, buf.read::<u64>(&mut index)?);
+        assert_eq!(7956016061199967596, buf.read::<u64>(&mut index)?);
+        index = 8;
+        buf.write_u32(index, 8675309)?;
+        index = 8;
+        assert_eq!(8675309, buf.read::<u32>(&mut index)?);
+
+        Ok(())
+    }
+
+    #[test]
+    fn clear_count() -> Result<()> {
+        let mut buffer = vec![0xFFu8; 64];
+        let mut buf = unsafe { ChangeBuffer::from_raw_parts(buffer.as_mut_ptr(), buffer.len()) };
+        buf.clear_count()?;
+        let mut index = 0;
+        assert_eq!(0, buf.read::<u32>(&mut index)?);
+        Ok(())
+    }
+
+    #[test]
+    fn read_different_types() -> Result<()> {
+        let mut buffer = vec![0u8; 64];
+        buffer[0..4].copy_from_slice(&42u32.to_le_bytes());
+        buffer[8..24].copy_from_slice(&123456789u128.to_le_bytes());
+        buffer[24..32].copy_from_slice(&1.5f64.to_le_bytes());
+
+        let buf = unsafe { ChangeBuffer::from_raw_parts(buffer.as_mut_ptr(), buffer.len()) };
+
+        let mut index = 0;
+        assert_eq!(42, buf.read::<u32>(&mut index)?);
+        assert_eq!(4, index);
+
+        index = 8;
+        assert_eq!(123456789u128, buf.read::<u128>(&mut index)?);
+
+        index = 24;
+        assert_eq!(1.5, buf.read::<f64>(&mut index)?);
+        Ok(())
+    }
+
+    #[test]
+    fn read_out_of_bounds() {
+        let mut buffer = vec![0u8; 8];
+        let buf = unsafe { ChangeBuffer::from_raw_parts(buffer.as_mut_ptr(), buffer.len()) };
+        let mut index = 4;
+        assert!(buf.read::<u64>(&mut index).is_err());
+    }
+
+    #[test]
+    fn write_out_of_bounds() {
+        let mut buffer = vec![0u8; 8];
+        let mut buf = unsafe { ChangeBuffer::from_raw_parts(buffer.as_mut_ptr(), buffer.len()) };
+        // 8-byte buffer, u32 at offset 5 needs bytes 5..9 — out of bounds
+        assert!(buf.write_u32(5, 123).is_err());
+    }
+}

libdd-trace-utils/src/change_buffer/mod.rs

@@ -0,0 +1,129 @@
+//! Change buffer.
+//!
+//! A change buffer is a contiguous shared memory area between libdatadog and an external runtime.
+//! In order to amortize the cost of crossing the FFI when using native spans, the runtime writes
+//! events into the change buffer instead of calling libdatadog many times, and only flushes by
+//! batch — that flush is where the call to libdatadog happens. Libdatadog then processes the change
+//! buffer and reconstructs the corresponding spans.
+//!
+//! The change buffer is currently designed and used for dd-trace-js, but the idea could be extended
+//! to other runtime where the FFI cost is high.
+
+#![allow(unused)]
+
+/// Errors that can occur when operating on a [`ChangeBuffer`] or [`ChangeBufferState`].
+#[derive(Debug)]
+pub enum ChangeBufferError {
+    SpanNotFound(u64),
+    /// A string index didn't have any corresponding entry in the string table.
+    StringNotFound(u32),
+    /// A read is out of bounds.
+    ReadOutOfBounds {
+        /// The starting offset of the read.
+        offset: usize,
+        /// The size in bytes of the value attempted to be read starting at `offset`.
+        /// We have `offset + value_len > buffer_len`.
+        value_len: usize,
+        /// The total size of the buffer.
+        buffer_len: usize,
+    },
+    /// A is write is out of bounds.
+    WriteOutOfBounds {
+        /// The starting offset of the write.
+        offset: usize,
+        /// The size in bytes of the value attempted to be written starting at `offset`.
+        /// We have `offset + value_len > buffer_len`.
+        value_len: usize,
+        /// The total size of the buffer.
+        buffer_len: usize,
+    },
+    /// Unknown opcode.
+    UnknownOpcode(u32),
+}
+
+impl std::fmt::Display for ChangeBufferError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ChangeBufferError::SpanNotFound(id) => write!(f, "span not found: {id}"),
+            ChangeBufferError::StringNotFound(id) => {
+                write!(f, "string not found internally: {id}")
+            }
+            ChangeBufferError::ReadOutOfBounds {
+                offset,
+                value_len,
+                buffer_len,
+            } => {
+                write!(f, "read out of bounds: offset={offset}, value_len={value_len}, buffer_len={buffer_len}")
+            }
+            ChangeBufferError::WriteOutOfBounds {
+                offset,
+                value_len,
+                buffer_len,
+            } => {
+                write!(f, "write out of bounds: offset={offset}, value_len={value_len}, buffer_len={buffer_len}")
+            }
+            ChangeBufferError::UnknownOpcode(val) => write!(f, "unknown opcode: {val}"),
+        }
+    }
+}
+
+impl std::error::Error for ChangeBufferError {}
+
+pub type Result<T> = std::result::Result<T, ChangeBufferError>;
+
+mod utils;
+use utils::*;
+
+mod trace;
+pub use trace::*;
+
+mod operation;
+use operation::*;
+
+mod buffer;
+pub use buffer::*;
+
+pub mod span_header;
+pub use span_header::{SpanHeader, SPAN_HEADER_SIZE};
+
+use crate::span::v04::Span;
+use crate::span::{SpanText, TraceData};
+
+fn vec_insert<K: PartialEq, V>(vec: &mut Vec<(K, V)>, key: K, value: V) {
+    for entry in vec.iter_mut() {
+        if entry.0 == key {
+            entry.1 = value;
+            return;
+        }
+    }
+    vec.push((key, value));
+}
+
+fn vec_get<'a, K: PartialEq, V>(vec: &'a [(K, V)], key: &K) -> Option<&'a V> {
+    for entry in vec {
+        if entry.0 == *key {
+            return Some(&entry.1);
+        }
+    }
+    None
+}
+
+fn deferred_meta_insert(vec: &mut Vec<(u32, u32)>, key_id: u32, val_id: u32) {
+    for entry in vec.iter_mut() {
+        if entry.0 == key_id {
+            entry.1 = val_id;
+            return;
+        }
+    }
+    vec.push((key_id, val_id));
+}
+
+fn deferred_metric_insert(vec: &mut Vec<(u32, f64)>, key_id: u32, val: f64) {
+    for entry in vec.iter_mut() {
+        if entry.0 == key_id {
+            entry.1 = val;
+            return;
+        }
+    }
+    vec.push((key_id, val));
+}