23920642300

Committed 02 Apr 2026 08:32PM UTC coverage: 94.091%. First build

Build # 23920642300

Build Type

Pull #149

github

Committed by

web-flow

Commit Message

Merge 52bd7a9de into b91526c2e

Pull Request Pull Request #149: feat: implement always-file architecture for daemon synchronization

Coverage Stats

25 of 45 new or added lines in 2 files covered. (55.56%)

1895 of 2014 relevant lines covered (94.09%)

12.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.21

/src/kmsg.rs

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) NVIDIA CORPORATION

use crate::macros::ResultExt;
use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader};
use std::os::unix::fs::OpenOptionsExt;
use std::sync::Once;
use std::time::{Duration, Instant};

static KERNLOG_INIT: Once = Once::new();

/// Socket buffer size (16MB = 16 * 1024 * 1024 = 16777216 bytes).
/// Large buffers prevent message loss during high-throughput GPU operations
/// where NVIDIA drivers may emit bursts of diagnostic data.
const SOCKET_BUFFER_SIZE: &str = "16777216";

/// Initialize kernel logging and tune socket buffer sizes.
/// Large buffers (16MB) prevent message loss during high-throughput GPU operations
/// where drivers may emit bursts of diagnostic data.
pub fn kernlog_setup() {
    KERNLOG_INIT.call_once(|| {
        let _ = kernlog::init();
    });
    log::set_max_level(log::LevelFilter::Off);
    for path in [
        "/proc/sys/net/core/rmem_default",
        "/proc/sys/net/core/wmem_default",
        "/proc/sys/net/core/rmem_max",
        "/proc/sys/net/core/wmem_max",
    ] {
        fs::write(path, SOCKET_BUFFER_SIZE.as_bytes()).or_panic(format_args!("write {path}"));
    }
}

/// Get a file handle for kernel message output.
/// Routes to /dev/kmsg when debug logging is enabled for visibility in dmesg,
/// otherwise /dev/null to suppress noise in production.
pub fn kmsg() -> File {
    kmsg_at(if log_enabled!(log::Level::Debug) {
        "/dev/kmsg"
    } else {
        "/dev/null"
    })
}

/// Open syslog file for reading daemon startup markers.
/// Maps /dev/kmsg to /run/syslog.log because daemon synchronization needs to
/// work without trace logging enabled. File-based sync is simpler and more
/// reliable than trying to coordinate log levels between writer and reader.
pub fn open_kmsg(path: &str) -> BufReader<File> {
    let log_path = if path == "/dev/kmsg" {
        crate::syslog::SYSLOG_FILE_PATH
    } else {
        path
    };

    // Try read-only first; if missing, create with secure perms then reopen
    let file = OpenOptions::new()
        .read(true)
        .custom_flags(libc::O_NONBLOCK)
        .open(log_path)
        .or_else(|e| {
            if e.kind() == std::io::ErrorKind::NotFound {
                // Create with restrictive permissions
                OpenOptions::new()
                    .write(true)
                    .create_new(true)
                    .mode(0o600)
                    .open(log_path)?;
                // Reopen read-only
                OpenOptions::new()
                    .read(true)
                    .custom_flags(libc::O_NONBLOCK)
                    .open(log_path)
            } else {
                Err(e)
            }
        })
        .or_panic(format_args!("open {log_path}"));

    BufReader::new(file)
}

/// Block until `marker` appears in `reader` or `timeout_secs` expires.
/// Calls try_poll() to drain /dev/log socket and write messages to file that
/// we're reading from. This loop is the syslog daemon for our minimal init.
pub fn wait_for_marker(reader: &mut BufReader<File>, marker: &str, timeout_secs: u32) {
    let deadline = Instant::now() + Duration::from_secs(timeout_secs as u64);
    let mut line = String::new();

    loop {
        crate::syslog::try_poll();
        if Instant::now() > deadline {
            panic!("timeout waiting for: {marker}");
        }
        line.clear();
        match reader.read_line(&mut line) {
            Ok(0) => std::thread::sleep(Duration::from_millis(500)),
            Ok(_) if line.contains(marker) => {
                info!("{marker}");
                return;
            }
            Ok(_) => {}
            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                std::thread::sleep(Duration::from_millis(500));
            }
            Err(_) => std::thread::sleep(Duration::from_millis(500)),
        }
    }
}

