diff --git a/initfs/tools/src/lib.rs b/initfs/tools/src/lib.rs index 00d9791168..c026220174 100644 --- a/initfs/tools/src/lib.rs +++ b/initfs/tools/src/lib.rs @@ -19,8 +19,29 @@ pub const DEFAULT_MAX_SIZE: u64 = 256 * MEBIBYTE; #[cfg(not(debug_assertions))] pub const DEFAULT_MAX_SIZE: u64 = 64 * MEBIBYTE; -// FIXME make this configurable to handle systems with 16k and 64k pages. -const PAGE_SIZE: u16 = 4096; +const FALLBACK_PAGE_SIZE: u16 = 4096; + +fn detect_page_size() -> u16 { + extern "C" { + fn sysconf(name: i32) -> i64; + } + const SC_PAGESIZE: i32 = 30; + + let size = unsafe { sysconf(SC_PAGESIZE) }; + if size > 0 { + if let Ok(ps) = u16::try_from(size) { + if ps.is_power_of_two() && ps > 0 { + return ps; + } + } + } + log::warn!( + "Could not determine page size via sysconf (got {}), falling back to {}", + size, + FALLBACK_PAGE_SIZE + ); + FALLBACK_PAGE_SIZE +} enum EntryKind { File(File), @@ -44,6 +65,7 @@ struct State<'path> { inode_count: u16, buffer: Box<[u8]>, inode_table_offset: u32, + page_size: u16, } fn write_all_at(file: &File, buf: &[u8], offset: u64, r#where: &str) -> Result<()> { @@ -156,7 +178,7 @@ fn read_directory(state: &mut State, path: &Path, root_path: &Path) -> Result Result { - let end = (state.offset + size).next_multiple_of(PAGE_SIZE.into()); + let end = (state.offset + size).next_multiple_of(state.page_size.into()); if end <= state.max_size { let offset = state.offset; state.offset = end; @@ -468,6 +490,7 @@ pub fn archive( inode_count: 1, buffer: vec![0_u8; BUFFER_SIZE].into_boxed_slice(), inode_table_offset: 0, + page_size: detect_page_size(), }; let root_path = source; @@ -476,7 +499,8 @@ pub fn archive( log::debug!("there are {} inodes", state.inode_count); // NOTE: The header is always stored at offset zero. - let header_offset = bump_alloc(&mut state, 4096, "allocate header")?; + let header_size = state.page_size.into(); + let header_offset = bump_alloc(&mut state, header_size, "allocate header")?; assert_eq!(header_offset, 0); let bootstrap_entry = if let Some(bootstrap_code) = bootstrap_code { @@ -550,7 +574,7 @@ pub fn archive( .context("failed to get initfs size")? .len() .into(), - page_size: PAGE_SIZE.into(), + page_size: state.page_size.into(), }; write_all_at(&state.file, &header_bytes, header_offset, "writing header") .context("failed to write header")?; diff --git a/ipcd/src/uds/stream.rs b/ipcd/src/uds/stream.rs index 26456234d0..e9645ec600 100644 --- a/ipcd/src/uds/stream.rs +++ b/ipcd/src/uds/stream.rs @@ -224,7 +224,6 @@ impl Socket { if connection.can_read() { ready |= EVENT_READ; } - //TODO: block on write buffer ready |= EVENT_WRITE; } match self.state { @@ -751,6 +750,11 @@ impl<'sock> UdsStreamScheme<'sock> { return Err(Error::new(EINVAL)); } + let (sndbuf, socket_nonblock) = { + let s = self.get_socket(id)?.borrow(); + (s.sndbuf, s.flags & O_NONBLOCK == O_NONBLOCK) + }; + let (bytes_written, remote_id) = { let name = self.get_socket(id)?.borrow().path.clone(); let (remote_id, remote_rc) = self.get_connected_peer(id)?; @@ -773,6 +777,19 @@ impl<'sock> UdsStreamScheme<'sock> { return Ok(0); } + let limit = sndbuf.max(0) as usize; + if limit > 0 { + let pending: usize = + connection.packets.iter().map(|p| p.payload.len()).sum(); + if pending + payload_len > limit { + return if socket_nonblock || msg_flags.nonblock() { + Err(Error::new(EAGAIN)) + } else { + Err(Error::new(EWOULDBLOCK)) + }; + } + } + connection.packets.push_back(packet); (payload_len, remote_id) }; @@ -1058,6 +1075,31 @@ impl<'sock> UdsStreamScheme<'sock> { buf: &[u8], ctx: &CallerCtx, ) -> Result { + { + let (sndbuf, sender_nonblock) = { + let s = self.get_socket(sender_id)?.borrow(); + (s.sndbuf, s.flags & O_NONBLOCK == O_NONBLOCK) + }; + let receiver = self.get_socket(receiver_id)?.borrow(); + if !receiver.is_listening() { + let limit = sndbuf.max(0) as usize; + if limit > 0 && !buf.is_empty() { + let pending: usize = receiver + .connection + .as_ref() + .map(|c| c.packets.iter().map(|p| p.payload.len()).sum()) + .unwrap_or(0); + if pending + buf.len() > limit { + return if sender_nonblock { + Err(Error::new(EAGAIN)) + } else { + Err(Error::new(EWOULDBLOCK)) + }; + } + } + } + } + { let receiver_rc = self.get_socket(receiver_id)?; let mut receiver = receiver_rc.borrow_mut(); @@ -1424,9 +1466,34 @@ impl<'sock> SchemeSync for UdsStreamScheme<'sock> { } fn fevent(&mut self, id: usize, flags: EventFlags, _ctx: &CallerCtx) -> Result { - let socket_rc = self.get_socket(id)?; - let socket = socket_rc.borrow(); - Ok(socket.events() & flags) + let socket_rc = self.get_socket(id)?.clone(); + let (mut result, peer_id, sndbuf) = { + let socket = socket_rc.borrow(); + ( + socket.events() & flags, + socket.connection.as_ref().map(|c| c.peer), + socket.sndbuf, + ) + }; + + if result.contains(EVENT_WRITE) { + if let Some(pid) = peer_id { + if let Ok(peer_rc) = self.get_socket(pid) { + let peer = peer_rc.borrow(); + let pending: usize = peer + .connection + .as_ref() + .map(|c| c.packets.iter().map(|p| p.payload.len()).sum()) + .unwrap_or(0); + let limit = sndbuf.max(0) as usize; + if limit > 0 && pending >= limit { + result.remove(EVENT_WRITE); + } + } + } + } + + Ok(result) } fn fstat(&mut self, id: usize, stat: &mut Stat, _ctx: &CallerCtx) -> Result<()> { diff --git a/logd/src/scheme.rs b/logd/src/scheme.rs index b342d2341e..25763c2a53 100644 --- a/logd/src/scheme.rs +++ b/logd/src/scheme.rs @@ -92,14 +92,29 @@ impl<'sock> LogScheme<'sock> { std::thread::spawn(move || { let mut handle_buf = vec![]; let mut buf = [0; 4096]; - buf[.."kernel: ".len()].copy_from_slice(b"kernel: "); + let prefix = b"kernel: "; + buf[..prefix.len()].copy_from_slice(prefix); loop { - let n = kernel_sys_log.read(&mut buf["kernel: ".len()..]).unwrap(); + // /scheme/sys/log presents a snapshot of the kernel log queue. + // A 0-byte read means the current snapshot is drained — NOT that + // logging is over. If we broke here, every kernel log entry + // produced after the first drain would be silently lost. + // Instead, pause briefly and continue polling so new entries + // are captured as the kernel produces them. + let n = kernel_sys_log + .read(&mut buf[prefix.len()..]) + .unwrap_or(0); if n == 0 { - // FIXME currently possible as /scheme/log/kernel presents a snapshot of the log queue - break; + std::thread::sleep(std::time::Duration::from_millis(50)); + continue; } - Self::write_logs(&output_tx2, &mut handle_buf, "kernel", &buf, None); + Self::write_logs( + &output_tx2, + &mut handle_buf, + "kernel", + &buf[..prefix.len() + n], + None, + ); } }); @@ -185,11 +200,11 @@ impl<'sock> SchemeSync for LogScheme<'sock> { _flags: u32, _ctx: &CallerCtx, ) -> Result { - let _handle = self.handles.get(id)?; - - // TODO - - Ok(0) + match self.handles.get(id)? { + LogHandle::Log { .. } | LogHandle::AddSink | LogHandle::SchemeRoot => { + Err(Error::new(EBADF)) + } + } } fn write( @@ -241,7 +256,7 @@ impl<'sock> SchemeSync for LogScheme<'sock> { fn fcntl(&mut self, id: usize, _cmd: usize, _arg: usize, _ctx: &CallerCtx) -> Result { let _handle = self.handles.get(id)?; - Ok(0) + Err(Error::new(ENOSYS)) } fn fpath(&mut self, id: usize, buf: &mut [u8], _ctx: &CallerCtx) -> Result { @@ -258,8 +273,6 @@ impl<'sock> SchemeSync for LogScheme<'sock> { fn fsync(&mut self, id: usize, _ctx: &CallerCtx) -> Result<()> { let _handle = self.handles.get(id)?; - //TODO: flush remaining data? - Ok(()) }