cubelinux/cubelinux-kernel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2d4ed0fd9f1fe979546565927a8304e1d8d624ed
cubelinux-kernel/drivers/cube/cubelinux_store.rs
T
CUBELinux build 2d4ed0fd9f CUBELinux.0.5: a torn log entry is overwritten, not buried 
The write path's last unproven claim was the one everything else rests on:
replay is prefix-trusting. Building the gate for it found that the claim was
false as implemented — in *both* implementations, and in the same way.

The log walkers advanced their offset past an entry's frame and only then checked
the checksum. A corrupted entry therefore never moved the *replay* point (it was
discarded, so the store looked right) but it did move the *append* point. The
consequences, in order of how bad they are:

- the kernel appended **after** the tear, burying the corruption inside a log that
  then looked well-formed — the exact opposite of the documented rule;
- the userspace log truncated to a prefix that still contained the corrupt frame,
  so every later append started past it and the corruption stayed in the log
  forever;
- and because each append then read back the same wrong prefix, every append wrote
  to the same offset and overwrote its predecessor — four mutations in, one on disk.

The fix is one line of ordering in three places: compute the frame's end, check the
length and the checksum, and only then move the offset. An entry that did not
validate may not move the point that says where the log ends.

Gate (kernel/verify-torn-tail.sh), on a store with an overwrite, an insertion and a
deletion whose last entry has been torn by zeroing its tail in place:

  whole     : 272 bytes, 3 records   (the delete applied — the contrast)
  reference : 349 bytes, 4 records   (userspace, torn bytes: the delete discarded)
  kernel    : 349 bytes, 4 records
  appended  : 667 bytes, 8 records   (the kernel appended over the tear)
  expected  : 667 bytes, 8 records   (userspace: torn tail dropped, then the same four)

Building the gate also corrected the gate itself: tearing a log by *truncating the
device* is not a torn log, it is a smaller device — the capacity shrinks, writes
past the end vanish into the page cache with no error anywhere, and the test
measures an artifact. A torn write leaves the device the same size and corrupts
bytes in place.

All seven gates pass on 0.5.
2026-09-18 21:15:43 -04:00

1280 lines
47 KiB
Rust
Raw Blame History

