fix(library-config): update Linux process context

libdd-library-config/Cargo.toml

@@ -15,7 +15,7 @@ crate-type = ["lib"]
 bench = false
 
 [features]
-default = ["process-context-reader", "process-context-writer"]
+default = ["process-context-writer"]
 otel-thread-ctx = ["process-context-writer"]
 process-context-reader = []
 process-context-writer = []

libdd-library-config/src/otel_process_ctx.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 //! Implementation of the Linux parts of the [OTEL process
-//! context specification](https://github.com/open-telemetry/opentelemetry-specification/pull/4719).
+//! context specification](https://github.com/open-telemetry/opentelemetry-specification/blob/main/oteps/profiles/4719-process-ctx.md).
 //!
 //! The update/read protocol is seqlock-style: the publisher marks the mapping as unavailable,
 //! writes the payload metadata, publishes a non-zero version, and readers accept a copy only if

libdd-library-config/src/otel_process_ctx/reader/copy_pipe_unix.rs

@@ -77,7 +77,11 @@ impl ProcessMemoryCopy for CopyPipe {
                                 io::ErrorKind::WouldBlock,
                                 "process context memory was unmapped during read",
                             ),
-                            pipe_dirty: false,
+                            // actually false on Linux: if EFAULT is returned, nothing was written;
+                            // should anything have been written already we would get a short write
+                            // However, this is not the case for macOS, despite what its manual
+                            // says: See https://github.com/apple-oss-distributions/xnu/blob/5c306bec31e314fa4d8bbdafb2f6f5a6b7e7b291/bsd/man/man2/write.2#L168-L186
+                            pipe_dirty: true,
                         });
                     }
                     _ => {
@@ -233,7 +237,7 @@ mod tests {
             .expect_err("a copy crossing into inaccessible memory should fail");
 
         assert_eq!(err.err.kind(), io::ErrorKind::WouldBlock);
-        assert!(!err.pipe_dirty);
+        assert!(err.pipe_dirty);
         // SAFETY: address and len came from mmap above.
         assert_eq!(unsafe { libc::munmap(address, len) }, 0);
     }

libdd-library-config/src/otel_process_ctx/writer.rs

@@ -5,7 +5,7 @@ use core::{
     convert::TryInto,
     marker::PhantomData,
     mem::swap,
-    ptr,
+    ptr::{self, NonNull},
     sync::atomic::{fence, AtomicPtr, AtomicU32, AtomicU64, Ordering},
 };
 use std::{
@@ -68,7 +68,7 @@ static PROCESS_CONTEXT_HANDLER: Mutex<Option<ProcessContextHandle>> = Mutex::new
 
 pub(super) trait HeaderMemoryHolder: Sized {
     fn new() -> io::Result<Self>;
-    fn as_ptr(&self) -> *mut MappingHeader;
+    fn as_ptr(&self) -> Option<NonNull<MappingHeader>>;
     fn make_discoverable(&mut self);
     fn unpublish_and_release(self) -> io::Result<()>;
     fn after_fork(self);
@@ -84,9 +84,6 @@ struct ProcessContextHandleGen<M: HeaderMemoryHolder, T: MonotonicTime> {
     /// Once published, and until the next update is complete, the backing allocation of
     /// `payload` might be read and thus must not move (e.g. by resizing or drop).
     payload: Vec<u8>,
-    /// The process id of the last publisher. This is useful to detect forks(), and publish a
-    /// new context accordingly.
-    pid: u32,
     monotonic_clock: PhantomData<T>,
 }
 
@@ -101,7 +98,11 @@ impl<M: HeaderMemoryHolder, T: MonotonicTime> ProcessContextHandleGen<M, T> {
         let mut mapping = M::new()?;
         let published_at_ns = T::monotonic_time_ns()?;
 
-        let header = mapping.as_ptr();
+        let header = mapping
+            .as_ptr()
+            // should never happen; as_ptr should only return None after a fork
+            .ok_or_else(|| io::Error::other("new process context header mapping is unavailable"))?
+            .as_ptr();
 
         // SAFETY: header points to a zero-filled, writable allocation of at least
         // mapping_size() bytes with MappingHeader alignment; field projections are in-bounds.
@@ -111,12 +112,12 @@ impl<M: HeaderMemoryHolder, T: MonotonicTime> ProcessContextHandleGen<M, T> {
         unsafe {
             ptr::addr_of_mut!((*header).signature).write(*SIGNATURE);
             ptr::addr_of_mut!((*header).version).write(PROCESS_CTX_VERSION);
-            (*header)
-                .payload_ptr
-                .store(payload.as_ptr().cast_mut(), Ordering::Relaxed);
             (*header)
                 .payload_size
                 .store(payload_size, Ordering::Relaxed);
+            (*header)
+                .payload_ptr
+                .store(payload.as_ptr().cast_mut(), Ordering::Relaxed);
 
             fence(Ordering::SeqCst);
             (*header)
@@ -129,14 +130,22 @@ impl<M: HeaderMemoryHolder, T: MonotonicTime> ProcessContextHandleGen<M, T> {
         Ok(ProcessContextHandleGen {
             mapping,
             payload,
-            pid: std::process::id(),
             monotonic_clock: PhantomData,
         })
     }
 
     /// Updates the context after initial publication.
     fn update(&mut self, payload: Vec<u8>) -> io::Result<()> {
-        let header = self.mapping.as_ptr();
+        let header = self
+            .mapping
+            .as_ptr()
+            .ok_or_else(|| {
+                io::Error::new(
+                    io::ErrorKind::InvalidInput,
+                    "process context header mapping is unavailable after fork",
+                )
+            })?
+            .as_ptr();
 
         let monotonic_published_at_ns = T::monotonic_time_ns()?;
         let payload_size: u32 = payload.len().try_into().map_err(|_| {
@@ -157,7 +166,10 @@ impl<M: HeaderMemoryHolder, T: MonotonicTime> ProcessContextHandleGen<M, T> {
                 .swap(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed)
         };
         if last_monotonic_published_at_ns == UNPUBLISHED_OR_UPDATING {
-            panic!("concurrent update of the process context is not supported");
+            return Err(io::Error::new(
+                io::ErrorKind::WouldBlock,
+                "context is already being updated",
+            ));
         }
 
         let monotonic_published_at_ns =
@@ -172,12 +184,12 @@ impl<M: HeaderMemoryHolder, T: MonotonicTime> ProcessContextHandleGen<M, T> {
         self.payload = payload;
 
         unsafe {
-            (*header)
-                .payload_ptr
-                .store(self.payload.as_ptr().cast_mut(), Ordering::Relaxed);
             (*header)
                 .payload_size
                 .store(payload_size, Ordering::Relaxed);
+            (*header)
+                .payload_ptr
+                .store(self.payload.as_ptr().cast_mut(), Ordering::Relaxed);
         }
 
         // Prevent the final timestamp publication from moving above either the payload metadata
@@ -214,8 +226,7 @@ fn lock_context_handle() -> io::Result<MutexGuard<'static, Option<ProcessContext
 ///
 /// - this is the first publication
 /// - [unpublish] has been called last
-/// - the previous context has been published from a different process id (that is, a `fork()`
-///   happened and we're the child process)
+/// - the previous header mapping is unavailable after `fork()`
 ///
 /// Then we follow the Publish protocol of the OTel process context specification (allocating a
 /// fresh header).
@@ -239,7 +250,7 @@ pub(super) fn publish_raw_payload(payload: Vec<u8>) -> io::Result<()> {
     let mut guard = lock_context_handle()?;
 
     match &mut *guard {
-        Some(handler) if handler.pid == std::process::id() => handler.update(payload),
+        Some(handler) if handler.mapping.as_ptr().is_some() => handler.update(payload),
         Some(handler) => {
             let mut local_handler = ProcessContextHandleGen::publish(payload)?;
             // If we've been forked, we need to prevent the mapping from being dropped
@@ -272,20 +283,22 @@ pub fn unpublish() -> io::Result<()> {
         mapping, payload, ..
     }) = guard.take()
     {
-        // Mark the context as unavailable before freeing the mapping/payload. The fence forces
-        // the writing CPU not to reorder the unavailable timestamp store and the deallocation
-        // stores. This gives readers more of a chance (but no guarantee) to observe an
-        // unavailable context before the publication is removed.
-        //
-        // SAFETY: the mapping is still live and valid, and the global mutex prevents
-        // concurrent in-process writers from mutating the plain header fields.
-        let header = mapping.as_ptr();
-        unsafe {
-            (*header)
-                .monotonic_published_at_ns
-                .store(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed);
+        if let Some(header) = mapping.as_ptr() {
+            // Mark the context as unavailable before freeing the mapping/payload. The fence
+            // forces the writing CPU not to reorder the unavailable timestamp store and the
+            // deallocation stores. This gives readers more of a chance (but no guarantee) to
+            // observe an unavailable context before the publication is removed.
+            //
+            // SAFETY: the mapping is still live and valid, and the global mutex prevents
+            // concurrent in-process writers from mutating the plain header fields.
+            let header = header.as_ptr();
+            unsafe {
+                (*header)
+                    .monotonic_published_at_ns
+                    .store(UNPUBLISHED_OR_UPDATING, Ordering::Relaxed);
+            }
+            fence(Ordering::SeqCst);
         }
-        fence(Ordering::SeqCst);
 
         // The payload will still drop if this fails, leaving a zero timestamp behind.
         mapping.unpublish_and_release()?;

libdd-library-config/src/otel_process_ctx/writer/linux.rs

@@ -4,7 +4,7 @@
 use core::{
     convert::TryInto,
     ffi::{c_void, CStr},
-    mem::{forget, ManuallyDrop},
+    mem::ManuallyDrop,
     ptr::{self, NonNull},
     time::Duration,
 };
@@ -19,12 +19,14 @@ use std::{
 /// # Safety
 ///
 /// The following invariants MUST always hold for safety and are guaranteed by [MemMapping]:
-/// - `start` is non-null, is coming from a previous call to `mmap` with a size value of
-///   `mapping_size()` and hasn't been unmmaped since.
+/// - when `only_for_pid` allows access from the current process, `start_addr` points to a live
+///   mapping of `mapping_size()` bytes created by `mmap`.
 /// - once `self` has been dropped, no memory access must be performed on the memory previously
-///   pointed to by `start`.
+///   pointed to by `start_addr`.
 pub(super) struct MemMapping {
     start_addr: NonNull<c_void>,
+    /// `Some(pid)` when `MADV_DONTFORK` succeeded, otherwise `None`.
+    only_for_pid: Option<u32>,
 }
 
 // SAFETY: MemMapping represents ownership over the mapped region. It never leaks or
@@ -33,23 +35,18 @@ unsafe impl Send for MemMapping {}
 
 impl super::HeaderMemoryHolder for MemMapping {
     fn new() -> io::Result<Self> {
-        let mapping = Self::new()?;
-        check_syscall_retval(
-            // SAFETY: MemMapping owns a live mapping of mapping_size() bytes.
-            unsafe {
-                libc::madvise(
-                    mapping.start_addr.as_ptr(),
-                    super::mapping_size(),
-                    libc::MADV_DONTFORK,
-                )
-            },
-            "madvise MADVISE_DONTFORK failed",
-        )?;
-        Ok(mapping)
+        Self::new()
     }
 
-    fn as_ptr(&self) -> *mut super::MappingHeader {
-        self.start_addr.as_ptr() as *mut super::MappingHeader
+    fn as_ptr(&self) -> Option<NonNull<super::MappingHeader>> {
+        if self
+            .only_for_pid
+            .is_some_and(|pid| pid != std::process::id())
+        {
+            None
+        } else {
+            Some(self.start_addr.cast())
+        }
     }
 
     fn make_discoverable(&mut self) {
@@ -63,7 +60,7 @@ impl super::HeaderMemoryHolder for MemMapping {
     }
 
     fn after_fork(self) {
-        forget(self);
+        drop(self);
     }
 }
 
@@ -92,7 +89,7 @@ impl MemMapping {
     fn new() -> io::Result<Self> {
         let size = super::mapping_size();
 
-        try_memfd(crate::otel_process_ctx::linux::MAPPING_NAME, libc::MFD_CLOEXEC | libc::MFD_NOEXEC_SEAL | libc::MFD_ALLOW_SEALING)
+        let mut mapping = try_memfd(crate::otel_process_ctx::linux::MAPPING_NAME, libc::MFD_CLOEXEC | libc::MFD_NOEXEC_SEAL | libc::MFD_ALLOW_SEALING)
             .or_else(|_| try_memfd(crate::otel_process_ctx::linux::MAPPING_NAME, libc::MFD_CLOEXEC | libc::MFD_ALLOW_SEALING))
             .and_then(|fd| {
                 // SAFETY: fd is a valid open file descriptor.
@@ -118,7 +115,10 @@ impl MemMapping {
                 )?;
 
                 // We (implicitly) close the file descriptor right away, but this ok
-                Ok(MemMapping { start_addr })
+                Ok(MemMapping {
+                    start_addr,
+                    only_for_pid: None,
+                })
             })
             // If any previous step failed, we fallback to an anonymous mapping
             .or_else(|_| {
@@ -137,8 +137,24 @@ impl MemMapping {
                     "mmap failed: couldn't create a memfd or anonymous mmapped region for process context publication"
                 )?;
 
-                Ok(MemMapping { start_addr })
-            })
+                Ok::<_, io::Error>(MemMapping {
+                    start_addr,
+                    only_for_pid: None,
+                })
+            })?;
+
+        // SAFETY: MemMapping owns a live mapping of mapping_size() bytes. Failure is harmless;
+        // the mapping then follows the default inheritance behavior.
+        mapping.only_for_pid = (unsafe {
+            libc::madvise(
+                mapping.start_addr.as_ptr(),
+                super::mapping_size(),
+                libc::MADV_DONTFORK,
+            )
+        } == 0)
+            .then_some(std::process::id());
+
+        Ok(mapping)
     }
 
     /// Makes this mapping discoverable by giving it a name.
@@ -190,6 +206,13 @@ impl MemMapping {
     /// Practically, `self` must be put in a `ManuallyDrop` wrapper and forgotten, or being in
     /// the process of being dropped.
     unsafe fn unmap(&mut self) -> io::Result<()> {
+        if self
+            .only_for_pid
+            .is_some_and(|pid| pid != std::process::id())
+        {
+            return Ok(());
+        }
+
         check_syscall_retval(
             // SAFETY: upheld by the caller.
             unsafe { libc::munmap(self.start_addr.as_ptr(), super::mapping_size()) },