#![cfg(target_os = "linux")]
use std::fs::File;
use std::os::unix::fs::FileExt;
use std::os::unix::io::AsRawFd;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicU64, Ordering};
use anyhow::{Result, anyhow, bail};
use io_uring::{IoUring, opcode, squeue, types};
use crate::archive_write::{
ArchiveWrite, Extent, encode_journal_row, journal_path, open_blobs,
};
const JOURNAL_STAGING: usize = 4096;
const MAX_WRITE_CHUNK: usize = 1 << 30;
const RING_ENTRIES: u32 = 64;
pub(crate) fn page_size() -> usize {
let n = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
if n > 0 { n as usize } else { 4096 }
}
pub(crate) fn memlock_limit_bytes() -> Option<u64> {
let mut lim = libc::rlimit {
rlim_cur: 0,
rlim_max: 0,
};
if unsafe { libc::getrlimit(libc::RLIMIT_MEMLOCK, &mut lim) } != 0 {
return None;
}
if lim.rlim_cur == libc::RLIM_INFINITY {
None
} else {
Some(lim.rlim_cur as u64)
}
}
pub fn probe() -> std::io::Result<()> {
let ring = IoUring::new(2)?;
let mut probe = io_uring::register::Probe::new();
ring.submitter().register_probe(&mut probe)?;
for (code, what) in [
(opcode::Write::CODE, "IORING_OP_WRITE"),
(opcode::WriteFixed::CODE, "IORING_OP_WRITE_FIXED"),
(opcode::Fsync::CODE, "IORING_OP_FSYNC"),
] {
if !probe.is_supported(code) {
return Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
format!("this kernel's io_uring has no {what}"),
));
}
}
Ok(())
}
struct Staging {
ptr: *mut u8,
len: usize,
}
impl Staging {
fn map(len: usize) -> Result<Self> {
let page = page_size();
let len = len.next_multiple_of(page).max(page);
let ptr = unsafe {
libc::mmap(
std::ptr::null_mut(),
len,
libc::PROT_READ | libc::PROT_WRITE,
libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
-1,
0,
)
};
if ptr == libc::MAP_FAILED {
return Err(anyhow!(
"uring: mmap {len} B for the registered staging buffer: {}",
std::io::Error::last_os_error()
));
}
let advised = unsafe { libc::madvise(ptr, len, libc::MADV_NOHUGEPAGE) };
if advised != 0 {
let e = std::io::Error::last_os_error();
let benign = matches!(
e.raw_os_error(),
Some(libc::EINVAL) | Some(libc::ENOSYS)
);
if !benign {
unsafe { libc::munmap(ptr, len) };
return Err(anyhow!(
"uring: madvise(MADV_NOHUGEPAGE) on the staging buffer: {e}. Without it a \
one-page registration can be charged as the whole 2 MiB huge page it sits \
in, which caps a process at four io_uring stores on a stock 8 MiB \
RLIMIT_MEMLOCK."
));
}
}
Ok(Self {
ptr: ptr as *mut u8,
len,
})
}
fn len(&self) -> usize {
self.len
}
fn as_ptr(&self) -> *const u8 {
self.ptr
}
fn as_mut_slice(&mut self) -> &mut [u8] {
unsafe { std::slice::from_raw_parts_mut(self.ptr, self.len) }
}
}
impl Drop for Staging {
fn drop(&mut self) {
unsafe { libc::munmap(self.ptr as *mut libc::c_void, self.len) };
}
}
struct Ring {
ring: IoUring,
staging: Staging,
journal_cursor: u64,
}
unsafe impl Send for Ring {}
pub struct UringWriter {
blobs: Arc<File>,
journal: File,
journal_file_path: PathBuf,
cursor: AtomicU64,
ring: Mutex<Ring>,
}
impl UringWriter {
pub fn create(archive: &Path) -> Result<Self> {
let (blobs, end) = open_blobs(archive)?;
let jpath = journal_path(archive);
let ring = IoUring::new(RING_ENTRIES).map_err(|e| {
anyhow!(
"uring: io_uring_setup failed ({e}). This kernel cannot run the io_uring arm \
(check /proc/sys/kernel/io_uring_disabled, seccomp, container policy). \
No fallback is substituted — a pwrite number reported as an io_uring number \
would be a lie."
)
})?;
let mut probe = io_uring::register::Probe::new();
ring.submitter()
.register_probe(&mut probe)
.map_err(|e| anyhow!("uring: register_probe: {e}"))?;
for (code, what) in [
(opcode::Write::CODE, "IORING_OP_WRITE"),
(opcode::WriteFixed::CODE, "IORING_OP_WRITE_FIXED"),
(opcode::Fsync::CODE, "IORING_OP_FSYNC"),
] {
if !probe.is_supported(code) {
bail!(
"uring: this kernel does not support {what}; the write→fsync→journal→fsync \
chain cannot be built. Not degrading to pwrite."
);
}
}
let staging = Staging::map(JOURNAL_STAGING)?;
unsafe {
let iov = libc::iovec {
iov_base: staging.as_ptr() as *mut libc::c_void,
iov_len: staging.len(),
};
ring.submitter()
.register_buffers(std::slice::from_ref(&iov))
.map_err(|e| {
anyhow!(
"uring: register_buffers ({} B, {} page(s)): {e}. A registration is \
pinned memory charged against RLIMIT_MEMLOCK, and the kernel keeps that \
count on the user_struct — it is per-UID and shared with every other \
process this user is running, not per-process. This one is currently \
{}. Every io_uring store holds one registration for as long as it is \
open, so N stores pin N pages; `fast` and `safe` register nothing and \
have no such ceiling. The page(s) named here are what was ASKED for; \
the kernel charges by compound_head, so a staging buffer that ended up \
inside a transparent huge page would be charged the whole 2 MiB — see \
`Staging`, which takes its own MADV_NOHUGEPAGE mapping so that cannot \
happen.",
JOURNAL_STAGING,
JOURNAL_STAGING.div_ceil(page_size()),
memlock_limit_bytes()
.map(|b| format!("{b} B"))
.unwrap_or_else(|| "unreadable".into()),
)
})?;
}
let journal = std::fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&jpath)
.map_err(|e| anyhow!("uring: open journal {}: {e}", jpath.display()))?;
let head = crate::archive_write::journal_head(&journal, &jpath)?;
let journal_cursor = if head.rewrite {
journal
.set_len(0)
.map_err(|e| anyhow!("uring: resetting the interrupted journal {}: {e}", jpath.display()))?;
journal.write_all_at(&head.schema, 0)?;
head.schema.len() as u64
} else {
journal
.metadata()
.map_err(|e| anyhow!("uring: stat journal {}: {e}", jpath.display()))?
.len()
};
Ok(Self {
blobs,
journal,
journal_file_path: jpath,
cursor: AtomicU64::new(end),
ring: Mutex::new(Ring {
ring,
staging,
journal_cursor,
}),
})
}
pub fn journal_path(archive: &Path) -> PathBuf {
journal_path(archive)
}
pub fn blobs(&self) -> &File {
&self.blobs
}
pub fn journal_file(&self) -> &Path {
&self.journal_file_path
}
}
impl ArchiveWrite for UringWriter {
fn append(&self, bytes: &[u8]) -> Result<Extent> {
let len = bytes.len() as u64;
let offset = self.cursor.fetch_add(len, Ordering::SeqCst);
let chunks = bytes.len().div_ceil(MAX_WRITE_CHUNK).max(1);
if chunks + 3 > RING_ENTRIES as usize {
bail!(
"uring: {} B needs {chunks} linked writes, more than the ring's {RING_ENTRIES} \
entries",
bytes.len()
);
}
let row = encode_journal_row(offset, len)?;
let mut g = self
.ring
.lock()
.map_err(|_| anyhow!("uring mutex poisoned"))?;
if row.len() > g.staging.len() {
bail!(
"uring: journal row is {} B, staging buffer is {} B",
row.len(),
g.staging.len()
);
}
g.staging.as_mut_slice()[..row.len()].copy_from_slice(&row);
let journal_at = g.journal_cursor;
let blob_fd = types::Fd(self.blobs.as_raw_fd());
let journal_fd = types::Fd(self.journal.as_raw_fd());
let staging_ptr = g.staging.as_ptr();
let mut sqes: Vec<squeue::Entry> = Vec::with_capacity(chunks + 3);
for c in 0..chunks {
let start = c * MAX_WRITE_CHUNK;
let n = (bytes.len() - start).min(MAX_WRITE_CHUNK);
sqes.push(
opcode::Write::new(blob_fd, unsafe { bytes.as_ptr().add(start) }, n as u32)
.offset(offset + start as u64)
.build()
.flags(squeue::Flags::IO_LINK)
.user_data(c as u64),
);
}
sqes.push(
opcode::Fsync::new(blob_fd)
.build()
.flags(squeue::Flags::IO_LINK)
.user_data(0xF5_00),
);
sqes.push(
opcode::WriteFixed::new(journal_fd, staging_ptr, row.len() as u32, 0)
.offset(journal_at)
.build()
.flags(squeue::Flags::IO_LINK)
.user_data(0x30_01),
);
sqes.push(opcode::Fsync::new(journal_fd).build().user_data(0xF5_01));
let want = sqes.len();
unsafe {
g.ring
.submission()
.push_multiple(&sqes)
.map_err(|e| anyhow!("uring: submission queue full: {e}"))?;
}
g.ring
.submit_and_wait(want)
.map_err(|e| anyhow!("uring: io_uring_enter: {e}"))?;
let mut written = 0i64;
let mut seen = 0usize;
let mut journal_written = 0i64;
for cqe in g.ring.completion() {
seen += 1;
let res = cqe.result();
if res < 0 {
let e = std::io::Error::from_raw_os_error(-res);
bail!(
"uring: op {:#x} failed: {e} (a linked chain cancels its tail with \
ECANCELED, so no journal row was written after a failed blob fsync)",
cqe.user_data()
);
}
match cqe.user_data() {
0xF5_00 | 0xF5_01 => {}
0x30_01 => journal_written = res as i64,
_ => written += res as i64,
}
}
if seen != want {
bail!("uring: expected {want} completions, saw {seen}");
}
if written as u64 != len {
bail!(
"uring: short write — asked for {len} B, kernel wrote {written} B (io_uring \
Write is not write_all)"
);
}
if journal_written as usize != row.len() {
bail!(
"uring: short journal write — {} B of {} B",
journal_written,
row.len()
);
}
g.journal_cursor = journal_at + row.len() as u64;
Ok((offset, len))
}
fn name(&self) -> &'static str {
"UringWriter"
}
fn durability(&self) -> &'static str {
"full — kernel-ordered blob fsync then journal fsync, one io_uring_enter"
}
fn blobs_shared(&self) -> Option<Arc<File>> {
Some(Arc::clone(&self.blobs))
}
}
#[cfg(test)]
mod tests {
#[test]
fn the_probe_answers_what_create_would_and_leaves_no_residue() {
let probed = super::probe();
let dir = std::env::temp_dir().join(format!("znippy-uring-probe-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let built = super::UringWriter::create(&dir.join("objects.pack"));
assert_eq!(probed.is_ok(), built.is_ok(), "probe {probed:?} vs create {}", built.as_ref().err().map(|e| e.to_string()).unwrap_or_default());
if probed.is_ok() {
for _ in 0..8 {
super::probe().expect("a probe per process start leaves no residue");
}
}
drop(built);
let _ = std::fs::remove_dir_all(&dir);
}
use super::*;
use crate::archive_write::{ArchiveWrite, read_journal};
fn loadavg() -> String {
std::fs::read_to_string("/proc/loadavg")
.unwrap_or_default()
.split_whitespace()
.take(3)
.collect::<Vec<_>>()
.join(" ")
}
static MEMLOCK_MEASUREMENT: Mutex<()> = Mutex::new(());
struct Measuring {
_process: std::sync::MutexGuard<'static, ()>,
_uid: std::fs::File,
}
fn measuring_memlock() -> Measuring {
use std::os::fd::AsRawFd as _;
let process = MEMLOCK_MEASUREMENT
.lock()
.unwrap_or_else(|p| p.into_inner());
let uid = unsafe { libc::getuid() };
let path = std::env::temp_dir().join(format!("znippy-git-memlock-{uid}.lock"));
let file = std::fs::OpenOptions::new()
.create(true)
.truncate(false)
.write(true)
.open(&path)
.unwrap_or_else(|e| panic!("opening {} for the memlock lock: {e}", path.display()));
let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) };
assert_eq!(
rc,
0,
"could not take the uid-wide memlock measurement lock at {}: {}",
path.display(),
std::io::Error::last_os_error()
);
Measuring {
_process: process,
_uid: file,
}
}
const HEADROOM_CAP_PAGES: usize = 2048;
struct Scratch {
ptr: *mut libc::c_void,
len: usize,
}
impl Scratch {
fn new(len: usize) -> Self {
let ptr = unsafe {
libc::mmap(
std::ptr::null_mut(),
len,
libc::PROT_READ | libc::PROT_WRITE,
libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
-1,
0,
)
};
assert_ne!(ptr, libc::MAP_FAILED, "the probe's scratch mapping");
unsafe { libc::madvise(ptr, len, libc::MADV_NOHUGEPAGE) };
Self { ptr, len }
}
fn as_ptr(&self) -> *mut libc::c_void {
self.ptr
}
}
impl Drop for Scratch {
fn drop(&mut self) {
unsafe { libc::munmap(self.ptr, self.len) };
}
}
struct Headroom {
probe: IoUring,
scratch: Scratch,
}
const CLEAN_READINGS: usize = 5;
const READING_ATTEMPTS: usize = 25;
const SETTLE_TRIES: usize = 64;
impl Headroom {
fn new() -> Self {
let h = Self {
probe: IoUring::new(8).expect("a probe ring"),
scratch: Scratch::new(HEADROOM_CAP_PAGES * page_size()),
};
assert!(
h.pages() < HEADROOM_CAP_PAGES,
"this uid can register the whole {HEADROOM_CAP_PAGES}-page probe cap, so the \
headroom instrument is clamped and every charge it reads would be 0. Raise \
HEADROOM_CAP_PAGES above RLIMIT_MEMLOCK, or run where the limit is the stock \
8 MiB."
);
h
}
fn charge_of<T>(&self, acquire: impl FnOnce() -> T) -> Option<usize> {
let before = self.pages();
let held = acquire();
let during = self.pages();
drop(held);
let mut after = self.pages();
for _ in 0..SETTLE_TRIES {
if after >= before {
break;
}
std::thread::yield_now();
after = self.pages();
}
(after == before && during <= before).then(|| before - during)
}
fn fits(&self, pages: usize) -> bool {
let iov = libc::iovec {
iov_base: self.scratch.as_ptr(),
iov_len: pages * page_size(),
};
let ok = unsafe {
self.probe
.submitter()
.register_buffers(std::slice::from_ref(&iov))
}
.is_ok();
if ok {
self.probe
.submitter()
.unregister_buffers()
.expect("unregister the probe buffer");
}
ok
}
fn pages(&self) -> usize {
if self.fits(HEADROOM_CAP_PAGES) {
return HEADROOM_CAP_PAGES;
}
let (mut lo, mut hi) = (0usize, HEADROOM_CAP_PAGES);
while lo + 1 < hi {
let mid = (lo + hi) / 2;
if self.fits(mid) { lo = mid } else { hi = mid }
}
lo
}
}
const THP_ATTEMPTS: usize = 8;
struct HugeArena {
raw: *mut libc::c_void,
raw_len: usize,
arena: *mut u8,
}
impl HugeArena {
const HUGE: usize = 2 << 20;
fn new() -> Option<Self> {
for _ in 0..THP_ATTEMPTS {
let a = Self::new_once()?;
if a.anon_huge_pages_kb() > 0 {
return Some(a);
}
}
None
}
fn anon_huge_pages_kb(&self) -> u64 {
let Ok(maps) = std::fs::read_to_string("/proc/self/smaps") else {
return 0;
};
let want = self.arena as usize;
let mut inside = false;
for line in maps.lines() {
if let Some((range, _)) = line.split_once(' ') {
if let Some((lo, hi)) = range.split_once('-') {
if let (Ok(lo), Ok(hi)) =
(usize::from_str_radix(lo, 16), usize::from_str_radix(hi, 16))
{
inside = (lo..hi).contains(&want);
continue;
}
}
}
if inside {
if let Some(v) = line.strip_prefix("AnonHugePages:") {
return v
.trim()
.trim_end_matches(" kB")
.parse()
.unwrap_or(0);
}
}
}
0
}
fn new_once() -> Option<Self> {
let raw_len = 2 * Self::HUGE;
let raw = unsafe {
libc::mmap(
std::ptr::null_mut(),
raw_len,
libc::PROT_READ | libc::PROT_WRITE,
libc::MAP_PRIVATE | libc::MAP_ANONYMOUS,
-1,
0,
)
};
if raw == libc::MAP_FAILED {
return None;
}
let arena = ((raw as usize + Self::HUGE - 1) & !(Self::HUGE - 1)) as *mut u8;
unsafe {
if libc::madvise(arena as *mut libc::c_void, Self::HUGE, libc::MADV_HUGEPAGE) != 0 {
libc::munmap(raw, raw_len);
return None;
}
std::ptr::write_bytes(arena, 1u8, Self::HUGE);
}
Some(Self {
raw,
raw_len,
arena,
})
}
fn a_page(&self) -> *const u8 {
unsafe { self.arena.add(4 * page_size()) }
}
}
impl Drop for HugeArena {
fn drop(&mut self) {
unsafe { libc::munmap(self.raw, self.raw_len) };
}
}
struct HeldRegistration<'a>(&'a IoUring);
impl Drop for HeldRegistration<'_> {
fn drop(&mut self) {
self.0
.submitter()
.unregister_buffers()
.expect("release the control registration");
}
}
fn gather<T>(what: &str, h: &Headroom, acquire: impl Fn(usize) -> T) -> Vec<usize> {
let mut readings = Vec::with_capacity(CLEAN_READINGS);
for i in 0..READING_ATTEMPTS {
if let Some(n) = h.charge_of(|| acquire(i)) {
readings.push(n);
if readings.len() == CLEAN_READINGS {
return readings;
}
}
}
panic!(
"only {} of {CLEAN_READINGS} readings of {what} survived the bracket in \
{READING_ATTEMPTS} attempts (got {readings:?}, load {}). RLIMIT_MEMLOCK is charged \
uid-wide and something else on this uid kept moving it across the measurement \
window. That is a statement about the box, not about the writer: this guard needs \
exclusive use of the counter, it takes a uid-wide flock to get it, and something \
outside that lock is registering or releasing pinned memory.",
readings.len(),
loadavg(),
)
}
fn median(mut readings: Vec<usize>) -> usize {
readings.sort_unstable();
readings[readings.len() / 2]
}
#[test]
fn a_registration_costs_one_page_and_not_the_huge_page_it_might_sit_in() {
let _serial = measuring_memlock();
let h = Headroom::new();
let Some(arena) = HugeArena::new() else {
eprintln!(
"SKIPPED a_registration_costs_one_page_and_not_the_huge_page_it_might_sit_in: \
this box would not back a 2 MiB arena with a transparent huge page in \
{THP_ATTEMPTS} tries — `AnonHugePages` stayed 0 in /proc/self/smaps every \
time, with /sys/kernel/mm/transparent_hugepage/enabled = {:?} and load {}. \
Either THP is off, or memory is too fragmented right now to find 2 MiB \
contiguous; without one this guard cannot exhibit the 512× registration \
charge it exists for. NOT a pass: run it on a box with THP available.",
std::fs::read_to_string("/sys/kernel/mm/transparent_hugepage/enabled")
.unwrap_or_default()
.trim(),
loadavg(),
);
return;
};
let ctl = IoUring::new(RING_ENTRIES).expect("a control ring");
let huge_charge = median(gather(
"one page inside a MADV_HUGEPAGE arena",
&h,
|_| {
let iov = libc::iovec {
iov_base: arena.a_page() as *mut libc::c_void,
iov_len: page_size(),
};
unsafe {
ctl.submitter()
.register_buffers(std::slice::from_ref(&iov))
.expect("registering one page of a huge page")
};
HeldRegistration(&ctl)
},
));
assert!(
huge_charge >= 256,
"the positive control charged only {huge_charge} page(s) for one page inside a \
MADV_HUGEPAGE arena, so transparent huge pages are not actually in play here and \
the assertion below would pass on any implementation"
);
let dir = crate::store::tests::tmpdir("uring-registration-charge");
let readings = gather("a UringWriter", &h, |i| {
let w = UringWriter::create(&dir.join(format!("objects.pack.{i}")))
.unwrap_or_else(|e| {
panic!(
"writer {i} could not even be opened, with {} page(s) of \
RLIMIT_MEMLOCK headroom left. One page inside a transparent huge page \
costs {huge_charge} on this box, so a staging buffer that is not its \
own MADV_NOHUGEPAGE mapping caps a process at four registrations: \
{e:#}",
h.pages()
)
});
let (o, l) = w.append(b"one pack down the io_uring chain").unwrap();
assert_eq!(
read_journal(w.journal_file()).unwrap(),
vec![(o, l)],
"writer {i} acked without a journal row out of the registered buffer"
);
w
});
let charge = median(readings.clone());
assert!(
charge <= 64,
"a UringWriter's registration was charged {charge} page(s) (clean readings \
{readings:?}). \
One page inside a transparent huge page costs {huge_charge} on this box, and that \
is what a `Box<[u8]>` buys you under an allocator that madvises its arenas — \
mimalloc, which `gunnar serve` runs on. The staging buffer must be its own \
MADV_NOHUGEPAGE mapping (`Staging::map`), which costs the ring plus exactly one \
page."
);
assert!(
charge >= 1,
"a UringWriter cost 0 page(s) of RLIMIT_MEMLOCK across {} bracketed readings \
{readings:?} — every one of them taken with the uid-wide budget verified unmoved, \
so this is not noise: nothing was registered and the bound above is vacuous",
readings.len()
);
eprintln!(
"load {}; a UringWriter costs {charge} page(s) of RLIMIT_MEMLOCK (clean readings \
{readings:?}); one page inside a huge page costs {huge_charge}",
loadavg(),
);
}
fn writers_the_limit_must_allow() -> (usize, usize, usize) {
let page = page_size();
let Some(limit) = memlock_limit_bytes().map(|b| b as usize) else {
panic!(
"RLIMIT_MEMLOCK is unlimited on this box, so it cannot exhibit the ceiling these \
guards exist for. That is a property of the box, not a pass: run them somewhere \
with the stock 8 MiB limit before believing the arm scales."
);
};
let n = (limit / (16 * page) + 32).min(limit / (4 * page));
assert!(
n >= 8,
"RLIMIT_MEMLOCK is only {limit} B on this box — too small for a guard to separate a \
1-page registration from a 16-page one"
);
(n, limit, page)
}
#[test]
fn enough_uring_writers_for_a_population_coexist_in_one_process() {
let _serial = measuring_memlock();
let (n, limit, page) = writers_the_limit_must_allow();
let dir = crate::store::tests::tmpdir("uring-memlock-ceiling");
let payload = b"one pack down the io_uring chain".to_vec();
let mut held: Vec<UringWriter> = Vec::with_capacity(n);
let mut extents = Vec::with_capacity(n);
for i in 0..n {
let w = match UringWriter::create(&dir.join(format!("objects.pack.{i}"))) {
Ok(w) => w,
Err(e) => panic!(
"the io_uring arm ran out of pinned memory at writer {i} of {n}: {e:#}\n\
RLIMIT_MEMLOCK here is {limit} B and one page is {page} B, so {n} writers \
need {} B pinned. A server holds one writer per repository; this is the \
ceiling on how many repositories the arm can serve.",
n * page
),
};
extents.push(
w.append(&payload)
.unwrap_or_else(|e| panic!("writer {i} of {n} could not append: {e:#}")),
);
held.push(w);
}
let last = held.last().unwrap();
let (o, l) = *extents.last().unwrap();
let mut back = vec![0u8; l as usize];
last.blobs().read_exact_at(&mut back, o).unwrap();
assert_eq!(back, payload, "writer {} did not store the bytes", n - 1);
assert_eq!(
read_journal(last.journal_file()).unwrap(),
vec![(o, l)],
"writer {} acked without a journal row naming the extent",
n - 1
);
eprintln!(
"load {}; {n} io_uring writers open at once, {} B pinned of a {limit} B \
RLIMIT_MEMLOCK ({page} B/writer; the 64 KiB registration this replaced would have \
needed {} B)",
loadavg(),
n * page,
n * 16 * page,
);
}
#[test]
fn a_reopened_uring_writer_appends_to_its_journal_rather_than_truncating_it() {
let dir = crate::store::tests::tmpdir("uring-journal-reopen");
let blobs = dir.join("objects.pack");
let mut want = Vec::new();
for (i, len) in [512usize, 512, 1024].into_iter().enumerate() {
let w = UringWriter::create(&blobs)
.unwrap_or_else(|e| panic!("open {i} of the same archive: {e:#}"));
want.push(w.append(&vec![b'a' + i as u8; len]).unwrap());
}
assert_eq!(
read_journal(&UringWriter::journal_path(&blobs)).unwrap(),
want,
"the reopened io_uring writer lost the journal rows written before it"
);
assert_eq!(want, vec![(0, 512), (512, 512), (1024, 1024)]);
eprintln!(
"load {}; three io_uring writers over one archive: journal {:?}",
loadavg(),
want
);
}
#[test]
fn a_journal_row_fits_the_page_that_is_pinned_for_it() {
let row = crate::archive_write::encode_journal_row(u64::MAX, u64::MAX).unwrap();
assert!(
row.len() <= JOURNAL_STAGING,
"a journal row is {} B and the registered staging buffer is {JOURNAL_STAGING} B",
row.len()
);
eprintln!(
"load {}; journal row {} B into a {JOURNAL_STAGING} B registered buffer ({} page)",
loadavg(),
row.len(),
JOURNAL_STAGING / page_size()
);
}
}