// SPDX-License-Identifier: GPL-2.0
//! CUBELinux store reader — the kernel reading CUBE coordinates off a block device.
//!
//! The store is addressed by coordinate, and the pinned image's layout is:
//!
//! ```text
//! [MAGIC "CUBE" 4][version 1][curve tag]
//! records: [SpaceId 32][Key 24][value length u64 LE][value]
//! ```
//!
//! A store on a device is that image followed by a **write-ahead log** — the mutations since
//! the last checkpoint, in the same `op | crc32 | space | key | len | value` framing
//! `cube-store/src/wal.rs` writes. The log is a delta: the image is authoritative and
//! self-contained, and a reader that ignores the log loses only the mutations recorded after
//! the last checkpoint. `DESIGN-cubelinux-write-path.md` is why the write path looks like
//! this at all.
//!
//! This module reads both and reports the store they *describe* — the image with the log
//! applied: the record count, the total value bytes, and an FNV-1a digest over every
//! `(space, key, length, value)` in the order a checkpoint would write them. The digest exists so the kernel and userspace can be *compared*
//! rather than assumed to agree — `cube-image digest <image>` prints the same line in the
//! same field order, and the QEMU gate fails if they differ by a byte.
//!
//! # Why a reader, and only a reader
//!
//! The store's write authority has not moved yet. `PLAN-kernel-cubelinux.md` records the
//! decision (the kernel owns the store) and the hazard that makes the order matter: a
//! kernel that writes while a userspace daemon still holds the same image in memory loses
//! one of the two writers' work, silently. Until that is settled, a reader is the correct
//! amount of authority for the kernel to hold.
//!
//! # Interface
//!
//! `/dev/cubelinux` — reading it performs the read (through the kernel's own file layer)
//! and returns one line. Nothing happens at module init, so there is no ordering to get
//! wrong against the block driver that provides the device.
//!
//! ```text
//! digest curve=0 bytes=1141 records=11 value_bytes=431 fnv1a64=161113085b1573b2 errors=0
//! ```
use core::fmt::{self, Write};
use kernel::{
alloc::AllocError,
bindings, c_str,
device::Device,
fs::{File, Kiocb},
iov::{IovIterDest, IovIterSource},
miscdevice::{MiscDevice, MiscDeviceOptions, MiscDeviceRegistration},
prelude::*,
sync::aref::ARef,
};
module! {
type: CubeStoreModule,
name: "cubelinux_store",
authors: ["CUBE OS"],
description: "CUBELinux store reader (coordinates in the kernel)",
license: "GPL",
}
/// The layout, restated here because the kernel cannot depend on the userspace crates.
/// `crates/cube-store-raw` is the source of truth; the digest comparison is what keeps
/// this copy honest.
const MAGIC: &[u8; 4] = b"CUBE";
/// The original format: magic, version, curve, then records until zero padding.
const VERSION_V1: u8 = 1;
/// The current format: the same, plus the image's byte extent and record count.
const VERSION: u8 = 2;
const HEADER_LEN_V1: usize = 6;
const HEADER_LEN_V2: usize = 6 + 8 + 8;
const SPACE_ID_LEN: usize = 32;
const RAW_KEY_LEN: usize = 24;
const RECORD_FIXED: usize = SPACE_ID_LEN + RAW_KEY_LEN + 8;
/// The device to read. A module parameter is the obvious next step; for the boot gate one
/// fixed name is honest and has one less way to be wrong.
const STORE_DEVICE: &core::ffi::CStr = c_str!("/dev/vda");
/// Refuse to pull an unbounded device into memory. The gate's images are tiny; this is the
/// guard against a wrong device name turning a read into an allocation storm.
const MAX_BYTES: usize = 64 * 1024 * 1024;
/// The log's own magic, distinct from the image's so a reader that opens the wrong one
/// cannot mistake it for the other.
const WAL_MAGIC: [u8; 4] = *b"CUBW";
const WAL_VERSION: u8 = 1;
/// Log header: magic(4) + version(1) + curve tag(1).
const WAL_HEADER_LEN: usize = 6;
/// Log entry, before the value: op(1) + crc32(4) + space(32) + key(24) + len(4).
const ENTRY_FIXED: usize = 1 + 4 + 32 + 24 + 4;
/// The log begins at the first 4 KiB boundary at or after the image. Fixed by geometry so
/// no superblock is needed to find it, and stated by the v2 header's extent.
///
/// That is the *bare* layout: an image on a device with a log after it. A device that is
/// meant to be checkpointed has a control block instead (see `Control`), because folding a
/// log into the only copy of an image is not atomic and a crash in the middle of it loses the
/// store.
const LOG_ALIGN: usize = 4096;
/// Control block magic, distinct from the image's `CUBE` and the log's `CUBW`.
const CTL_MAGIC: [u8; 4] = *b"CUBS";
const CTL_VERSION: u8 = 1;
/// One control copy, and where the two live.
const CTL_COPY_LEN: usize = 2048;
const CTL_COPY_A: usize = 0;
const CTL_COPY_B: usize = 2048;
/// Where the two image slots begin.
const CTL_DATA_OFF: u64 = 4096;
/// Bytes of a control copy covered by its checksum.
const CTL_SUMMED: usize = 44;
/// The one mutation this device accepts so far. The coordinate interface's opcodes will be
/// the language's verbs; this is the first of them.
const OP_PUT: u8 = 1;
/// Fold the log into the image.
const OP_SYNC: u8 = 3;
/// FNV-1a offset basis.
const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
/// A line of output, built in place. No allocation: the read path formats into a fixed
/// buffer, so it cannot fail for want of memory while holding a file open.
struct Line {
buf: [u8; 256],
len: usize,
}
impl Line {
fn new() -> Self {
Line {
buf: [0; 256],
len: 0,
}
}
fn as_bytes(&self) -> &[u8] {
&self.buf[..self.len]
}
}
impl Write for Line {
fn write_str(&mut self, s: &str) -> fmt::Result {
let bytes = s.as_bytes();
if self.len + bytes.len() > self.buf.len() {
return Err(fmt::Error);
}
self.buf[self.len..self.len + bytes.len()].copy_from_slice(bytes);
self.len += bytes.len();
Ok(())
}
}
/// FNV-1a 64 over a run of bytes, continuing from `h`.
fn fnv1a64(bytes: &[u8], mut h: u64) -> u64 {
for b in bytes {
h ^= *b as u64;
h = h.wrapping_mul(0x0000_0100_0000_01b3);
}
h
}
/// Read the whole store image through the kernel's file layer.
///
/// `kernel_read` on the block device rather than a raw bio: it is the path this kernel
/// version exposes to Rust, it goes through the page cache the way any other read does, and
/// it needs no C helper. If the store ever has to be read before the VFS is up (a root
/// filesystem, say), that is the moment to reach for the block layer directly.
fn read_image(image: &mut KVVec<u8>) -> Result<()> {
// O_RDONLY is 0 in Linux; filp_open takes the raw flags word.
// SAFETY: STORE_DEVICE is a NUL-terminated C string literal, and filp_open either
// returns a valid `struct file *` or an error pointer, which is checked below.
let file = unsafe { bindings::filp_open(STORE_DEVICE.as_ptr().cast::<u8>(), 0, 0) };
let file = kernel::error::from_err_ptr(file)?;
if file.is_null() {
return Err(EINVAL);
}
let mut pos: bindings::loff_t = 0;
let mut chunk = KVVec::<u8>::with_capacity(4096, GFP_KERNEL)?;
chunk.extend_from_slice(&[0u8; 4096][..], GFP_KERNEL)?;
let mut result = Ok(());
loop {
// SAFETY: `file` is a live `struct file *` from filp_open; `chunk` is a 4096-byte
// kernel buffer we own; `pos` is a valid loff_t. kernel_read reads at most
// `chunk.len()` bytes into the buffer and does not retain either pointer.
let n = unsafe {
bindings::kernel_read(
file,
chunk.as_mut_ptr().cast::<core::ffi::c_void>(),
chunk.len(),
&mut pos,
)
};
if n < 0 {
result = Err(Error::from_errno(n as i32));
break;
}
if n == 0 {
break; // end of device
}
let n = n as usize;
if let Err(e) = image.extend_from_slice(&chunk[..n], GFP_KERNEL) {
result = Err(e.into());
break;
}
if image.len() >= MAX_BYTES {
break;
}
}
// SAFETY: `file` came from filp_open and has not been closed; the owner argument is
// only meaningful for locks that no one holds here.
unsafe { bindings::filp_close(file, core::ptr::null_mut()) };
// Less than the shortest header is not an image at all; `digest` reports the rest.
if result.is_ok() && image.len() < HEADER_LEN_V1 {
return Err(EINVAL);
}
result
}
/// Walk the records of a store image, digesting each one, exactly as
/// `cube-store-raw::iter_records` + `cube-image digest` do.
///
/// The two rules that are easy to get wrong, both from the userspace parser: a record whose
/// value runs past the buffer is a truncated record and an error, and an all-zero frame is
/// *end of records* only when every remaining byte is also zero — otherwise a real record
/// at the origin, followed by padding, would be read as an empty one.
/// What a header says, in both formats.
struct Header {
version: u8,
curve: u8,
image_bytes: Option<u64>,
record_count: Option<u64>,
}
/// Read the header, or say what is wrong with it.
fn parse_header(image: &[u8]) -> Result<Header, &'static str> {
if image.len() < 5 || &image[0..4] != MAGIC {
return Err("not-a-store");
}
let version = image[4];
let curve = image[5];
match version {
VERSION_V1 => Ok(Header {
version,
curve,
image_bytes: None,
record_count: None,
}),
VERSION => {
if image.len() < HEADER_LEN_V2 {
return Err("short-v2-header");
}
let mut word = [0u8; 8];
word.copy_from_slice(&image[6..14]);
let image_bytes = u64::from_le_bytes(word);
word.copy_from_slice(&image[14..22]);
let record_count = u64::from_le_bytes(word);
// A header that does not cover itself is corrupt, and guessing at the extent
// of an image means possibly reading somebody else's bytes.
if (image_bytes as usize) < HEADER_LEN_V2 {
return Err("bad-extent");
}
Ok(Header {
version,
curve,
image_bytes: Some(image_bytes),
record_count: Some(record_count),
})
}
_ => Err("unsupported-version"),
}
}
fn digest(image: &[u8]) -> Line {
let mut line = Line::new();
// Where the image is, and which log belongs to it: a store device says so in its control
// block, a bare image says so in its own header.
let layout = match resolve_layout(image) {
Ok(l) => l,
Err(what) => {
let _ = write!(line, "error={what}");
return line;
}
};
let live = &image[layout.image_off..];
let header = match parse_header(live) {
Ok(h) => h,
Err(what) => {
let _ = write!(line, "error={what}");
return line;
}
};
let end = match header.image_bytes {
Some(n) => core::cmp::min(n as usize, live.len()),
None => live.len(),
};
// With a v2 image, the store is the image *plus* whatever log follows it. The order the
// digest is taken in is then the order a checkpoint would write: sorted by coordinate,
// later writes winning. Without a log there is nothing to merge, and the image's own
// order is the answer — which is also what keeps v1 images readable and unchanged.
if header.image_bytes.is_some() {
let log = log_window(image, &layout);
match merged_digest(&live[..end], log, &header) {
Ok(line) => return line,
// A log that exists but cannot be read is worth saying out loud: the digest
// alone would look like a shorter store.
Err(what) if what != "no-log" => {
let mut line = Line::new();
let _ = write!(line, "error={what}");
return line;
}
Err(_) => {}
}
}
let mut h = FNV_OFFSET;
let mut count: u64 = 0;
let mut value_bytes: u64 = 0;
let mut errors: u64 = 0;
let mut off = layout.image_off
+ if header.version == VERSION_V1 {
HEADER_LEN_V1
} else {
HEADER_LEN_V2
};
let mut remaining = header.record_count;
let end = layout.image_off + end;
while off + RECORD_FIXED <= end {
if remaining == Some(0) {
break;
}
let frame = &image[off..off + RECORD_FIXED];
// Padding is not a record — but only if it is padding all the way down, and only
// in v1, where nothing else says where the records stop. A v2 image states its
// count, so a zero frame inside that count is a record like any other.
if remaining.is_none()
&& frame.iter().all(|b| *b == 0)
&& image[off..].iter().all(|b| *b == 0)
{
break;
}
let space = &image[off..off + SPACE_ID_LEN];
let key = &image[off + SPACE_ID_LEN..off + SPACE_ID_LEN + RAW_KEY_LEN];
let mut len_bytes = [0u8; 8];
len_bytes.copy_from_slice(&image[off + SPACE_ID_LEN + RAW_KEY_LEN..off + RECORD_FIXED]);
let value_len = u64::from_le_bytes(len_bytes) as usize;
let value_at = off + RECORD_FIXED;
// A promised record that is not there is a truncated image, and saying so beats
// returning a shorter list that looks complete.
if value_at + value_len > end {
errors += 1;
break;
}
let value = &image[value_at..value_at + value_len];
h = fnv1a64(space, h);
h = fnv1a64(key, h);
h = fnv1a64(&(value_len as u64).to_le_bytes(), h);
h = fnv1a64(value, h);
count += 1;
value_bytes += value_len as u64;
off = value_at + value_len;
if let Some(n) = remaining.as_mut() {
*n -= 1;
}
}
// The count was a promise; not meeting it is an error the caller must see.
if let Some(n) = remaining {
if n > 0 {
errors += 1;
}
}
// Field order matches cube-image's `digest` line so the two diff directly.
let _ = write!(
line,
"digest version={} curve={} bytes={} records={} value_bytes={} fnv1a64={:016x} errors={}",
header.version,
header.curve,
image.len(),
count,
value_bytes,
h,
errors
);
line
}
/// What a store device's control block says: where the two image slots and the log are,
/// which slot is live, and how much of the log is in use.
#[derive(Clone, Copy)]
struct Control {
curve: u8,
slot_bytes: u64,
log_capacity: u64,
log_used: u64,
generation: u64,
active_slot: u8,
}
impl Control {
/// Parse a control copy. `None` when it is not one, or its checksum does not hold — a
/// torn update in the middle of a checkpoint is exactly what that checksum is for.
fn decode(b: &[u8]) -> Option<Control> {
if b.len() < CTL_SUMMED + 4 || &b[0..4] != CTL_MAGIC {
return None;
}
let word = |at: usize| -> u64 {
let mut w = [0u8; 8];
w.copy_from_slice(&b[at..at + 8]);
u64::from_le_bytes(w)
};
let mut c = [0u8; 4];
c.copy_from_slice(&b[44..48]);
if u32::from_le_bytes(c) != crc32(&b[..CTL_SUMMED]) {
return None;
}
let control = Control {
curve: b[5],
slot_bytes: word(8),
log_capacity: word(16),
log_used: word(24),
generation: word(32),
active_slot: b[40],
};
if b[4] != CTL_VERSION || control.active_slot > 1 || control.slot_bytes == 0 {
return None;
}
Some(control)
}
/// Encode into a control copy, checksum included.
fn encode(&self, b: &mut [u8]) {
for byte in b.iter_mut() {
*byte = 0;
}
b[0..4].copy_from_slice(&CTL_MAGIC);
b[4] = CTL_VERSION;
b[5] = self.curve;
b[8..16].copy_from_slice(&self.slot_bytes.to_le_bytes());
b[16..24].copy_from_slice(&self.log_capacity.to_le_bytes());
b[24..32].copy_from_slice(&self.log_used.to_le_bytes());
b[32..40].copy_from_slice(&self.generation.to_le_bytes());
b[40] = self.active_slot;
let crc = crc32(&b[..CTL_SUMMED]);
b[44..48].copy_from_slice(&crc.to_le_bytes());
}
fn image_off(&self) -> u64 {
CTL_DATA_OFF + self.active_slot as u64 * self.slot_bytes
}
fn spare_off(&self) -> u64 {
CTL_DATA_OFF + (1 - self.active_slot) as u64 * self.slot_bytes
}
fn log_off(&self) -> u64 {
CTL_DATA_OFF + 2 * self.slot_bytes
}
}
/// The valid control block: the copy with the higher generation. A reader that finds one
/// copy torn uses the other, which is the entire reason there are two.
fn read_control(b: &[u8]) -> Option<Control> {
if b.len() < CTL_COPY_B + CTL_COPY_LEN {
return None;
}
let a = Control::decode(&b[CTL_COPY_A..CTL_COPY_A + CTL_COPY_LEN]);
let b_copy = Control::decode(&b[CTL_COPY_B..CTL_COPY_B + CTL_COPY_LEN]);
match (a, b_copy) {
(Some(x), Some(y)) => Some(if x.generation >= y.generation { x } else { y }),
(Some(x), None) => Some(x),
(None, Some(y)) => Some(y),
(None, None) => None,
}
}
/// Where the image is and what log belongs to it.
struct Layout {
/// Byte offset of the live image.
image_off: usize,
/// Byte offset of its log.
log_off: usize,
/// Bytes of log in use.
log_used: usize,
/// The control block, when this is a store device rather than a bare image.
control: Option<Control>,
}
/// Work out the layout: a store device when the control block is there, a bare image when it
/// is not. Both are readable; only the store layout can be checkpointed atomically, which is
/// why `sync` refuses on the other.
fn resolve_layout(b: &[u8]) -> Result<Layout, &'static str> {
if b.len() >= CTL_COPY_B + CTL_COPY_LEN && &b[0..4] == CTL_MAGIC {
let control = read_control(b).ok_or("control-block-corrupt")?;
let log_off = control.log_off() as usize;
let used = control.log_used as usize;
if log_off + used > b.len() || used > control.log_capacity as usize {
return Err("log-outside-device");
}
return Ok(Layout {
image_off: control.image_off() as usize,
log_off,
log_used: used,
control: Some(control),
});
}
// A bare image. A v1 image has no extent and therefore no log — it is the whole device
// as far as this module is concerned, and it is read the way it always was.
let header = parse_header(b)?;
let log_off = match header.image_bytes {
Some(n) => log_offset(n),
None => b.len(),
};
Ok(Layout {
image_off: 0,
log_off,
// A bare device records no log length, so the log is bounded by the buffer and by
// the entries that parse: `apply_log` stops at the first one that does not.
log_used: b.len().saturating_sub(log_off),
control: None,
})
}
/// One record on its way into the merged store. Fixed size, so a `KVVec` of these sorts
/// in place; values live in a pool beside them and are referenced by offset.
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
struct Entry {
space: [u8; 32],
key: [u8; 24],
/// Order in which this entry arrived. Breaks ties between the same coordinate, so the
/// later write wins after sorting.
seq: u32,
value_off: u32,
value_len: u32,
deleted: bool,
}
/// The store as a sorted set of records, which is what a checkpoint writes and what the
/// digest is taken over.
struct Merged {
pool: KVVec<u8>,
entries: KVVec<Entry>,
seq: u32,
/// Log entries that were seen but not applied, because they were torn or malformed.
dropped: u64,
}
impl Merged {
fn new() -> Result<Self, AllocError> {
Ok(Merged {
pool: KVVec::new(),
entries: KVVec::new(),
seq: 0,
dropped: 0,
})
}
/// Add a record. `deleted` marks a removal, which is kept in the list so it can
/// override an older value for the same coordinate.
fn add(&mut self, space: &[u8], key: &[u8], value: &[u8], deleted: bool) -> Result<(), AllocError> {
let mut sp = [0u8; 32];
sp.copy_from_slice(&space[..32]);
let mut k = [0u8; 24];
k.copy_from_slice(&key[..24]);
let value_off = self.pool.len() as u32;
self.pool.extend_from_slice(value, GFP_KERNEL)?;
let seq = self.seq;
self.seq += 1;
self.entries.push(
Entry {
space: sp,
key: k,
seq,
value_off,
value_len: value.len() as u32,
deleted,
},
GFP_KERNEL,
)?;
Ok(())
}
/// Put the entries in the order a checkpoint would write them: by space, then by key,
/// with the later write of a coordinate last — so a group's final entry decides it.
fn sort_entries(&mut self) {
self.entries.as_mut_slice().sort_unstable();
}
fn value(&self, e: &Entry) -> &[u8] {
let off = e.value_off as usize;
&self.pool.as_slice()[off..off + e.value_len as usize]
}
}
/// Read the log that follows the image, if there is one, and apply its entries.
///
/// The log is a delta on the image, so its entries override image records for the same
/// coordinate. Recovery is prefix-trusting, exactly as userspace does it: entries are
/// replayed from the start and the first one that is short, mis-framed or fails its
/// checksum ends the log. A fault in the middle cannot be told from a torn tail without a
/// second copy, so the log stops there and reports what it dropped instead of guessing.
fn apply_log(log: &[u8], merged: &mut Merged) -> Result<Option<u64>, &'static str> {
if log.len() < WAL_HEADER_LEN || &log[0..4] != WAL_MAGIC {
return Ok(None);
}
if log[4] != WAL_VERSION {
return Err("unsupported-log-version");
}
let mut applied: u64 = 0;
let mut off = WAL_HEADER_LEN;
while off + ENTRY_FIXED <= log.len() {
let start = off;
let op = log[off];
if op != 1 && op != 2 {
break;
}
let mut word = [0u8; 4];
word.copy_from_slice(&log[off + 1..off + 5]);
let crc = u32::from_le_bytes(word);
let space = &log[off + 5..off + 37];
let key = &log[off + 37..off + 61];
word.copy_from_slice(&log[off + 61..off + 65]);
let len = u32::from_le_bytes(word) as usize;
let frame_end = start + ENTRY_FIXED + len;
if frame_end > log.len() {
break;
}
// The checksum covers space, key, length and value, so a corrupted entry is
// stopped at rather than applied.
if crc32(&log[start + 5..frame_end]) != crc {
break;
}
let value = &log[start + ENTRY_FIXED..frame_end];
off = frame_end;
merged
.add(space, key, value, op == 2)
.map_err(|_| "out-of-memory")?;
applied += 1;
}
merged.dropped = (log.len() - off) as u64;
Ok(Some(applied))
}
/// CRC-32 (IEEE 802.3), bitwise — the same polynomial and the same coverage as
/// `cube-store/src/wal.rs`, because a log written on either side must validate on both.
fn crc32(bytes: &[u8]) -> u32 {
let mut crc = 0xFFFF_FFFFu32;
for byte in bytes {
crc ^= *byte as u32;
for _ in 0..8 {
let mask = (crc & 1).wrapping_neg();
crc = (crc >> 1) ^ (0xEDB8_8320 & mask);
}
}
!crc
}
/// The log window that belongs to the live image.
fn log_window<'a>(device: &'a [u8], layout: &Layout) -> &'a [u8] {
let end = core::cmp::min(
layout.log_off + WAL_HEADER_LEN + layout.log_used,
device.len(),
);
if layout.log_off <= end {
&device[layout.log_off..end]
} else {
&[]
}
}
/// The store the image and its log together describe: the image's records, then the log's
/// entries applied over them. This is what a checkpoint writes and what the digest is taken
/// over, so both go through here rather than each walking the bytes its own way.
fn build_merged(image: &[u8], log: &[u8], header: &Header) -> Result<(Merged, u64), &'static str> {
let mut merged = Merged::new().map_err(|_| "out-of-memory")?;
let extent = header.image_bytes.ok_or("v1-has-no-extent")? as usize;
let extent = core::cmp::min(extent, image.len());
let mut off = if header.version == VERSION_V1 {
HEADER_LEN_V1
} else {
HEADER_LEN_V2
};
let mut seen: u64 = 0;
while off + RECORD_FIXED <= extent {
if let Some(c) = header.record_count {
if seen >= c {
break;
}
}
let space = &image[off..off + SPACE_ID_LEN];
let key = &image[off + SPACE_ID_LEN..off + SPACE_ID_LEN + RAW_KEY_LEN];
let mut word = [0u8; 8];
word.copy_from_slice(&image[off + SPACE_ID_LEN + RAW_KEY_LEN..off + RECORD_FIXED]);
let value_len = u64::from_le_bytes(word) as usize;
let value_at = off + RECORD_FIXED;
if value_at + value_len > extent {
return Err("truncated-image");
}
merged
.add(space, key, &image[value_at..value_at + value_len], false)
.map_err(|_| "out-of-memory")?;
seen += 1;
off = value_at + value_len;
}
let applied = match apply_log(log, &mut merged)? {
Some(n) => n,
None => 0,
};
Ok((merged, applied))
}
/// Digest the store the image and its log together describe.
///
/// `image` is the live image's bytes alone; `log` is exactly the window of log that belongs
/// to it. Returns `Err("no-log")` when there is none, so the caller can fall back to reading
/// the image by itself.
fn merged_digest(image: &[u8], log: &[u8], header: &Header) -> Result<Line, &'static str> {
if log.len() < WAL_HEADER_LEN || &log[0..4] != WAL_MAGIC {
return Err("no-log");
}
let (mut merged, applied) = build_merged(image, log, header)?;
merged.sort_entries();
let entries = merged.entries.as_slice();
let mut h = FNV_OFFSET;
let mut count: u64 = 0;
let mut value_bytes: u64 = 0;
let mut i = 0;
while i < entries.len() {
let head = &entries[i];
let mut last = i;
while last + 1 < entries.len()
&& entries[last + 1].space == head.space
&& entries[last + 1].key == head.key
{
last += 1;
}
let winner = &entries[last];
i = last + 1;
if winner.deleted {
continue;
}
let value = merged.value(winner);
h = fnv1a64(&winner.space, h);
h = fnv1a64(&winner.key, h);
h = fnv1a64(&(value.len() as u64).to_le_bytes(), h);
h = fnv1a64(value, h);
count += 1;
value_bytes += value.len() as u64;
}
// `bytes` is the size a checkpoint of this store would produce — the same number
// userspace reports for the image it writes from the same records.
let folded = HEADER_LEN_V2 as u64 + count * RECORD_FIXED as u64 + value_bytes;
let mut line = Line::new();
// The digest line is field-for-field what `cube-image digest` prints, so the two are
// compared by diff. What the log contributed goes on its own line: a padded log region
// makes a byte count meaningless, an entry count does not.
let _ = write!(
line,
"digest version={} curve={} bytes={} records={} value_bytes={} fnv1a64={:016x} errors=0\n",
header.version, header.curve, folded, count, value_bytes, h
);
let _ = write!(line, "log entries={} applied over the image\n", applied);
Ok(line)
}
/// Where a bare image's log starts, given the image's declared extent.
fn log_offset(image_bytes: u64) -> usize {
let n = image_bytes as usize;
(n + LOG_ALIGN - 1) & !(LOG_ALIGN - 1)
}
/// Length of a bare log's valid prefix: the header plus every entry that parses and passes
/// its checksum. This is the append position, and the reason a torn tail is overwritten
/// rather than appended to.
fn log_valid_len(log: &[u8]) -> Result<usize, &'static str> {
if log.len() < WAL_HEADER_LEN {
return Err("short-log");
}
if &log[0..4] != WAL_MAGIC {
return Err("no-log");
}
if log[4] != WAL_VERSION {
return Err("unsupported-log-version");
}
let mut off = WAL_HEADER_LEN;
while off + ENTRY_FIXED <= log.len() {
let start = off;
let op = log[off];
if op != 1 && op != 2 {
break;
}
let mut word = [0u8; 4];
word.copy_from_slice(&log[off + 1..off + 5]);
let crc = u32::from_le_bytes(word);
word.copy_from_slice(&log[off + 61..off + 65]);
let len = u32::from_le_bytes(word) as usize;
// Only a validated entry moves the append point. Advancing first and checking after
// counts a torn entry as part of the prefix, so the next append lands *after* the
// corruption and buries it — the opposite of the rule that a torn tail is overwritten.
let frame_end = start + ENTRY_FIXED + len;
if frame_end > log.len() {
break;
}
if crc32(&log[start + 5..frame_end]) != crc {
break;
}
off = frame_end;
}
Ok(off)
}
/// Morton-encode a point into the store's 24-byte key — the same interleaving
/// `cube-core`'s `Curve for Morton` does, because a key written here has to be the key a
/// userspace reader decodes. Bit `i` of each axis lands at bit `3i` of the 192-bit key,
/// counting from the least significant bit, which lives in the *last* byte.
fn morton_encode(x: u64, y: u64, z: u64) -> [u8; 24] {
let mut k = [0u8; 24];
let put = |n: usize, v: bool, k: &mut [u8; 24]| {
if !v {
return;
}
let byte = 23 - (n / 8);
k[byte] |= 1 << (n % 8);
};
for i in 0..64 {
put(3 * i, (x >> i) & 1 != 0, &mut k);
put(3 * i + 1, (y >> i) & 1 != 0, &mut k);
put(3 * i + 2, (z >> i) & 1 != 0, &mut k);
}
k
}
/// A mutation to append: the byte-plane write the store contract calls `put`.
struct Mutation {
space: [u8; 32],
key: [u8; 24],
value: KVVec<u8>,
}
/// Build a log entry: `op | crc32 | space | key | len | value`.
fn encode_entry(space: &[u8; 32], key: &[u8; 24], value: &[u8]) -> Result<KVVec<u8>, AllocError> {
let total = ENTRY_FIXED + value.len();
let mut entry = KVVec::<u8>::with_capacity(total, GFP_KERNEL)?;
entry.extend_from_slice(&[1u8], GFP_KERNEL)?;
entry.extend_from_slice(&[0u8; 4][..], GFP_KERNEL)?;
entry.extend_from_slice(space, GFP_KERNEL)?;
entry.extend_from_slice(key, GFP_KERNEL)?;
entry.extend_from_slice(&(value.len() as u32).to_le_bytes(), GFP_KERNEL)?;
entry.extend_from_slice(value, GFP_KERNEL)?;
// The checksum covers everything after the crc field.
let crc = crc32(&entry.as_slice()[5..]);
entry.as_mut_slice()[1..5].copy_from_slice(&crc.to_le_bytes());
Ok(entry)
}
/// Write `bytes` at `off` and flush, or say why not. The one place the module writes.
fn write_and_sync(file: *mut bindings::file, off: u64, bytes: &[u8]) -> Result<(), Error> {
let mut at: bindings::loff_t = off as bindings::loff_t;
// SAFETY: the caller passes a live `struct file *`; `bytes` outlives the call; `at` is a
// valid loff_t.
let wrote = unsafe {
bindings::kernel_write(
file,
bytes.as_ptr().cast::<core::ffi::c_void>(),
bytes.len(),
&mut at,
)
};
if wrote < 0 {
return Err(Error::from_errno(wrote as i32));
}
if wrote as usize != bytes.len() {
return Err(EIO);
}
// Durability before acknowledgement: a short write that was never flushed is a write that
// did not happen.
// SAFETY: the caller passes a live, writable file; datasync is 0 (a full sync).
let rc = unsafe { bindings::vfs_fsync(file, 0) };
if rc < 0 {
return Err(Error::from_errno(rc));
}
Ok(())
}
/// The control copy an update should write: the older of the two, so the current one stays
/// readable throughout.
fn older_copy(generation: u64) -> usize {
if generation % 2 == 0 {
CTL_COPY_A
} else {
CTL_COPY_B
}
}
/// Append a mutation to the log and make it durable, returning the updated control block
/// when the layout has one.
///
/// The order is the durability contract: the entry is written and flushed to the device
/// *before* the caller is told it happened. A mutation this function has returned `Ok` for
/// survives a power cut; one it has not may or may not, which is exactly the boundary an
/// acknowledgement is supposed to mark.
///
/// On a store device the entry goes after the log bytes already in use, and the control
/// block's count is raised afterwards — entry first, count second, so a crash between them
/// loses an *unacknowledged* mutation rather than counting one that is not there. A count
/// that ran ahead would make replay read past valid data; a count that lags only forgets.
fn append(image: &[u8], layout: &Layout, m: &Mutation) -> Result<Option<Control>, Error> {
let entry = encode_entry(&m.space, &m.key, m.value.as_slice())?;
// Where it goes: after the bytes in use on a store device, after the valid prefix on a
// bare image, where nothing records the length.
let region_at = core::cmp::min(layout.log_off, image.len());
let region = &image[region_at..];
let (used, capacity) = match layout.control {
// `log_used` counts entry bytes *after* the log's header, so the header is not
// something an append has to account for in the count.
Some(c) => (c.log_used as usize, c.log_capacity as usize),
None => {
// A region with no header is a log that has never been written; it is empty, not
// broken, and the header goes down before the first entry.
let used = match log_valid_len(region) {
Ok(valid) => valid - WAL_HEADER_LEN,
Err("no-log") | Err("short-log") => 0,
Err(what) => {
pr_err!("cubelinux: log region unreadable: {}\n", what);
return Err(EINVAL);
}
};
(used, region.len())
}
};
if WAL_HEADER_LEN + used + entry.len() > capacity {
pr_err!(
"cubelinux: the log is full ({} of {} bytes); checkpoint before appending\n",
used,
capacity
);
return Err(ENOSPC);
}
// SAFETY: STORE_DEVICE is a NUL-terminated literal; filp_open returns a valid file or an
// error pointer, which is checked. O_RDWR is 2.
let file = unsafe { bindings::filp_open(STORE_DEVICE.as_ptr().cast::<u8>(), 2, 0) };
let file = kernel::error::from_err_ptr(file)?;
if file.is_null() {
return Err(EINVAL);
}
let mut result: Result<(), Error> = Ok(());
// A log region that has never been written is zeros, not a log: give it a header, the
// same thing the userspace log does when its file does not exist.
// A log region with no header is not a log: give it one, whether this is a bare image or
// a store device, so a reader always finds the framing it expects.
if region.len() < WAL_HEADER_LEN || &region[0..4] != WAL_MAGIC {
let mut hdr = [0u8; WAL_HEADER_LEN];
hdr[0..4].copy_from_slice(&WAL_MAGIC);
hdr[4] = WAL_VERSION;
hdr[5] = 0; // morton
result = write_and_sync(file, layout.log_off as u64, &hdr);
}
if result.is_ok() {
result = write_and_sync(
file,
(region_at + WAL_HEADER_LEN + used) as u64,
entry.as_slice(),
);
}
// The count second, so replay can never be told about an entry that is not there.
let mut updated = None;
if result.is_ok() {
if let Some(c) = layout.control {
let next = Control {
log_used: c.log_used + entry.len() as u64,
generation: c.generation + 1,
..c
};
let mut buf = [0u8; CTL_COPY_LEN];
next.encode(&mut buf);
result = write_and_sync(file, older_copy(c.generation) as u64, &buf);
if result.is_ok() {
updated = Some(next);
}
}
}
// SAFETY: `file` came from filp_open and has not been closed.
unsafe { bindings::filp_close(file, core::ptr::null_mut()) };
result.map(|_| updated)
}
/// Serialize the merged store into the pinned v2 image.
///
/// The winners are collected first because the header states the image's extent and its
/// record count, and those are only known once the coalescing is done.
fn serialize_image(merged: &mut Merged, curve: u8) -> Result<KVVec<u8>, AllocError> {
merged.sort_entries();
let entries = merged.entries.as_slice();
let mut winners = KVVec::<u32>::new();
let mut values_len: u64 = 0;
let mut i = 0;
while i < entries.len() {
let head = &entries[i];
let mut last = i;
while last + 1 < entries.len()
&& entries[last + 1].space == head.space
&& entries[last + 1].key == head.key
{
last += 1;
}
if !entries[last].deleted {
winners.push(last as u32, GFP_KERNEL)?;
values_len += entries[last].value_len as u64;
}
i = last + 1;
}
let image_bytes = HEADER_LEN_V2 as u64 + winners.len() as u64 * RECORD_FIXED as u64 + values_len;
let mut out = KVVec::<u8>::with_capacity(image_bytes as usize, GFP_KERNEL)?;
out.extend_from_slice(MAGIC, GFP_KERNEL)?;
out.extend_from_slice(&[VERSION, curve], GFP_KERNEL)?;
out.extend_from_slice(&image_bytes.to_le_bytes(), GFP_KERNEL)?;
out.extend_from_slice(&(winners.len() as u64).to_le_bytes(), GFP_KERNEL)?;
for w in winners.as_slice() {
let e = &merged.entries.as_slice()[*w as usize];
let value = merged.value(e);
out.extend_from_slice(&e.space, GFP_KERNEL)?;
out.extend_from_slice(&e.key, GFP_KERNEL)?;
out.extend_from_slice(&(value.len() as u64).to_le_bytes(), GFP_KERNEL)?;
out.extend_from_slice(value, GFP_KERNEL)?;
}
Ok(out)
}
/// Fold the log into the image: write the merged store into the spare slot, then make it
/// live.
///
/// The order is what makes this atomic. The new image goes to the slot nothing is reading and
/// is flushed; only then does the control block change, and that change is one small write to
/// the control copy that is *not* current. At every instant there is either the old image with
/// a log that still describes the mutations since it, or the new image with an empty log. A
/// crash in the middle of the copy leaves the previous store intact, which is the entire
/// reason for two slots.
fn checkpoint(ctl: &Control, merged: &mut Merged) -> Result<u64, Error> {
let new_image = serialize_image(merged, ctl.curve)?;
if new_image.len() as u64 > ctl.slot_bytes {
pr_err!(
"cubelinux: checkpoint needs {} bytes but a slot holds {}\n",
new_image.len(),
ctl.slot_bytes
);
return Err(ENOSPC);
}
// SAFETY: as in `append`.
let file = unsafe { bindings::filp_open(STORE_DEVICE.as_ptr().cast::<u8>(), 2, 0) };
let file = kernel::error::from_err_ptr(file)?;
if file.is_null() {
return Err(EINVAL);
}
let mut result = write_and_sync(file, ctl.spare_off(), new_image.as_slice());
if result.is_ok() {
let next = Control {
log_used: 0,
generation: ctl.generation + 1,
active_slot: 1 - ctl.active_slot,
..*ctl
};
let mut buf = [0u8; CTL_COPY_LEN];
next.encode(&mut buf);
result = write_and_sync(file, older_copy(ctl.generation) as u64, &buf);
}
// SAFETY: `file` came from filp_open and has not been closed.
unsafe { bindings::filp_close(file, core::ptr::null_mut()) };
result.map(|_| new_image.len() as u64)
}
/// The module's registration; holds the misc device for as long as the module lives.
#[pin_data]
struct CubeStoreModule {
#[pin]
_miscdev: MiscDeviceRegistration<CubeStore>,
}
impl kernel::InPlaceModule for CubeStoreModule {
fn init(_module: &'static ThisModule) -> impl PinInit<Self, Error> {
pr_info!("cubelinux: store reader registered at /dev/cubelinux\n");
try_pin_init!(Self {
_miscdev <- MiscDeviceRegistration::register(MiscDeviceOptions {
name: c_str!("cubelinux"),
}),
})
}
}
#[pin_data]
struct CubeStore {
dev: ARef<Device>,
}
#[vtable]
impl MiscDevice for CubeStore {
type Ptr = Pin<KBox<Self>>;
fn open(_file: &File, misc: &MiscDeviceRegistration<Self>) -> Result<Pin<KBox<Self>>> {
KBox::try_pin_init(
try_pin_init! {
CubeStore { dev: ARef::from(misc.device()) }
},
GFP_KERNEL,
)
}
/// Write one operation, framed as the argument block the coordinate interface will pass.
///
/// ```text
/// put op=1 | space(32) | x(8) | y(8) | z(8) | len(4) | value[len]
/// sync op=3
/// ```
///
/// `put` is the byte plane — bytes at a coordinate, with no header written beside them —
/// which is what the store contract defines and what the differential gate compares
/// against userspace's `cell put`. The header tier sits above this.
///
/// `sync` folds the log into the image. It is a real operation rather than a test hook: a
/// caller that wants the log reclaimed — a shutdown, a snapshot, a handover — is entitled
/// to ask.
fn write_iter(kiocb: Kiocb<'_, Self::Ptr>, iov: &mut IovIterSource<'_>) -> Result<usize> {
let mut buf = KVVec::<u8>::new();
let len = iov.copy_from_iter_vec(&mut buf, GFP_KERNEL)?;
let bytes = buf.as_slice();
if bytes.is_empty() {
return Err(EINVAL);
}
let me = kiocb.file();
// Both operations need the device: its layout says where the image and log are.
let mut device = KVVec::<u8>::new();
read_image(&mut device)?;
let layout = match resolve_layout(&device) {
Ok(l) => l,
Err(what) => {
dev_err!(me.dev, "cubelinux: cannot resolve the store layout: {}\n", what);
return Err(EINVAL);
}
};
match bytes[0] {
OP_PUT => {
if bytes.len() < 1 + 32 + 24 + 4 {
return Err(EINVAL);
}
let mut space = [0u8; 32];
space.copy_from_slice(&bytes[1..33]);
let axis = |at: usize| -> u64 {
let mut w = [0u8; 8];
w.copy_from_slice(&bytes[at..at + 8]);
u64::from_le_bytes(w)
};
let (x, y, z) = (axis(33), axis(41), axis(49));
let mut w = [0u8; 4];
w.copy_from_slice(&bytes[57..61]);
let value_len = u32::from_le_bytes(w) as usize;
if bytes.len() < 61 + value_len {
return Err(EINVAL);
}
let mut value = KVVec::<u8>::new();
value.extend_from_slice(&bytes[61..61 + value_len], GFP_KERNEL)?;
let mutation = Mutation {
space,
key: morton_encode(x, y, z),
value,
};
match append(&device, &layout, &mutation) {
Ok(_) => {
dev_info!(
me.dev,
"cubelinux: appended {} bytes for a {} byte value\n",
ENTRY_FIXED + value_len,
value_len
);
Ok(len)
}
Err(e) => {
dev_err!(me.dev, "cubelinux: append failed: {:?}\n", e);
Err(e)
}
}
}
OP_SYNC => {
let ctl = match layout.control {
Some(c) => c,
None => {
// A bare image has one slot, so folding into it is not atomic and a
// crash in the middle would lose the store. Refusing is the honest
// answer; the fix is to format the device as a store.
dev_err!(
me.dev,
"cubelinux: cannot checkpoint a bare image (no control block)\n"
);
return Err(EINVAL);
}
};
// The store the image and its log describe, ready to be written out.
let live = &device[layout.image_off..];
let header = match parse_header(live) {
Ok(h) => h,
Err(what) => {
dev_err!(me.dev, "cubelinux: {}\n", what);
return Err(EINVAL);
}
};
let window = log_window(&device, &layout);
let mut merged = match build_merged(live, window, &header) {
Ok(m) => m,
Err(what) => {
dev_err!(me.dev, "cubelinux: cannot build the store: {}\n", what);
return Err(EINVAL);
}
};
match checkpoint(&ctl, &mut merged.0) {
Ok(bytes) => {
dev_info!(me.dev, "cubelinux: checkpointed {} bytes\n", bytes);
Ok(len)
}
Err(e) => {
dev_err!(me.dev, "cubelinux: checkpoint failed: {:?}\n", e);
Err(e)
}
}
}
other => {
dev_err!(me.dev, "cubelinux: unknown operation {}\n", other);
Err(EINVAL)
}
}
}
fn read_iter(mut kiocb: Kiocb<'_, Self::Ptr>, iov: &mut IovIterDest<'_>) -> Result<usize> {
// The work happens here, on demand, rather than at init: by the time anything can
// open this device, the block driver that provides the store is certainly up.
let mut image = KVVec::<u8>::new();
let line = match read_image(&mut image) {
Ok(()) => digest(&image),
Err(e) => {
let mut line = Line::new();
let _ = write!(line, "error={:?}", e);
line
}
};
pr_info!("cubelinux: {}\n", core::str::from_utf8(line.as_bytes()).unwrap_or("<bad utf8>"));
let me = kiocb.file();
dev_info!(me.dev, "cubelinux: store read complete\n");
iov.simple_read_from_buffer(kiocb.ki_pos_mut(), line.as_bytes())
}
}
Reference in New Issue View Git Blame Copy Permalink