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:
Red Bear OS
2026-07-27 08:23:43 +09:00
parent 36dddf2300
commit 4ba511830f
3 changed files with 126 additions and 22 deletions
+29 -5
View File
@@ -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<Di
}
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 {
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")?;
+71 -4
View File
@@ -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<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 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> {
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<()> {
+26 -13
View File
@@ -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<usize> {
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<usize> {
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> {
@@ -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(())
}