diff --git a/libdd-library-config/Cargo.toml b/libdd-library-config/Cargo.toml index 54b2d57083..b46e7a7a3e 100644 --- a/libdd-library-config/Cargo.toml +++ b/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 = [] diff --git a/libdd-library-config/src/otel_process_ctx.rs b/libdd-library-config/src/otel_process_ctx.rs index 53229e4ce5..8a34cb6801 100644 --- a/libdd-library-config/src/otel_process_ctx.rs +++ b/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 diff --git a/libdd-library-config/src/otel_process_ctx/reader/copy_pipe_unix.rs b/libdd-library-config/src/otel_process_ctx/reader/copy_pipe_unix.rs index 9211fd0769..4c75d29dfb 100644 --- a/libdd-library-config/src/otel_process_ctx/reader/copy_pipe_unix.rs +++ b/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); } diff --git a/libdd-library-config/src/otel_process_ctx/writer.rs b/libdd-library-config/src/otel_process_ctx/writer.rs index e4367b3f6d..408a3e929a 100644 --- a/libdd-library-config/src/otel_process_ctx/writer.rs +++ b/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> = Mutex::new pub(super) trait HeaderMemoryHolder: Sized { fn new() -> io::Result; - fn as_ptr(&self) -> *mut MappingHeader; + fn as_ptr(&self) -> Option>; fn make_discoverable(&mut self); fn unpublish_and_release(self) -> io::Result<()>; fn after_fork(self); @@ -84,9 +84,6 @@ struct ProcessContextHandleGen { /// 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, - /// The process id of the last publisher. This is useful to detect forks(), and publish a - /// new context accordingly. - pid: u32, monotonic_clock: PhantomData, } @@ -101,7 +98,11 @@ impl ProcessContextHandleGen { 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 ProcessContextHandleGen { 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 ProcessContextHandleGen { Ok(ProcessContextHandleGen { mapping, payload, - pid: std::process::id(), monotonic_clock: PhantomData, }) } /// Updates the context after initial publication. fn update(&mut self, payload: Vec) -> 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 ProcessContextHandleGen { .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 ProcessContextHandleGen { 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) -> 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()?; diff --git a/libdd-library-config/src/otel_process_ctx/writer/linux.rs b/libdd-library-config/src/otel_process_ctx/writer/linux.rs index 7c5ef2f145..1004811d89 100644 --- a/libdd-library-config/src/otel_process_ctx/writer/linux.rs +++ b/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, + /// `Some(pid)` when `MADV_DONTFORK` succeeded, otherwise `None`. + only_for_pid: Option, } // 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 { - 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> { + 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 { 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()) },