/// Internal: open the given path for writing. Extracted for testability.
fn kmsg_at(path: &str) -> File {
    OpenOptions::new()
        .write(true)
        .open(path)
        .or_panic(format_args!("open {path}"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_utils::require_root;
    use serial_test::serial;
    use std::io::Write;
    use std::panic;
    use tempfile::NamedTempFile;

    #[test]
    fn test_kmsg_at_dev_null() {
        // /dev/null is always writable, no root needed
        let _file = kmsg_at("/dev/null");
    }

    #[test]
    fn test_kmsg_at_nonexistent() {
        let result = panic::catch_unwind(|| {
            kmsg_at("/nonexistent/path");
        });
        assert!(result.is_err());
    }

    #[test]
    fn test_kmsg_at_temp_file() {
        // Create a temp file to verify we can write to it
        let temp = NamedTempFile::new().unwrap();
        let path = temp.path().to_str().unwrap();
        let mut file = kmsg_at(path);
        assert!(file.write_all(b"test").is_ok());
    }

    #[test]
    #[serial]
    fn test_kmsg_routes_to_dev_null_when_log_off() {
        // Default log level is Off, so kmsg() should open /dev/null
        log::set_max_level(log::LevelFilter::Off);
        let _file = kmsg();
    }

    #[test]
    #[serial]
    fn test_kmsg_routes_to_kmsg_when_debug() {
        require_root();
        // When debug is enabled, kmsg() should open /dev/kmsg
        log::set_max_level(log::LevelFilter::Debug);
        let _file = kmsg();
        log::set_max_level(log::LevelFilter::Off);
    }

    #[test]
    #[serial]
    fn test_kernlog_setup() {
        require_root();

        const PATHS: [&str; 4] = [
            "/proc/sys/net/core/rmem_default",
            "/proc/sys/net/core/wmem_default",
            "/proc/sys/net/core/rmem_max",
            "/proc/sys/net/core/wmem_max",
        ];

        // RAII guard to restore original values after test
        struct Restore(Vec<(&'static str, String)>);
        impl Drop for Restore {
            fn drop(&mut self) {
                for (path, value) in &self.0 {
                    let _ = fs::write(path, value.as_bytes());
                }
            }
        }

        let saved: Vec<_> = PATHS
            .iter()
            .filter_map(|&p| fs::read_to_string(p).ok().map(|v| (p, v)))
            .collect();
        let _restore = Restore(saved);

        kernlog_setup();

        for &path in &PATHS {
            let v = fs::read_to_string(path).expect("should read sysctl");
            assert_eq!(
                v.trim(),
                SOCKET_BUFFER_SIZE,
                "sysctl {} should be {}",
                path,
                SOCKET_BUFFER_SIZE
            );
        }
    }

    // === wait_for_marker tests ===

    #[test]
    fn test_wait_for_marker_finds_marker() {
        let mut tmp = NamedTempFile::new().unwrap();
        writeln!(tmp, "some noise").unwrap();
        writeln!(tmp, "FM starting NvLink Inband foo").unwrap();
        writeln!(tmp, "more noise").unwrap();
        tmp.flush().unwrap();

        wait_for_marker(
            &mut open_kmsg(tmp.path().to_str().unwrap()),
            "FM starting NvLink Inband",
            5,
        );
    }

    #[test]
    fn test_wait_for_marker_finds_marker_at_end() {
        let mut tmp = NamedTempFile::new().unwrap();
        writeln!(tmp, "line 1").unwrap();
        writeln!(tmp, "line 2").unwrap();
        writeln!(tmp, "FM starting NvLink Inband").unwrap();
        tmp.flush().unwrap();

        wait_for_marker(
            &mut open_kmsg(tmp.path().to_str().unwrap()),
            "FM starting NvLink Inband",
            5,
        );
    }

    #[test]
    fn test_wait_for_marker_no_marker_panics() {
        let mut tmp = NamedTempFile::new().unwrap();
        writeln!(tmp, "no match here").unwrap();
        tmp.flush().unwrap();

        let result = panic::catch_unwind(|| {
            wait_for_marker(
                &mut open_kmsg(tmp.path().to_str().unwrap()),
                "FM starting NvLink Inband",
                1,
            );
        });
        assert!(result.is_err());
    }

    #[test]
    fn test_wait_for_marker_empty_file_panics() {
        let tmp = NamedTempFile::new().unwrap();

        let result = panic::catch_unwind(|| {
            wait_for_marker(
                &mut open_kmsg(tmp.path().to_str().unwrap()),
                "FM starting NvLink Inband",
                1,
            );
        });
        assert!(result.is_err());
    }

    #[test]
    fn test_wait_for_marker_nonexistent_file_panics() {
        let result = panic::catch_unwind(|| {
            wait_for_marker(&mut open_kmsg("/nonexistent/path"), "marker", 1);
        });
        assert!(result.is_err());
    }

    #[test]
    fn test_wait_for_marker_partial_match_not_enough() {
        let mut tmp = NamedTempFile::new().unwrap();
        writeln!(tmp, "FM starting").unwrap();
        writeln!(tmp, "NvLink Inband").unwrap();
        tmp.flush().unwrap();

        // Marker spans two lines — should not match
        let result = panic::catch_unwind(|| {
            wait_for_marker(
                &mut open_kmsg(tmp.path().to_str().unwrap()),
                "FM starting NvLink Inband",
                1,
            );
        });
        assert!(result.is_err());
    }

    #[test]
    #[serial]
    fn test_wait_for_marker_on_dev_kmsg() {
        require_root();

        // Clear any previous test data to avoid false positives
        let _ = fs::remove_file(crate::syslog::SYSLOG_FILE_PATH);

        // With always-file architecture, open_kmsg("/dev/kmsg") always reads from syslog file
        let mut reader = open_kmsg("/dev/kmsg");
        let marker = "NVRC_TEST_MARKER_12345";

        // Write directly to the syslog file (simulating what syslog.rs does)
        let mut file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(crate::syslog::SYSLOG_FILE_PATH)
            .expect("open syslog file");
        writeln!(file, "{}", marker).expect("write marker");
        file.flush().expect("flush");

        wait_for_marker(&mut reader, marker, 5);
    }
}

1	// SPDX-License-Identifier: Apache-2.0
2	// Copyright (c) NVIDIA CORPORATION
3
4	use crate::macros::ResultExt;
5	use std::fs::{self, File, OpenOptions};
6	use std::io::{BufRead, BufReader};
7	use std::os::unix::fs::OpenOptionsExt;
8	use std::sync::Once;
9	use std::time::{Duration, Instant};
10
11	static KERNLOG_INIT: Once = Once::new();
12
13	/// Socket buffer size (16MB = 16 * 1024 * 1024 = 16777216 bytes).
14	/// Large buffers prevent message loss during high-throughput GPU operations
15	/// where NVIDIA drivers may emit bursts of diagnostic data.
16	const SOCKET_BUFFER_SIZE: &str = "16777216";
17
18	/// Initialize kernel logging and tune socket buffer sizes.
19	/// Large buffers (16MB) prevent message loss during high-throughput GPU operations
20	/// where drivers may emit bursts of diagnostic data.
21	pub fn kernlog_setup() {	2✔
22	KERNLOG_INIT.call_once(\|\| {	2✔
23	let _ = kernlog::init();	2✔
24	});	2✔
25	log::set_max_level(log::LevelFilter::Off);	2✔
26	for path in [	8✔
27	"/proc/sys/net/core/rmem_default",	2✔
28	"/proc/sys/net/core/wmem_default",	2✔
29	"/proc/sys/net/core/rmem_max",	2✔
30	"/proc/sys/net/core/wmem_max",	2✔
31	] {	8✔
32	fs::write(path, SOCKET_BUFFER_SIZE.as_bytes()).or_panic(format_args!("write {path}"));	8✔
33	}	8✔
34	}	2✔
35
36	/// Get a file handle for kernel message output.
37	/// Routes to /dev/kmsg when debug logging is enabled for visibility in dmesg,
38	/// otherwise /dev/null to suppress noise in production.
39	pub fn kmsg() -> File {	156✔
40	kmsg_at(if log_enabled!(log::Level::Debug) {	156✔
41	"/dev/kmsg"	2✔
42	} else {
43	"/dev/null"	154✔
44	})
45	}	156✔
46
47	/// Open syslog file for reading daemon startup markers.
48	/// Maps /dev/kmsg to /run/syslog.log because daemon synchronization needs to
49	/// work without trace logging enabled. File-based sync is simpler and more
50	/// reliable than trying to coordinate log levels between writer and reader.
51	pub fn open_kmsg(path: &str) -> BufReader<File> {	14✔
52	let log_path = if path == "/dev/kmsg" {	14✔
53	crate::syslog::SYSLOG_FILE_PATH	2✔
54	} else {
55	path	12✔
56	};
57
58	// Try read-only first; if missing, create with secure perms then reopen
59	let file = OpenOptions::new()	14✔
60	.read(true)	14✔
61	.custom_flags(libc::O_NONBLOCK)	14✔
62	.open(log_path)	14✔
63	.or_else(\|e\| {	14✔
64	if e.kind() == std::io::ErrorKind::NotFound {	4✔
65	// Create with restrictive permissions
66	OpenOptions::new()	4✔
67	.write(true)	4✔
68	.create_new(true)	4✔
69	.mode(0o600)	4✔
70	.open(log_path)?;	4✔
71	// Reopen read-only
72	OpenOptions::new()	2✔
73	.read(true)	2✔
74	.custom_flags(libc::O_NONBLOCK)	2✔
75	.open(log_path)	2✔
76	} else {
NEW 77	Err(e)	×
78	}
79	})	4✔
80	.or_panic(format_args!("open {log_path}"));	14✔
81
82	BufReader::new(file)	14✔
83	}	14✔
84
85	/// Block until `marker` appears in `reader` or `timeout_secs` expires.
86	/// Calls try_poll() to drain /dev/log socket and write messages to file that
87	/// we're reading from. This loop is the syslog daemon for our minimal init.
88	pub fn wait_for_marker(reader: &mut BufReader<File>, marker: &str, timeout_secs: u32) {	12✔
89	let deadline = Instant::now() + Duration::from_secs(timeout_secs as u64);	12✔
90	let mut line = String::new();	12✔
91
92	loop {
93	crate::syslog::try_poll();	36✔
94	if Instant::now() > deadline {	36✔
95	panic!("timeout waiting for: {marker}");	6✔
96	}	30✔
97	line.clear();	30✔
98	match reader.read_line(&mut line) {	30✔
99	Ok(0) => std::thread::sleep(Duration::from_millis(500)),	12✔
100	Ok(_) if line.contains(marker) => {	18✔
101	info!("{marker}");	6✔
102	return;	6✔
103	}
104	Ok(_) => {}	12✔
105	Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {	×
106	std::thread::sleep(Duration::from_millis(500));	×
107	}	×
108	Err(_) => std::thread::sleep(Duration::from_millis(500)),	×
109	}
110	}
111	}	6✔
112
113	/// Internal: open the given path for writing. Extracted for testability.
114	fn kmsg_at(path: &str) -> File {	162✔
115	OpenOptions::new()	162✔
116	.write(true)	162✔
117	.open(path)	162✔
118	.or_panic(format_args!("open {path}"))	162✔
119	}	162✔
120
121	#[cfg(test)]
122	mod tests {
123	use super::*;
124	use crate::test_utils::require_root;
125	use serial_test::serial;
126	use std::io::Write;
127	use std::panic;
128	use tempfile::NamedTempFile;
129
130	#[test]
131	fn test_kmsg_at_dev_null() {	2✔
132	// /dev/null is always writable, no root needed
133	let _file = kmsg_at("/dev/null");	2✔
134	}	2✔
135
136	#[test]
137	fn test_kmsg_at_nonexistent() {	2✔
138	let result = panic::catch_unwind(\|\| {	2✔
139	kmsg_at("/nonexistent/path");	2✔
140	});	2✔
141	assert!(result.is_err());	2✔
142	}	2✔
143
144	#[test]
145	fn test_kmsg_at_temp_file() {	2✔
146	// Create a temp file to verify we can write to it
147	let temp = NamedTempFile::new().unwrap();	2✔
148	let path = temp.path().to_str().unwrap();	2✔
149	let mut file = kmsg_at(path);	2✔
150	assert!(file.write_all(b"test").is_ok());	2✔
151	}	2✔
152
153	#[test]
154	#[serial]
155	fn test_kmsg_routes_to_dev_null_when_log_off() {	2✔
156	// Default log level is Off, so kmsg() should open /dev/null
157	log::set_max_level(log::LevelFilter::Off);	2✔
158	let _file = kmsg();	2✔
159	}
160
161	#[test]
162	#[serial]
163	fn test_kmsg_routes_to_kmsg_when_debug() {	2✔
164	require_root();	2✔
165	// When debug is enabled, kmsg() should open /dev/kmsg
166	log::set_max_level(log::LevelFilter::Debug);	2✔
167	let _file = kmsg();	2✔
168	log::set_max_level(log::LevelFilter::Off);	2✔
169	}
170
171	#[test]
172	#[serial]
173	fn test_kernlog_setup() {	2✔
174	require_root();	2✔
175
176	const PATHS: [&str; 4] = [
177	"/proc/sys/net/core/rmem_default",
178	"/proc/sys/net/core/wmem_default",
179	"/proc/sys/net/core/rmem_max",
180	"/proc/sys/net/core/wmem_max",
181	];
182
183	// RAII guard to restore original values after test
184	struct Restore(Vec<(&'static str, String)>);
185	impl Drop for Restore {
186	fn drop(&mut self) {	2✔
187	for (path, value) in &self.0 {	8✔
188	let _ = fs::write(path, value.as_bytes());	8✔
189	}	8✔
190	}	2✔
191	}
192
193	let saved: Vec<_> = PATHS	2✔
194	.iter()	2✔
195	.filter_map(\|&p\| fs::read_to_string(p).ok().map(\|v\| (p, v)))	8✔
196	.collect();	2✔
197	let _restore = Restore(saved);	2✔
198
199	kernlog_setup();	2✔
200
201	for &path in &PATHS {	8✔
202	let v = fs::read_to_string(path).expect("should read sysctl");	8✔
203	assert_eq!(	8✔
204	v.trim(),	8✔
205	SOCKET_BUFFER_SIZE,
206	"sysctl {} should be {}",
207	path,
208	SOCKET_BUFFER_SIZE
209	);
210	}
211	}
212
213	// === wait_for_marker tests ===
214
215	#[test]
216	fn test_wait_for_marker_finds_marker() {	2✔
217	let mut tmp = NamedTempFile::new().unwrap();	2✔
218	writeln!(tmp, "some noise").unwrap();	2✔
219	writeln!(tmp, "FM starting NvLink Inband foo").unwrap();	2✔
220	writeln!(tmp, "more noise").unwrap();	2✔
221	tmp.flush().unwrap();	2✔
222
223	wait_for_marker(	2✔
224	&mut open_kmsg(tmp.path().to_str().unwrap()),	2✔
225	"FM starting NvLink Inband",	2✔
226	5,
227	);
228	}	2✔
229
230	#[test]
231	fn test_wait_for_marker_finds_marker_at_end() {	2✔
232	let mut tmp = NamedTempFile::new().unwrap();	2✔
233	writeln!(tmp, "line 1").unwrap();	2✔
234	writeln!(tmp, "line 2").unwrap();	2✔
235	writeln!(tmp, "FM starting NvLink Inband").unwrap();	2✔
236	tmp.flush().unwrap();	2✔
237
238	wait_for_marker(	2✔
239	&mut open_kmsg(tmp.path().to_str().unwrap()),	2✔
240	"FM starting NvLink Inband",	2✔
241	5,
242	);
243	}	2✔
244
245	#[test]
246	fn test_wait_for_marker_no_marker_panics() {	2✔
247	let mut tmp = NamedTempFile::new().unwrap();	2✔
248	writeln!(tmp, "no match here").unwrap();	2✔
249	tmp.flush().unwrap();	2✔
250
251	let result = panic::catch_unwind(\|\| {	2✔
252	wait_for_marker(	2✔
253	&mut open_kmsg(tmp.path().to_str().unwrap()),	2✔
254	"FM starting NvLink Inband",	2✔
255	1,
256	);
257	});	2✔
258	assert!(result.is_err());	2✔
259	}	2✔
260
261	#[test]
262	fn test_wait_for_marker_empty_file_panics() {	2✔
263	let tmp = NamedTempFile::new().unwrap();	2✔
264
265	let result = panic::catch_unwind(\|\| {	2✔
266	wait_for_marker(	2✔
267	&mut open_kmsg(tmp.path().to_str().unwrap()),	2✔
268	"FM starting NvLink Inband",	2✔
269	1,
270	);
271	});	2✔
272	assert!(result.is_err());	2✔
273	}	2✔
274
275	#[test]
276	fn test_wait_for_marker_nonexistent_file_panics() {	2✔
277	let result = panic::catch_unwind(\|\| {	2✔
278	wait_for_marker(&mut open_kmsg("/nonexistent/path"), "marker", 1);	2✔
279	});	2✔
280	assert!(result.is_err());	2✔
281	}	2✔
282
283	#[test]
284	fn test_wait_for_marker_partial_match_not_enough() {	2✔
285	let mut tmp = NamedTempFile::new().unwrap();	2✔
286	writeln!(tmp, "FM starting").unwrap();	2✔
287	writeln!(tmp, "NvLink Inband").unwrap();	2✔
288	tmp.flush().unwrap();	2✔
289
290	// Marker spans two lines — should not match
291	let result = panic::catch_unwind(\|\| {	2✔
292	wait_for_marker(	2✔
293	&mut open_kmsg(tmp.path().to_str().unwrap()),	2✔
294	"FM starting NvLink Inband",	2✔
295	1,
296	);
297	});	2✔
298	assert!(result.is_err());	2✔
299	}	2✔
300
301	#[test]
302	#[serial]
303	fn test_wait_for_marker_on_dev_kmsg() {	2✔
304	require_root();	2✔
305
306	// Clear any previous test data to avoid false positives
307	let _ = fs::remove_file(crate::syslog::SYSLOG_FILE_PATH);	2✔
308
309	// With always-file architecture, open_kmsg("/dev/kmsg") always reads from syslog file
310	let mut reader = open_kmsg("/dev/kmsg");	2✔
311	let marker = "NVRC_TEST_MARKER_12345";	2✔
312
313	// Write directly to the syslog file (simulating what syslog.rs does)
314	let mut file = OpenOptions::new()	2✔
315	.create(true)	2✔
316	.append(true)	2✔
317	.open(crate::syslog::SYSLOG_FILE_PATH)	2✔
318	.expect("open syslog file");	2✔
319	writeln!(file, "{}", marker).expect("write marker");	2✔
320	file.flush().expect("flush");	2✔
321
322	wait_for_marker(&mut reader, marker, 5);	2✔
323	}
324	}

NVIDIA / nvrc / 23920642300

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous