logd/ipcd/initfs: fix WARNING-level stubs and error handling
logd scheme.rs: - Kernel log reader: don't break loop on 0-byte read. /scheme/sys/log presents a snapshot; breaking loses all future entries. Now polls with a 50ms sleep on drain, continuing to capture new kernel logs. Also slice buffer to actual bytes read to avoid processing stale data. - read(): return EBADF instead of silent Ok(0) stub. Logd handles are write-only sinks; reading is not a valid operation. - fcntl(): return ENOSYS instead of silent Ok(0). No fcntl operations are implemented for log handles. - fsync(): remove TODO comment; behavior (Ok(())) is correct since the output channel guarantees FIFO delivery. ipcd uds/stream.rs: - events(): remove 'TODO: block on write buffer'. Backpressure is now applied at the scheme level. - fevent(): filter EVENT_WRITE based on peer receive buffer fullness against sender's SO_SNDBUF. Prevents busy-spin when buffer is full. - write_inner(): check backpressure before accepting data. Return EAGAIN (non-blocking) or EWOULDBLOCK (blocking) when peer buffer would exceed SO_SNDBUF limit. Only applies to established connections. - handle_sendmsg(): same backpressure check for sendmsg path. initfs tools/src/lib.rs: - Replace hardcoded PAGE_SIZE=4096 constant with runtime detection via sysconf(_SC_PAGESIZE). Handles 16K/64K page ARM hardware correctly. Falls back to 4096 with logged warning if sysconf fails or returns an invalid value.
This commit is contained in:
+29
-5
@@ -19,8 +19,29 @@ pub const DEFAULT_MAX_SIZE: u64 = 256 * MEBIBYTE;
|
|||||||
#[cfg(not(debug_assertions))]
|
#[cfg(not(debug_assertions))]
|
||||||
pub const DEFAULT_MAX_SIZE: u64 = 64 * MEBIBYTE;
|
pub const DEFAULT_MAX_SIZE: u64 = 64 * MEBIBYTE;
|
||||||
|
|
||||||
// FIXME make this configurable to handle systems with 16k and 64k pages.
|
const FALLBACK_PAGE_SIZE: u16 = 4096;
|
||||||
const 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 {
|
enum EntryKind {
|
||||||
File(File),
|
File(File),
|
||||||
@@ -44,6 +65,7 @@ struct State<'path> {
|
|||||||
inode_count: u16,
|
inode_count: u16,
|
||||||
buffer: Box<[u8]>,
|
buffer: Box<[u8]>,
|
||||||
inode_table_offset: u32,
|
inode_table_offset: u32,
|
||||||
|
page_size: u16,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn write_all_at(file: &File, buf: &[u8], offset: u64, r#where: &str) -> Result<()> {
|
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<Di
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn bump_alloc(state: &mut State, size: u64, why: &str) -> Result<u64> {
|
fn bump_alloc(state: &mut State, size: u64, why: &str) -> Result<u64> {
|
||||||
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 {
|
if end <= state.max_size {
|
||||||
let offset = state.offset;
|
let offset = state.offset;
|
||||||
state.offset = end;
|
state.offset = end;
|
||||||
@@ -468,6 +490,7 @@ pub fn archive(
|
|||||||
inode_count: 1,
|
inode_count: 1,
|
||||||
buffer: vec![0_u8; BUFFER_SIZE].into_boxed_slice(),
|
buffer: vec![0_u8; BUFFER_SIZE].into_boxed_slice(),
|
||||||
inode_table_offset: 0,
|
inode_table_offset: 0,
|
||||||
|
page_size: detect_page_size(),
|
||||||
};
|
};
|
||||||
|
|
||||||
let root_path = source;
|
let root_path = source;
|
||||||
@@ -476,7 +499,8 @@ pub fn archive(
|
|||||||
log::debug!("there are {} inodes", state.inode_count);
|
log::debug!("there are {} inodes", state.inode_count);
|
||||||
|
|
||||||
// NOTE: The header is always stored at offset zero.
|
// 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);
|
assert_eq!(header_offset, 0);
|
||||||
|
|
||||||
let bootstrap_entry = if let Some(bootstrap_code) = bootstrap_code {
|
let bootstrap_entry = if let Some(bootstrap_code) = bootstrap_code {
|
||||||
@@ -550,7 +574,7 @@ pub fn archive(
|
|||||||
.context("failed to get initfs size")?
|
.context("failed to get initfs size")?
|
||||||
.len()
|
.len()
|
||||||
.into(),
|
.into(),
|
||||||
page_size: PAGE_SIZE.into(),
|
page_size: state.page_size.into(),
|
||||||
};
|
};
|
||||||
write_all_at(&state.file, &header_bytes, header_offset, "writing header")
|
write_all_at(&state.file, &header_bytes, header_offset, "writing header")
|
||||||
.context("failed to write header")?;
|
.context("failed to write header")?;
|
||||||
|
|||||||
+71
-4
@@ -224,7 +224,6 @@ impl Socket {
|
|||||||
if connection.can_read() {
|
if connection.can_read() {
|
||||||
ready |= EVENT_READ;
|
ready |= EVENT_READ;
|
||||||
}
|
}
|
||||||
//TODO: block on write buffer
|
|
||||||
ready |= EVENT_WRITE;
|
ready |= EVENT_WRITE;
|
||||||
}
|
}
|
||||||
match self.state {
|
match self.state {
|
||||||
@@ -751,6 +750,11 @@ impl<'sock> UdsStreamScheme<'sock> {
|
|||||||
return Err(Error::new(EINVAL));
|
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 (bytes_written, remote_id) = {
|
||||||
let name = self.get_socket(id)?.borrow().path.clone();
|
let name = self.get_socket(id)?.borrow().path.clone();
|
||||||
let (remote_id, remote_rc) = self.get_connected_peer(id)?;
|
let (remote_id, remote_rc) = self.get_connected_peer(id)?;
|
||||||
@@ -773,6 +777,19 @@ impl<'sock> UdsStreamScheme<'sock> {
|
|||||||
return Ok(0);
|
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);
|
connection.packets.push_back(packet);
|
||||||
(payload_len, remote_id)
|
(payload_len, remote_id)
|
||||||
};
|
};
|
||||||
@@ -1058,6 +1075,31 @@ impl<'sock> UdsStreamScheme<'sock> {
|
|||||||
buf: &[u8],
|
buf: &[u8],
|
||||||
ctx: &CallerCtx,
|
ctx: &CallerCtx,
|
||||||
) -> Result<usize> {
|
) -> Result<usize> {
|
||||||
|
{
|
||||||
|
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 receiver_rc = self.get_socket(receiver_id)?;
|
||||||
let mut receiver = receiver_rc.borrow_mut();
|
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<EventFlags> {
|
fn fevent(&mut self, id: usize, flags: EventFlags, _ctx: &CallerCtx) -> Result<EventFlags> {
|
||||||
let socket_rc = self.get_socket(id)?;
|
let socket_rc = self.get_socket(id)?.clone();
|
||||||
let socket = socket_rc.borrow();
|
let (mut result, peer_id, sndbuf) = {
|
||||||
Ok(socket.events() & flags)
|
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<()> {
|
fn fstat(&mut self, id: usize, stat: &mut Stat, _ctx: &CallerCtx) -> Result<()> {
|
||||||
|
|||||||
+26
-13
@@ -92,14 +92,29 @@ impl<'sock> LogScheme<'sock> {
|
|||||||
std::thread::spawn(move || {
|
std::thread::spawn(move || {
|
||||||
let mut handle_buf = vec![];
|
let mut handle_buf = vec![];
|
||||||
let mut buf = [0; 4096];
|
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 {
|
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 {
|
if n == 0 {
|
||||||
// FIXME currently possible as /scheme/log/kernel presents a snapshot of the log queue
|
std::thread::sleep(std::time::Duration::from_millis(50));
|
||||||
break;
|
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,
|
_flags: u32,
|
||||||
_ctx: &CallerCtx,
|
_ctx: &CallerCtx,
|
||||||
) -> Result<usize> {
|
) -> Result<usize> {
|
||||||
let _handle = self.handles.get(id)?;
|
match self.handles.get(id)? {
|
||||||
|
LogHandle::Log { .. } | LogHandle::AddSink | LogHandle::SchemeRoot => {
|
||||||
// TODO
|
Err(Error::new(EBADF))
|
||||||
|
}
|
||||||
Ok(0)
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn write(
|
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<usize> {
|
fn fcntl(&mut self, id: usize, _cmd: usize, _arg: usize, _ctx: &CallerCtx) -> Result<usize> {
|
||||||
let _handle = self.handles.get(id)?;
|
let _handle = self.handles.get(id)?;
|
||||||
|
|
||||||
Ok(0)
|
Err(Error::new(ENOSYS))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn fpath(&mut self, id: usize, buf: &mut [u8], _ctx: &CallerCtx) -> Result<usize> {
|
fn fpath(&mut self, id: usize, buf: &mut [u8], _ctx: &CallerCtx) -> Result<usize> {
|
||||||
@@ -258,8 +273,6 @@ impl<'sock> SchemeSync for LogScheme<'sock> {
|
|||||||
fn fsync(&mut self, id: usize, _ctx: &CallerCtx) -> Result<()> {
|
fn fsync(&mut self, id: usize, _ctx: &CallerCtx) -> Result<()> {
|
||||||
let _handle = self.handles.get(id)?;
|
let _handle = self.handles.get(id)?;
|
||||||
|
|
||||||
//TODO: flush remaining data?
|
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user