use std::sync::{Arc, OnceLock};
use crate::directories::OwnedBytes;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PinMode {
Mlock,
Copy,
}
#[derive(Debug, Clone, Copy)]
pub struct PinPolicy {
pub budget_bytes: u64,
pub mode: PinMode,
}
impl PinPolicy {
pub const fn disabled() -> Self {
Self {
budget_bytes: 0,
mode: PinMode::Mlock,
}
}
pub fn is_enabled(&self) -> bool {
self.budget_bytes > 0
}
pub fn from_env() -> Self {
let budget_mb: u64 = std::env::var("HERMES_PIN_METADATA_BUDGET_MB")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(0);
let mode = match std::env::var("HERMES_PIN_MODE").as_deref() {
Ok("copy") => PinMode::Copy,
Ok("mlock") | Err(_) => PinMode::Mlock,
Ok(other) => {
log::warn!("HERMES_PIN_MODE '{}' unknown; using mlock", other);
PinMode::Mlock
}
};
Self {
budget_bytes: budget_mb * 1024 * 1024,
mode,
}
}
}
static PIN_POLICY: OnceLock<PinPolicy> = OnceLock::new();
pub fn set_pin_policy(policy: PinPolicy) -> bool {
let ok = PIN_POLICY.set(policy).is_ok();
if !ok {
log::warn!("pin policy already initialized; set_pin_policy ignored");
}
ok
}
pub fn pin_policy() -> &'static PinPolicy {
PIN_POLICY.get_or_init(PinPolicy::from_env)
}
#[derive(Debug, Default, Clone, Copy)]
pub struct PinReport {
pub intended_bytes: u64,
pub pinned_bytes: u64,
pub skipped_budget_bytes: u64,
pub failed_bytes: u64,
}
struct HeapPinGuard {
page_start: *mut libc::c_void,
page_len: usize,
}
unsafe impl Send for HeapPinGuard {}
unsafe impl Sync for HeapPinGuard {}
impl Drop for HeapPinGuard {
fn drop(&mut self) {
if unsafe { libc::munlock(self.page_start, self.page_len) } != 0 {
log::warn!(
"[pin] munlock failed for {} ANN heap bytes: {}",
self.page_len,
std::io::Error::last_os_error()
);
}
}
}
#[derive(Default)]
pub(crate) struct HeapPinSet {
guards: Vec<HeapPinGuard>,
owners: Vec<Arc<dyn std::any::Any + Send + Sync>>,
report: PinReport,
}
impl HeapPinSet {
pub(crate) fn report(&self) -> PinReport {
self.report
}
pub(crate) fn retain_owner<T: std::any::Any + Send + Sync>(&mut self, owner: Arc<T>) {
self.owners.push(owner);
}
pub(crate) fn pin_slice<T>(
&mut self,
slice: &[T],
label: &str,
mode: PinMode,
remaining: &mut u64,
) {
let len = std::mem::size_of_val(slice);
if len == 0 {
return;
}
let Ok(len_u64) = u64::try_from(len) else {
self.report.failed_bytes = u64::MAX;
log::warn!("[pin] ANN region {label} is too large to account");
return;
};
self.report.intended_bytes = self.report.intended_bytes.saturating_add(len_u64);
if len_u64 > *remaining {
self.report.skipped_budget_bytes =
self.report.skipped_budget_bytes.saturating_add(len_u64);
log::debug!(
"[pin] ANN budget exhausted: skipping {} ({} bytes, {} remaining)",
label,
len_u64,
*remaining
);
return;
}
if mode == PinMode::Copy {
*remaining -= len_u64;
self.report.pinned_bytes = self.report.pinned_bytes.saturating_add(len_u64);
return;
}
let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
let page_size = usize::try_from(page_size).ok().filter(|&size| size > 0);
let Some(page_size) = page_size else {
self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
log::warn!("[pin] cannot determine page size while locking {label}");
return;
};
let address = slice.as_ptr() as usize;
let page_start = address / page_size * page_size;
let Some(end) = address.checked_add(len) else {
self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
log::warn!("[pin] ANN region address overflow while locking {label}");
return;
};
let Some(rounded_end) = end
.checked_add(page_size - 1)
.map(|value| value / page_size * page_size)
else {
self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
log::warn!("[pin] ANN region page range overflow while locking {label}");
return;
};
let page_len = rounded_end - page_start;
let page_start = page_start as *mut libc::c_void;
if unsafe { libc::mlock(page_start.cast_const(), page_len) } == 0 {
self.guards.push(HeapPinGuard {
page_start,
page_len,
});
*remaining -= len_u64;
self.report.pinned_bytes = self.report.pinned_bytes.saturating_add(len_u64);
} else {
self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
log::warn!(
"[pin] mlock failed for ANN {} ({} bytes): {} — check RLIMIT_MEMLOCK/CAP_IPC_LOCK; continuing unpinned",
label,
len_u64,
std::io::Error::last_os_error()
);
}
}
}
pub(crate) fn pin_section(
bytes: &mut OwnedBytes,
label: &str,
mode: PinMode,
remaining: &mut u64,
report: &mut PinReport,
) {
if !bytes.is_mmap() || bytes.is_empty() {
return;
}
let len = bytes.len() as u64;
report.intended_bytes += len;
if len > *remaining {
report.skipped_budget_bytes += len;
log::debug!(
"[pin] budget exhausted: skipping {} ({} bytes, {} remaining)",
label,
len,
*remaining
);
return;
}
match mode {
PinMode::Mlock => {
if bytes.mlock() {
*remaining -= len;
report.pinned_bytes += len;
} else {
report.failed_bytes += len;
log::warn!(
"[pin] mlock failed for {} ({} bytes) — check RLIMIT_MEMLOCK; \
continuing unpinned",
label,
len
);
}
}
PinMode::Copy => {
*bytes = OwnedBytes::new(bytes.to_vec());
*remaining -= len;
report.pinned_bytes += len;
}
}
}