From b4cb3854194ddda1afe66ebe965948fcb05b7bf1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Veljko=20Rvovi=C4=87?= Date: Wed, 29 Jul 2026 12:16:15 +0000 Subject: [PATCH 1/4] Fix duplicate MMV filenames in tests Three tests in countvector.rs and gaugevector.rs used the same "count_vector_test" MMV filename. If run in parallel, tests could interleave and fail. Give each test its own unique MMV filename. --- src/client/metric/countvector.rs | 2 +- src/client/metric/gaugevector.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/client/metric/countvector.rs b/src/client/metric/countvector.rs index 13cf4af..8992e35 100644 --- a/src/client/metric/countvector.rs +++ b/src/client/metric/countvector.rs @@ -204,7 +204,7 @@ pub fn test_multiple_initvals() { assert_eq!(cv.val("b").unwrap(), 2); assert_eq!(cv.val("c").unwrap(), 3); - Client::new("count_vector_test") + Client::new("count_vector_multiple_initvals_test") .unwrap() .export(&mut [&mut cv]) .unwrap(); diff --git a/src/client/metric/gaugevector.rs b/src/client/metric/gaugevector.rs index 8372161..68d0d68 100644 --- a/src/client/metric/gaugevector.rs +++ b/src/client/metric/gaugevector.rs @@ -135,7 +135,7 @@ pub fn test() { assert_eq!(gv.val("b").unwrap(), 1.5); assert_eq!(gv.val("c").unwrap(), 1.5); - Client::new("count_vector_test") + Client::new("gauge_vector_test") .unwrap() .export(&mut [&mut gv]) .unwrap(); From b892ca44f6909fa7590974c3572ed37f9e0f1ebf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Veljko=20Rvovi=C4=87?= Date: Wed, 29 Jul 2026 10:06:05 +0000 Subject: [PATCH 2/4] Fix ignored partial write when zero-filling the MMV file Client::export used Write::write, which is only guaranteed to write some of the buffer, to pre-size the mmap'd MMV file with zeros. A short write would leave the file smaller than mmv_size, silently corrupting everything mapped after it. clippy already flags this as unused_io_amount (deny-by-default). Use write_all instead, which loops until the whole buffer is written or an error occurs. Assisted-by: Cursor:claude-opus-5 --- src/client/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/client/mod.rs b/src/client/mod.rs index 94e65e0..be18717 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -283,7 +283,7 @@ impl Client { .truncate(true) .open(&self.mmv_path)?; - file.write(&vec![0; mmv_size])?; + file.write_all(&vec![0; mmv_size])?; ws.mmap_view = Some(Mmap::open(&file, Protection::ReadWrite)?.into_view_sync()); From 583825aae14ac47735190edb8d5597c74bc208f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Veljko=20Rvovi=C4=87?= Date: Wed, 29 Jul 2026 10:18:12 +0000 Subject: [PATCH 3/4] Drop byteorder, regex, time, and nix/kernel32-sys in favor of std Drops four third-party dependencies in favor of std, reducing maintenance surface and modernizing away from older/unmaintained crates. byteorder: replaced by a ReadBytesExt/WriteBytesExt pair in src/byteio. The old code hardcoded LittleEndian, which was technically incorrect: looking at mmv src code, MMV files should use native-endian byte order. Hence the new code will also use native order. regex: existed only to parse PCP_KEY=value lines out of pcp.conf, now a hand-rolled byte-slice parser with the same semantics. nix/kernel32-sys: get_process_id had one #[cfg] arm per platform, both replaced by std::process::id. time: Timer moves from wall-clock Tm/Duration to the monotonic std::time::Instant. Also improves windows arm for osstr_from_bytes which asserted pcp.conf holds valid UTF-8 via from_utf8_unchecked. It now checks and skips the line instead. Assisted-by: Cursor:claude-opus-5 --- Cargo.toml | 9 --- src/byteio.rs | 63 ++++++++++++++++++ src/client/metric/mod.rs | 71 ++++++++++---------- src/client/metric/timer.rs | 21 +++--- src/client/mod.rs | 133 +++++++++++++++++++------------------ src/lib.rs | 10 +-- src/mmv/mod.rs | 70 ++++++++++--------- 7 files changed, 212 insertions(+), 165 deletions(-) create mode 100644 src/byteio.rs diff --git a/Cargo.toml b/Cargo.toml index 4712f74..3871a10 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,12 +10,9 @@ keywords = ["performance", "instrumentation", "metric", "pcp", "mmv"] [dependencies] bitflags = "0.9.1" -byteorder = "1.0.0" hdrsample = "4.0.0" lazy_static = "0.2.8" memmap = "0.5.2" -regex = "0.2" -time = "0.1" [dev-dependencies] rand = "0.3.15" @@ -23,9 +20,3 @@ hyper = "0.11.2" futures = "0.1.14" curl = "0.4.8" iron = "0.5.1" - -[target.'cfg(unix)'.dependencies] -nix = "0.8.0" - -[target.'cfg(windows)'.dependencies] -kernel32-sys = "0.2.2" diff --git a/src/byteio.rs b/src/byteio.rs new file mode 100644 index 0000000..1924a16 --- /dev/null +++ b/src/byteio.rs @@ -0,0 +1,63 @@ +//! Native-endian read/write helpers +//! +//! MMV files are a same-host IPC mechanism, written by an instrumented process +//! and read back via mmap by `pmdammv` running on that same machine, never +//! transferred across hosts. The mmv(5) spec and the reference `libpcp_mmv` C +//! implementation both write plain native integers with no conversion, so these +//! methods use the host's native byte order rather than a fixed one. + +use std::io::{self, Read, Write}; + +pub trait ReadBytesExt: Read { + fn read_u8(&mut self) -> io::Result { + let mut buf = [0u8; 1]; + self.read_exact(&mut buf)?; + Ok(buf[0]) + } + + fn read_i32(&mut self) -> io::Result { + let mut buf = [0u8; 4]; + self.read_exact(&mut buf)?; + Ok(i32::from_ne_bytes(buf)) + } + + fn read_u32(&mut self) -> io::Result { + let mut buf = [0u8; 4]; + self.read_exact(&mut buf)?; + Ok(u32::from_ne_bytes(buf)) + } + + fn read_i64(&mut self) -> io::Result { + let mut buf = [0u8; 8]; + self.read_exact(&mut buf)?; + Ok(i64::from_ne_bytes(buf)) + } + + fn read_u64(&mut self) -> io::Result { + let mut buf = [0u8; 8]; + self.read_exact(&mut buf)?; + Ok(u64::from_ne_bytes(buf)) + } +} + +impl ReadBytesExt for R {} + +pub trait WriteBytesExt: Write { + fn write_i32(&mut self, n: i32) -> io::Result<()> { + self.write_all(&n.to_ne_bytes()) + } + + fn write_u32(&mut self, n: u32) -> io::Result<()> { + self.write_all(&n.to_ne_bytes()) + } + + fn write_i64(&mut self, n: i64) -> io::Result<()> { + self.write_all(&n.to_ne_bytes()) + } + + fn write_u64(&mut self, n: u64) -> io::Result<()> { + self.write_all(&n.to_ne_bytes()) + } +} + +impl WriteBytesExt for W {} diff --git a/src/client/metric/mod.rs b/src/client/metric/mod.rs index 8eac27c..8a3192e 100644 --- a/src/client/metric/mod.rs +++ b/src/client/metric/mod.rs @@ -1,4 +1,4 @@ -use byteorder::WriteBytesExt; +use crate::byteio::WriteBytesExt; use memmap::{Mmap, MmapViewSync, Protection}; use std::collections::hash_map::{DefaultHasher, HashMap}; use std::collections::hash_set::Iter; @@ -12,9 +12,9 @@ use std::str; use super::super::mmv::{MTCode, Version}; use super::super::{ - Endian, INDOM_BIT_LEN, INDOM_BLOCK_LEN, INSTANCE_BLOCK_LEN_MMV1, INSTANCE_BLOCK_LEN_MMV2, - ITEM_BIT_LEN, METRIC_BLOCK_LEN_MMV1, METRIC_BLOCK_LEN_MMV2, MMV1_NAME_MAX_LEN, - NUMERIC_VALUE_SIZE, STRING_BLOCK_LEN, VALUE_BLOCK_LEN, + INDOM_BIT_LEN, INDOM_BLOCK_LEN, INSTANCE_BLOCK_LEN_MMV1, INSTANCE_BLOCK_LEN_MMV2, ITEM_BIT_LEN, + METRIC_BLOCK_LEN_MMV1, METRIC_BLOCK_LEN_MMV2, MMV1_NAME_MAX_LEN, NUMERIC_VALUE_SIZE, + STRING_BLOCK_LEN, VALUE_BLOCK_LEN, }; mod counter; @@ -38,7 +38,7 @@ pub use self::histogram::Histogram; pub use self::histogram::RecordError as HistRecordError; mod private { - use byteorder::WriteBytesExt; + use crate::byteio::WriteBytesExt; use std::io; /// Generic type for any Metric's value @@ -172,7 +172,7 @@ macro_rules! impl_metric_type_for ( fn write(&self, w: &mut W) -> io::Result<()> { - w.write_u64::( + w.write_u64( unsafe { mem::transmute::<$typ, $base_typ>(*self) as u64 } @@ -820,28 +820,28 @@ impl Metric { } Version::V2 => { let name_off = write_mmv_string(ws, c, &self.name, false)?; - c.write_u64::(name_off)?; + c.write_u64(name_off)?; } } // item - c.write_u32::(self.item)?; + c.write_u32(self.item)?; // type code - c.write_u32::(self.val.type_code())?; + c.write_u32(self.val.type_code())?; // sem - c.write_u32::(self.sem as u32)?; + c.write_u32(self.sem as u32)?; // unit - c.write_u32::(self.unit)?; + c.write_u32(self.unit)?; // indom - c.write_u32::(self.indom)?; + c.write_u32(self.indom)?; // zero pad - c.write_u32::(0)?; + c.write_u32(0)?; // short help let short_help_off = write_mmv_string(ws, c, &self.shorthelp, false)?; - c.write_u64::(short_help_off)?; + c.write_u64(short_help_off)?; // long help let long_help_off = write_mmv_string(ws, c, &self.longhelp, false)?; - c.write_u64::(long_help_off)?; + c.write_u64(long_help_off)?; if write_value_blk { let (value_offset, value_size) = @@ -975,9 +975,9 @@ fn write_indom_and_instances<'a>( let indom_off = ws.indom_sec_off + INDOM_BLOCK_LEN * ws.indom_idx; c.set_position(indom_off); // indom id - c.write_u32::(indom.id)?; + c.write_u32(indom.id)?; // number of instances - c.write_u32::(indom.instance_count())?; + c.write_u32(indom.instance_count())?; // offset to instances let instance_blk_len = match mmv_ver { @@ -985,14 +985,14 @@ fn write_indom_and_instances<'a>( Version::V2 => INSTANCE_BLOCK_LEN_MMV2, }; let mut instance_blk_off = ws.instance_sec_off + instance_blk_len * ws.instance_idx; - c.write_u64::(instance_blk_off)?; + c.write_u64(instance_blk_off)?; // short help let short_help_off = write_mmv_string(ws, c, indom.shorthelp(), false)?; - c.write_u64::(short_help_off)?; + c.write_u64(short_help_off)?; // long help let long_help_off = write_mmv_string(ws, c, indom.longhelp(), false)?; - c.write_u64::(long_help_off)?; + c.write_u64(long_help_off)?; // write instances and record their offsets let mut instance_blk_offs = HashMap::with_capacity(indom.instances.len()); @@ -1000,11 +1000,11 @@ fn write_indom_and_instances<'a>( c.set_position(instance_blk_off); // indom offset - c.write_u64::(indom_off)?; + c.write_u64(indom_off)?; // zero pad - c.write_u32::(0)?; + c.write_u32(0)?; // instance id - c.write_u32::(Indom::instance_id(&instance))?; + c.write_u32(Indom::instance_id(&instance))?; // instance match mmv_ver { @@ -1014,7 +1014,7 @@ fn write_indom_and_instances<'a>( } Version::V2 => { let instance_off = write_mmv_string(ws, c, instance, false)?; - c.write_u64::(instance_off)?; + c.write_u64(instance_off)?; } } @@ -1060,7 +1060,7 @@ fn write_value_block( let (value_offset, value_size); if value.type_code() == MTCode::String as u32 { // numeric value - c.write_u64::(0)?; + c.write_u64(0)?; // string offset @@ -1073,7 +1073,7 @@ fn write_value_block( let str_val = unsafe { str::from_utf8_unchecked(&str_buf) }; let string_val_off = write_mmv_string(ws, c, str_val, true)?; - c.write_u64::(string_val_off)?; + c.write_u64(string_val_off)?; value_offset = string_val_off as usize; value_size = STRING_BLOCK_LEN as usize; @@ -1084,12 +1084,12 @@ fn write_value_block( // numeric value value.write(&mut c)?; // string offset - c.write_u64::(0)?; + c.write_u64(0)?; } // offset to metric block - c.write_u64::(metric_blk_off)?; + c.write_u64(metric_blk_off)?; // offset to instance block - c.write_u64::(instance_blk_off)?; + c.write_u64(instance_blk_off)?; c.set_position(orig_pos); Ok((value_offset, value_size)) @@ -1347,7 +1347,7 @@ fn test_mmv2_string_blocks() { #[test] fn test_random_numeric_metrics() { use super::Client; - use byteorder::ReadBytesExt; + use crate::byteio::ReadBytesExt; use rand::{thread_rng, Rng}; let mut metrics = Vec::new(); @@ -1409,14 +1409,14 @@ fn test_random_numeric_metrics() { for (m, v) in metrics.iter_mut().zip(new_vals) { let mut slice = unsafe { m.mmap_view.as_slice() }; - assert_eq!(v, slice.read_u64::().unwrap() as u32); + assert_eq!(v, slice.read_u64().unwrap() as u32); } } #[test] fn test_simple_metrics() { use super::Client; - use byteorder::ReadBytesExt; + use crate::byteio::ReadBytesExt; use rand::{thread_rng, Rng}; use std::ffi::CStr; use std::mem::transmute; @@ -1471,7 +1471,7 @@ fn test_simple_metrics() { let mut freq_slice = unsafe { freq.mmap_view.as_slice() }; assert_eq!(new_freq, unsafe { - transmute::(freq_slice.read_u64::().unwrap()) + transmute::(freq_slice.read_u64().unwrap()) }); let color_slice = unsafe { color.mmap_view.as_slice() }; @@ -1479,10 +1479,7 @@ fn test_simple_metrics() { assert_eq!(new_color, cstr.to_str().unwrap()); let mut photon_slice = unsafe { photons.mmap_view.as_slice() }; - assert_eq!( - new_photon_count, - photon_slice.read_u64::().unwrap() as u32 - ); + assert_eq!(new_photon_count, photon_slice.read_u64().unwrap() as u32); // TODO: after implementing mmvdump functionality, test the // bytes of the entier MMV file diff --git a/src/client/metric/timer.rs b/src/client/metric/timer.rs index 24e83ef..8aab11f 100644 --- a/src/client/metric/timer.rs +++ b/src/client/metric/timer.rs @@ -1,6 +1,5 @@ use super::*; -use time; -use time::Tm; +use std::time::Instant; /// A timer metric for tracking elapsed time /// @@ -8,7 +7,7 @@ use time::Tm; pub struct Timer { metric: Metric, time_scale: Time, - start_time: Option, + start_time: Option, } /// Error encountered while starting or stopping a timer @@ -58,7 +57,7 @@ impl Timer { if self.start_time.is_some() { return Err(Error::TimerAlreadyStarted); } - self.start_time = Some(time::now()); + self.start_time = Some(Instant::now()); Ok(()) } @@ -70,15 +69,15 @@ impl Timer { pub fn stop(&mut self) -> Result { match self.start_time { Some(start_time) => { - let duration = time::now() - start_time; + let duration = start_time.elapsed(); let elapsed = match self.time_scale { - Time::NSec => duration.num_nanoseconds().unwrap_or(0), - Time::USec => duration.num_microseconds().unwrap_or(0), - Time::MSec => duration.num_microseconds().unwrap_or(0), - Time::Sec => duration.num_seconds(), - Time::Min => duration.num_minutes(), - Time::Hour => duration.num_hours(), + Time::NSec => duration.as_nanos() as i64, + Time::USec => duration.as_micros() as i64, + Time::MSec => duration.as_micros() as i64, + Time::Sec => duration.as_secs() as i64, + Time::Min => (duration.as_secs() / 60) as i64, + Time::Hour => (duration.as_secs() / 3600) as i64, }; let val = *self.metric.val(); diff --git a/src/client/mod.rs b/src/client/mod.rs index be18717..698d41c 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -1,6 +1,5 @@ -use byteorder::WriteBytesExt; +use crate::byteio::WriteBytesExt; use memmap::{Mmap, Protection}; -use regex::bytes::Regex; use std::env; use std::ffi::{OsStr, OsString}; use std::fmt; @@ -10,14 +9,14 @@ use std::io; use std::io::prelude::*; use std::io::{BufReader, Cursor}; use std::path::{Path, PathBuf, MAIN_SEPARATOR}; +use std::process; use std::str; -use time; +use std::time::{SystemTime, UNIX_EPOCH}; use super::mmv::Version; use super::{ - Endian, CLUSTER_ID_BIT_LEN, HDR_LEN, INDOM_BLOCK_LEN, INSTANCE_BLOCK_LEN_MMV1, - INSTANCE_BLOCK_LEN_MMV2, METRIC_BLOCK_LEN_MMV1, METRIC_BLOCK_LEN_MMV2, STRING_BLOCK_LEN, - TOC_BLOCK_LEN, VALUE_BLOCK_LEN, + CLUSTER_ID_BIT_LEN, HDR_LEN, INDOM_BLOCK_LEN, INSTANCE_BLOCK_LEN_MMV1, INSTANCE_BLOCK_LEN_MMV2, + METRIC_BLOCK_LEN_MMV1, METRIC_BLOCK_LEN_MMV2, STRING_BLOCK_LEN, TOC_BLOCK_LEN, VALUE_BLOCK_LEN, }; pub mod metric; @@ -26,27 +25,22 @@ use self::metric::{MMVWriter, MMVWriterState}; static PCP_TMP_DIR_KEY: &'static str = "PCP_TMP_DIR"; static MMV_DIR_SUFFIX: &'static str = "mmv"; -#[cfg(unix)] -fn get_process_id() -> i32 { - use nix; - nix::unistd::getpid() -} - -#[cfg(windows)] fn get_process_id() -> i32 { - use kernel32; - unsafe { kernel32::GetCurrentProcessId() as i32 } + process::id() as i32 } #[cfg(unix)] -fn osstr_from_bytes(slice: &[u8]) -> &OsStr { +fn osstr_from_bytes(slice: &[u8]) -> Option<&OsStr> { use std::os::unix::ffi::OsStrExt; - OsStr::from_bytes(slice) + Some(OsStr::from_bytes(slice)) } +/// Windows stores `OsStr` as WTF-8 and has no borrowed constructor from bytes, +/// so the only route from `&[u8]` to `&OsStr` is via `&str`. Bytes that aren't +/// valid UTF-8 are rejected rather than assumed. #[cfg(windows)] -fn osstr_from_bytes(slice: &[u8]) -> &OsStr { - OsStr::new(unsafe { str::from_utf8_unchecked(slice) }) +fn osstr_from_bytes(slice: &[u8]) -> Option<&OsStr> { + str::from_utf8(slice).ok().map(OsStr::new) } fn get_pcp_root() -> PathBuf { @@ -69,33 +63,46 @@ fn init_pcp_conf(pcp_root: &Path) -> io::Result<()> { parse_pcp_conf(pcp_conf) } +/// Parses one `PCP_VARIABLE_NAME=value` line, per the syntax in +/// `man 5 pcp.conf`: no space around the `=`, and values are unquoted and +/// may contain spaces. +/// +/// Returns `None` for anything else, including an unterminated last line. +fn parse_pcp_conf_line(line: &[u8]) -> Option<(&[u8], &[u8])> { + let line = line.strip_suffix(b"\n")?; + let eq = line.iter().position(|&b| b == b'=')?; + let (key, val) = (&line[..eq], &line[eq + 1..]); + + let suffix = key.strip_prefix(b"PCP_")?; + if suffix.is_empty() + || !suffix + .iter() + .all(|&b| b.is_ascii_alphanumeric() || b == b'_') + { + return None; + } + + if val.len() < 2 { + return None; + } + let is_quote = |b: u8| b == b'"' || b == b'\''; + if is_quote(val[0]) || is_quote(val[val.len() - 1]) { + return None; + } + + Some((key, val)) +} + fn parse_pcp_conf>(conf_path: P) -> io::Result<()> { let pcp_conf = File::open(conf_path)?; let mut buf_reader = BufReader::new(pcp_conf); - /* According to man 5 pcp.conf, syntax rules for pcp.conf are - 1. general syntax is PCP_VARIABLE_NAME=value to end of line - 2. blank lines and lines begining with # are ignored - 3. variable names that aren't prefixed with PCP_ are silently ignored - 4. there should be no space between the variable name and the literal = - 5. values may contain spaces and should not be quoted - */ - lazy_static! { - static ref RE: Regex = - Regex::new("(?-u)^(PCP_[[:alnum:]_]+)=([^\"\'].*[^\"\'])\n$").unwrap(); - } - let mut line = Vec::new(); while buf_reader.read_until(b'\n', &mut line)? > 0 { - match RE.captures(&line) { - Some(caps) => match (caps.get(1), caps.get(2)) { - (Some(key), Some(val)) => env::set_var( - osstr_from_bytes(key.as_bytes()), - osstr_from_bytes(val.as_bytes()), - ), - _ => {} - }, - _ => {} + if let Some((key, val)) = parse_pcp_conf_line(&line) { + if let (Some(key), Some(val)) = (osstr_from_bytes(key), osstr_from_bytes(val)) { + env::set_var(key, val); + } } line.clear(); } @@ -306,7 +313,7 @@ impl Client { // unlock header; has to be done last c.set_position(ws.gen2_off); - c.write_i64::(ws.gen)?; + c.write_i64(ws.gen)?; Ok(()) } @@ -332,24 +339,27 @@ fn write_mmv_header( // version match mmv_ver { - Version::V1 => c.write_u32::(1)?, - Version::V2 => c.write_u32::(2)?, + Version::V1 => c.write_u32(1)?, + Version::V2 => c.write_u32(2)?, } // generation1 - ws.gen = time::now().to_timespec().sec; - c.write_i64::(ws.gen)?; + ws.gen = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() as i64; + c.write_i64(ws.gen)?; // generation2 ws.gen2_off = c.position(); - c.write_i64::(0)?; + c.write_i64(0)?; // no. of toc blocks - c.write_u32::(ws.n_toc as u32)?; + c.write_u32(ws.n_toc as u32)?; // flags - c.write_u32::(ws.flags)?; + c.write_u32(ws.flags)?; // pid - c.write_i32::(get_process_id())?; + c.write_i32(get_process_id())?; // cluster id - c.write_u32::(ws.cluster_id) + c.write_u32(ws.cluster_id) } fn write_toc_block( @@ -360,18 +370,18 @@ fn write_toc_block( ) -> io::Result<()> { if entries > 0 { // section type - c.write_u32::(sec)?; + c.write_u32(sec)?; // no. of entries - c.write_u32::(entries)?; + c.write_u32(entries)?; // section offset - c.write_u64::(sec_off)?; + c.write_u64(sec_off)?; } Ok(()) } #[test] fn test_mmv_header() { - use byteorder::ReadBytesExt; + use crate::byteio::ReadBytesExt; use rand::{thread_rng, Rng}; let cluster_id = thread_rng().gen::(); @@ -392,20 +402,17 @@ fn test_mmv_header() { assert_eq!('V' as u8, cursor.read_u8().unwrap()); assert_eq!(0, cursor.read_u8().unwrap()); // test version number - assert_eq!(1, cursor.read_u32::().unwrap()); + assert_eq!(1, cursor.read_u32().unwrap()); // test generation - assert_eq!( - cursor.read_i64::().unwrap(), - cursor.read_i64::().unwrap() - ); + assert_eq!(cursor.read_i64().unwrap(), cursor.read_i64().unwrap()); // test no. of toc blocks - assert_eq!(0, cursor.read_i32::().unwrap()); + assert_eq!(0, cursor.read_i32().unwrap()); // test flags - assert_eq!(flags.bits(), cursor.read_u32::().unwrap()); + assert_eq!(flags.bits(), cursor.read_u32().unwrap()); // test pid - assert_eq!(get_process_id(), cursor.read_i32::().unwrap()); + assert_eq!(get_process_id(), cursor.read_i32().unwrap()); // cluster id - assert_eq!(client.cluster_id(), cursor.read_u32::().unwrap()); + assert_eq!(client.cluster_id(), cursor.read_u32().unwrap()); } #[test] diff --git a/src/lib.rs b/src/lib.rs index 24d3c31..3a82f89 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,16 +1,9 @@ -extern crate byteorder; extern crate hdrsample; extern crate memmap; -extern crate regex; -extern crate time; #[macro_use] extern crate bitflags; #[macro_use] extern crate lazy_static; -#[cfg(windows)] -extern crate kernel32; -#[cfg(unix)] -extern crate nix; #[cfg(test)] extern crate rand; @@ -32,10 +25,9 @@ const MMV1_NAME_MAX_LEN: u64 = 64; const INSTANCE_BLOCK_LEN_MMV2: u64 = 24; const METRIC_BLOCK_LEN_MMV2: u64 = 48; -type Endian = byteorder::LittleEndian; - #[macro_use] mod private; +mod byteio; pub mod client; pub mod mmv; diff --git a/src/mmv/mod.rs b/src/mmv/mod.rs index 7a68000..3b0fe7b 100644 --- a/src/mmv/mod.rs +++ b/src/mmv/mod.rs @@ -1,4 +1,4 @@ -use byteorder::ReadBytesExt; +use crate::byteio::ReadBytesExt; use std::collections::BTreeMap; use std::ffi::CStr; // Used to read null-terminated strings in MMV files use std::fmt; @@ -69,9 +69,7 @@ impl fmt::Display for MTCode { } } -use super::{ - Endian, CLUSTER_ID_BIT_LEN, INDOM_BIT_LEN, ITEM_BIT_LEN, MMV1_NAME_MAX_LEN, STRING_BLOCK_LEN, -}; +use super::{CLUSTER_ID_BIT_LEN, INDOM_BIT_LEN, ITEM_BIT_LEN, MMV1_NAME_MAX_LEN, STRING_BLOCK_LEN}; fn is_valid_indom(indom: u32) -> bool { indom != 0 && (indom >> INDOM_BIT_LEN) == 0 @@ -246,7 +244,7 @@ impl Header { return_mmvdumperror!("Invalid MMV", 0); } - let version = r.read_u32::()?; + let version = r.read_u32()?; let mmv_ver = match Version::from_u32(version) { Some(ver) => ver, None => { @@ -254,21 +252,21 @@ impl Header { } }; - let gen1 = r.read_i64::()?; - let gen2 = r.read_i64::()?; + let gen1 = r.read_i64()?; + let gen2 = r.read_i64()?; if gen1 != gen2 { return_mmvdumperror!("Generation timestamps don't match", 0); } - let toc_count = r.read_u32::()?; + let toc_count = r.read_u32()?; if toc_count > 5 || toc_count < 2 { return_mmvdumperror!("Invalid TOC count", toc_count); } - let flags = r.read_u32::()?; - let pid = r.read_i32::()?; + let flags = r.read_u32()?; + let pid = r.read_i32()?; - let cluster_id = r.read_u32::()?; + let cluster_id = r.read_u32()?; if !is_valid_cluster_id(cluster_id) { return_mmvdumperror!("Invalid cluster ID", cluster_id); } @@ -318,14 +316,14 @@ impl TocBlk { impl TocBlk { fn from_reader(r: &mut R) -> Result { - let sec = r.read_u32::()?; + let sec = r.read_u32()?; if sec > 5 { return_mmvdumperror!("Invalid TOC type", sec); } - let entries = r.read_u32::()?; + let entries = r.read_u32()?; - let sec_offset = r.read_u64::()?; + let sec_offset = r.read_u64()?; if !is_valid_blk_offset(sec_offset) { return_mmvdumperror!("Invalid section offset", sec_offset); } @@ -403,22 +401,22 @@ impl MetricBlk { let cstr = unsafe { CStr::from_ptr(name_bytes.as_ptr() as *const i8) }; VersionSpecificString::String(cstr.to_str()?.to_owned()) } - Version::V2 => VersionSpecificString::Offset(r.read_u64::()?), + Version::V2 => VersionSpecificString::Offset(r.read_u64()?), }; - let item = r.read_u32::()?; - let typ = r.read_u32::()?; - let sem = r.read_u32::()?; - let unit = r.read_u32::()?; - let indom = r.read_u32::()?; + let item = r.read_u32()?; + let typ = r.read_u32()?; + let sem = r.read_u32()?; + let unit = r.read_u32()?; + let indom = r.read_u32()?; - let pad = r.read_u32::()?; + let pad = r.read_u32()?; if pad != 0 { return_mmvdumperror!("Invalid pad bytes", pad); } - let short_help_offset = r.read_u64::()?; - let long_help_offset = r.read_u64::()?; + let short_help_offset = r.read_u64()?; + let long_help_offset = r.read_u64()?; Ok(MetricBlk { name: name, @@ -486,10 +484,10 @@ impl ValueBlk { impl ValueBlk { fn from_reader(r: &mut R) -> Result { - let value = r.read_u64::()?; - let string_offset = r.read_u64::()?; - let metric_offset = r.read_u64::()?; - let instance_offset = r.read_u64::()?; + let value = r.read_u64()?; + let string_offset = r.read_u64()?; + let metric_offset = r.read_u64()?; + let instance_offset = r.read_u64()?; Ok(ValueBlk { value: value, @@ -550,11 +548,11 @@ impl IndomBlk { impl IndomBlk { fn from_reader(r: &mut R) -> Result { - let indom = r.read_u32::()?; - let instances = r.read_u32::()?; - let instances_offset = r.read_u64::()?; - let short_help_offset = r.read_u64::()?; - let long_help_offset = r.read_u64::()?; + let indom = r.read_u32()?; + let instances = r.read_u32()?; + let instances_offset = r.read_u64()?; + let short_help_offset = r.read_u64()?; + let long_help_offset = r.read_u64()?; Ok(IndomBlk { indom: { @@ -618,14 +616,14 @@ impl InstanceBlk { impl InstanceBlk { fn from_reader(r: &mut R, ver: Version) -> Result { - let indom_offset = r.read_u64::()?; + let indom_offset = r.read_u64()?; - let pad = r.read_u32::()?; + let pad = r.read_u32()?; if pad != 0 { return_mmvdumperror!("Invalid pad bytes", pad); } - let internal_id = r.read_i32::()?; + let internal_id = r.read_i32()?; let external_id = match ver { Version::V1 => { @@ -634,7 +632,7 @@ impl InstanceBlk { let cstr = unsafe { CStr::from_ptr(external_id_bytes.as_ptr() as *const i8) }; VersionSpecificString::String(cstr.to_str()?.to_owned()) } - Version::V2 => VersionSpecificString::Offset(r.read_u64::()?), + Version::V2 => VersionSpecificString::Offset(r.read_u64()?), }; Ok(InstanceBlk { From d5fd0af2e2415dd37f72b9a27e6b87408a598b09 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Veljko=20Rvovi=C4=87?= Date: Wed, 29 Jul 2026 11:26:36 +0000 Subject: [PATCH 4/4] Migrate from memmap to memmap2 memmap 0.5.x is unmaintained (last release 2016) and had known soundness issues around unmapping memory. memmap2 is the actively maintained successor. memmap 0.5's MmapViewSync let a single mapping be split into independently-owned "views" over disjoint sub-ranges (via split_at), which is how each Metric/Instance got a private handle to just its own value's bytes for later updates. memmap2 has no equivalent - it just hands out one MmapMut for the whole mapping and expects callers to handle everything else themselves. Replaced that with a small MmapView type (in client::metric::private, alongside MMVWriterState): an Arc> plus an offset/len, cheaply cloneable and slice-able. It will now be used to hand out sub-views, with the Mutex taken for the read/write. This is different from the old design in one respect: writes to disjoint sub-views are now serialized through a shared lock rather than being raw pointer writes. This is strictly safer and shouldn't be observable functionally. Client::export() now maps the freshly-created file with memmap2::MmapMut::map_mut and takes the lock once, via MmapView::lock_whole, for the duration of the header/TOC/metric writing. The old code instead took an unsafe aliasing clone of the view and called as_mut_slice on it, so the whole-file cursor and the per-value views were two overlapping mutable handles to one mapping; holding the lock replaces that with a single writer. No API changes for consumers of the crate. Assisted-by: Cursor:claude-opus-5 --- Cargo.toml | 2 +- src/client/metric/mod.rs | 126 ++++++++++++++++++++++++++++----------- src/client/mod.rs | 16 +++-- src/lib.rs | 2 +- 4 files changed, 102 insertions(+), 44 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 3871a10..d4b477c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,7 @@ keywords = ["performance", "instrumentation", "metric", "pcp", "mmv"] bitflags = "0.9.1" hdrsample = "4.0.0" lazy_static = "0.2.8" -memmap = "0.5.2" +memmap2 = "0.9" [dev-dependencies] rand = "0.3.15" diff --git a/src/client/metric/mod.rs b/src/client/metric/mod.rs index 8a3192e..4cab41d 100644 --- a/src/client/metric/mod.rs +++ b/src/client/metric/mod.rs @@ -1,5 +1,4 @@ use crate::byteio::WriteBytesExt; -use memmap::{Mmap, MmapViewSync, Protection}; use std::collections::hash_map::{DefaultHasher, HashMap}; use std::collections::hash_set::Iter; use std::collections::HashSet; @@ -55,12 +54,74 @@ mod private { fn write(&self, writer: &mut W) -> io::Result<()>; } - use memmap::MmapViewSync; + use memmap2::MmapMut; use std::collections::HashMap; + use std::sync::{Arc, Mutex, MutexGuard}; + + /// A cloneable handle to a byte range within a shared memory map, letting + /// each metric value update its own bytes. + /// + /// Sub-views from `slice` must be kept disjoint: the mutex keeps writes + /// memory-safe, but overlapping ranges clobber each other's values. + #[derive(Clone)] + pub struct MmapView { + mmap: Arc>, + offset: usize, + len: usize, + } + + impl MmapView { + /// Placeholder mapping for metrics that haven't been exported yet. + pub fn anonymous(len: usize) -> io::Result { + Ok(MmapView::whole(MmapMut::map_anon(len)?)) + } + + pub fn whole(mmap: MmapMut) -> Self { + let len = mmap.len(); + MmapView { + mmap: Arc::new(Mutex::new(mmap)), + offset: 0, + len, + } + } + + /// `offset` is relative to this view, not to the underlying map. + pub fn slice(&self, offset: usize, len: usize) -> io::Result { + if offset.checked_add(len).map_or(true, |end| end > self.len) { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "MmapView::slice out of bounds", + )); + } + Ok(MmapView { + mmap: self.mmap.clone(), + offset: self.offset + offset, + len, + }) + } + + /// Locks the whole underlying map, not just this view's range. Held + /// for the length of an export, so nothing called under it may take + /// the same lock. + pub fn lock_whole(&self) -> MutexGuard<'_, MmapMut> { + self.mmap.lock().unwrap() + } + + pub fn write_value(&mut self, value: &T) -> io::Result<()> { + let mut guard = self.mmap.lock().unwrap(); + let mut slice = &mut guard[self.offset..self.offset + self.len]; + value.write(&mut slice) + } + + pub fn to_vec(&self) -> Vec { + let guard = self.mmap.lock().unwrap(); + guard[self.offset..self.offset + self.len].to_vec() + } + } pub struct MMVWriterState { // Mmap view of the entier MMV file - pub mmap_view: Option, + pub mmap_view: Option, // generation numbers pub gen: i64, @@ -158,7 +219,7 @@ mod private { } pub(super) use self::private::MetricType; -pub(super) use self::private::{MMVWriter, MMVWriterState}; +pub(super) use self::private::{MMVWriter, MMVWriterState, MmapView}; macro_rules! impl_metric_type_for ( ($typ:tt, $base_typ:tt, $type_code:expr) => ( @@ -520,15 +581,11 @@ pub struct Metric { shorthelp: String, longhelp: String, val: T, - mmap_view: MmapViewSync, + mmap_view: MmapView, } lazy_static! { - static ref SCRATCH_VIEW: MmapViewSync = { - Mmap::anonymous(STRING_BLOCK_LEN as usize, Protection::ReadWrite) - .unwrap() - .into_view_sync() - }; + static ref SCRATCH_VIEW: MmapView = MmapView::anonymous(STRING_BLOCK_LEN as usize).unwrap(); } impl Metric { @@ -573,7 +630,7 @@ impl Metric { shorthelp: shorthelp.to_owned(), longhelp: longhelp.to_owned(), val: init_val, - mmap_view: unsafe { SCRATCH_VIEW.clone() }, + mmap_view: SCRATCH_VIEW.clone(), }) } @@ -590,7 +647,7 @@ impl Metric { /// If the metric isn't exported, this method will still /// succeed and update the value. pub fn set_val(&mut self, new_val: T) -> io::Result<()> { - new_val.write(unsafe { &mut self.mmap_view.as_mut_slice() })?; + self.mmap_view.write_value(&new_val)?; self.val = new_val; Ok(()) } @@ -706,7 +763,7 @@ impl Indom { struct Instance { val: T, - mmap_view: MmapViewSync, + mmap_view: MmapView, } /// An instance metric is a set of related metrics with same @@ -736,7 +793,7 @@ impl InstanceMetric { for instance_str in &indom.instances { let instance = Instance { val: init_val.clone(), - mmap_view: unsafe { SCRATCH_VIEW.clone() }, + mmap_view: SCRATCH_VIEW.clone(), }; vals.insert(instance_str.to_owned(), instance); } @@ -770,7 +827,7 @@ impl InstanceMetric { /// found, returns `None`. pub fn set_val(&mut self, instance: &str, new_val: T) -> Option> { self.vals.get_mut(instance).map(|i| { - new_val.write(unsafe { &mut i.mmap_view.as_mut_slice() })?; + i.mmap_view.write_value(&new_val)?; i.val = new_val; Ok(()) }) @@ -847,9 +904,11 @@ impl Metric { let (value_offset, value_size) = write_value_block(ws, c, &self.val, metric_blk_off, 0)?; - let mmap_view = unsafe { ws.mmap_view.as_mut().unwrap().clone() }; - let (_, value_mmap_view, _) = three_way_split(mmap_view, value_offset, value_size)?; - self.mmap_view = value_mmap_view; + self.mmap_view = ws + .mmap_view + .as_ref() + .unwrap() + .slice(value_offset, value_size)?; } ws.metric_blk_idx += 1; @@ -915,9 +974,11 @@ impl MMVWriter for InstanceMetric { write_value_block(ws, c, &instance.val, metric_blk_off, instance_blk_off)?; // set mmap_view for instance - let mmap_view = unsafe { ws.mmap_view.as_mut().unwrap().clone() }; - let (_, value_mmap_view, _) = three_way_split(mmap_view, value_offset, value_size)?; - instance.mmap_view = value_mmap_view; + instance.mmap_view = ws + .mmap_view + .as_ref() + .unwrap() + .slice(value_offset, value_size)?; } Ok(()) @@ -1030,16 +1091,6 @@ fn write_indom_and_instances<'a>( Ok(cloned_offs) } -fn three_way_split( - view: MmapViewSync, - mid_idx: usize, - mid_len: usize, -) -> io::Result<(MmapViewSync, MmapViewSync, MmapViewSync)> { - let (left_view, mid_right_view) = view.split_at(mid_idx).unwrap(); - let (mid_view, right_view) = mid_right_view.split_at(mid_len).unwrap(); - Ok((left_view, mid_view, right_view)) -} - // writes `value` at end of value section, updates value count in value TOC, // and returns the offset `val` was written at and it's size - (offset, size) // @@ -1408,7 +1459,8 @@ fn test_random_numeric_metrics() { } for (m, v) in metrics.iter_mut().zip(new_vals) { - let mut slice = unsafe { m.mmap_view.as_slice() }; + let bytes = m.mmap_view.to_vec(); + let mut slice = &bytes[..]; assert_eq!(v, slice.read_u64().unwrap() as u32); } } @@ -1469,16 +1521,18 @@ fn test_simple_metrics() { let new_photon_count = thread_rng().gen::(); assert!(photons.set_val(new_photon_count).is_ok()); - let mut freq_slice = unsafe { freq.mmap_view.as_slice() }; + let freq_bytes = freq.mmap_view.to_vec(); + let mut freq_slice = &freq_bytes[..]; assert_eq!(new_freq, unsafe { transmute::(freq_slice.read_u64().unwrap()) }); - let color_slice = unsafe { color.mmap_view.as_slice() }; - let cstr = unsafe { CStr::from_ptr(color_slice.as_ptr() as *const i8) }; + let color_bytes = color.mmap_view.to_vec(); + let cstr = unsafe { CStr::from_ptr(color_bytes.as_ptr() as *const i8) }; assert_eq!(new_color, cstr.to_str().unwrap()); - let mut photon_slice = unsafe { photons.mmap_view.as_slice() }; + let photon_bytes = photons.mmap_view.to_vec(); + let mut photon_slice = &photon_bytes[..]; assert_eq!(new_photon_count, photon_slice.read_u64().unwrap() as u32); // TODO: after implementing mmvdump functionality, test the diff --git a/src/client/mod.rs b/src/client/mod.rs index 698d41c..383b335 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -1,5 +1,5 @@ use crate::byteio::WriteBytesExt; -use memmap::{Mmap, Protection}; +use memmap2::MmapMut; use std::env; use std::ffi::{OsStr, OsString}; use std::fmt; @@ -20,7 +20,7 @@ use super::{ }; pub mod metric; -use self::metric::{MMVWriter, MMVWriterState}; +use self::metric::{MMVWriter, MMVWriterState, MmapView}; static PCP_TMP_DIR_KEY: &'static str = "PCP_TMP_DIR"; static MMV_DIR_SUFFIX: &'static str = "mmv"; @@ -292,13 +292,17 @@ impl Client { file.write_all(&vec![0; mmv_size])?; - ws.mmap_view = Some(Mmap::open(&file, Protection::ReadWrite)?.into_view_sync()); - - let mut mmap_view = unsafe { ws.mmap_view.as_mut().unwrap().clone() }; - let mut c = Cursor::new(unsafe { mmap_view.as_mut_slice() }); + // Safety: the file was just created and zero-filled above, and no + // other writer has it open. + let mmap_view = MmapView::whole(unsafe { MmapMut::map_mut(&file)? }); + ws.mmap_view = Some(mmap_view.clone()); ws.flags = self.flags.bits(); ws.cluster_id = self.cluster_id; + + let mut guard = mmap_view.lock_whole(); + let mut c = Cursor::new(&mut guard[..]); + write_mmv_header(&mut ws, &mut c, mmv_ver)?; write_toc_block(1, ws.n_indoms as u32, ws.indom_sec_off, &mut c)?; diff --git a/src/lib.rs b/src/lib.rs index 3a82f89..b6ec9e0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,5 +1,5 @@ extern crate hdrsample; -extern crate memmap; +extern crate memmap2; #[macro_use] extern crate bitflags; #[macro_use]