hermes_core/segment/
pin.rs1use std::sync::{Arc, OnceLock};
15
16use crate::directories::OwnedBytes;
17
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub enum PinMode {
21 Mlock,
25 Copy,
28}
29
30#[derive(Debug, Clone, Copy)]
32pub struct PinPolicy {
33 pub budget_bytes: u64,
35 pub mode: PinMode,
36}
37
38impl PinPolicy {
39 pub const fn disabled() -> Self {
40 Self {
41 budget_bytes: 0,
42 mode: PinMode::Mlock,
43 }
44 }
45
46 pub fn is_enabled(&self) -> bool {
47 self.budget_bytes > 0
48 }
49
50 pub fn from_env() -> Self {
59 let budget_mb: u64 = std::env::var("HERMES_PIN_METADATA_BUDGET_MB")
60 .ok()
61 .and_then(|v| v.parse().ok())
62 .unwrap_or(0);
63 let mode = match std::env::var("HERMES_PIN_MODE").as_deref() {
64 Ok("copy") => PinMode::Copy,
65 Ok("mlock") | Err(_) => PinMode::Mlock,
66 Ok(other) => {
67 log::warn!("HERMES_PIN_MODE '{}' unknown; using mlock", other);
68 PinMode::Mlock
69 }
70 };
71 Self {
72 budget_bytes: budget_mb * 1024 * 1024,
73 mode,
74 }
75 }
76}
77
78static PIN_POLICY: OnceLock<PinPolicy> = OnceLock::new();
79
80pub fn set_pin_policy(policy: PinPolicy) -> bool {
84 let ok = PIN_POLICY.set(policy).is_ok();
85 if !ok {
86 log::warn!("pin policy already initialized; set_pin_policy ignored");
87 }
88 ok
89}
90
91pub fn pin_policy() -> &'static PinPolicy {
93 PIN_POLICY.get_or_init(PinPolicy::from_env)
94}
95
96#[derive(Debug, Default, Clone, Copy)]
98pub struct PinReport {
99 pub intended_bytes: u64,
101 pub pinned_bytes: u64,
103 pub skipped_budget_bytes: u64,
105 pub failed_bytes: u64,
107 pub heap_copy_bytes: u64,
110}
111
112struct HeapPinGuard {
117 page_start: *mut libc::c_void,
118 page_len: usize,
119}
120
121unsafe impl Send for HeapPinGuard {}
124unsafe impl Sync for HeapPinGuard {}
125
126impl Drop for HeapPinGuard {
127 fn drop(&mut self) {
128 if unsafe { libc::munlock(self.page_start, self.page_len) } != 0 {
129 log::warn!(
130 "[pin] munlock failed for {} of ANN heap: {}",
131 crate::format_bytes(self.page_len as u64),
132 std::io::Error::last_os_error()
133 );
134 }
135 }
136}
137
138#[derive(Default)]
141pub(crate) struct HeapPinSet {
142 guards: Vec<HeapPinGuard>,
143 owners: Vec<Arc<dyn std::any::Any + Send + Sync>>,
146 report: PinReport,
147}
148
149impl HeapPinSet {
150 pub(crate) fn report(&self) -> PinReport {
151 self.report
152 }
153
154 pub(crate) fn retain_owner<T: std::any::Any + Send + Sync>(&mut self, owner: Arc<T>) {
155 self.owners.push(owner);
156 }
157
158 pub(crate) fn pin_slice<T>(
163 &mut self,
164 slice: &[T],
165 label: &str,
166 mode: PinMode,
167 remaining: &mut u64,
168 ) {
169 let len = std::mem::size_of_val(slice);
170 if len == 0 {
171 return;
172 }
173 let Ok(len_u64) = u64::try_from(len) else {
174 self.report.failed_bytes = u64::MAX;
175 log::warn!("[pin] ANN region {label} is too large to account");
176 return;
177 };
178 self.report.intended_bytes = self.report.intended_bytes.saturating_add(len_u64);
179 if len_u64 > *remaining {
180 self.report.skipped_budget_bytes =
181 self.report.skipped_budget_bytes.saturating_add(len_u64);
182 log::debug!(
183 "[pin] ANN budget exhausted: skipping {} ({}, {} remaining)",
184 label,
185 crate::format_bytes(len_u64),
186 crate::format_bytes(*remaining)
187 );
188 return;
189 }
190
191 if mode == PinMode::Copy {
192 *remaining -= len_u64;
193 self.report.pinned_bytes = self.report.pinned_bytes.saturating_add(len_u64);
194 return;
195 }
196
197 let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
198 let page_size = usize::try_from(page_size).ok().filter(|&size| size > 0);
199 let Some(page_size) = page_size else {
200 self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
201 log::warn!("[pin] cannot determine page size while locking {label}");
202 return;
203 };
204 let address = slice.as_ptr() as usize;
205 let page_start = address / page_size * page_size;
206 let Some(end) = address.checked_add(len) else {
207 self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
208 log::warn!("[pin] ANN region address overflow while locking {label}");
209 return;
210 };
211 let Some(rounded_end) = end
212 .checked_add(page_size - 1)
213 .map(|value| value / page_size * page_size)
214 else {
215 self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
216 log::warn!("[pin] ANN region page range overflow while locking {label}");
217 return;
218 };
219 let page_len = rounded_end - page_start;
220 let page_start = page_start as *mut libc::c_void;
221 if unsafe { libc::mlock(page_start.cast_const(), page_len) } == 0 {
222 self.guards.push(HeapPinGuard {
223 page_start,
224 page_len,
225 });
226 *remaining -= len_u64;
227 self.report.pinned_bytes = self.report.pinned_bytes.saturating_add(len_u64);
228 } else {
229 self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
230 log::warn!(
231 "[pin] mlock failed for ANN {} ({}): {} — check RLIMIT_MEMLOCK/CAP_IPC_LOCK; continuing unpinned",
232 label,
233 crate::format_bytes(len_u64),
234 std::io::Error::last_os_error()
235 );
236 }
237 }
238}
239
240pub(crate) fn pin_section(
247 bytes: &mut OwnedBytes,
248 label: &str,
249 mode: PinMode,
250 remaining: &mut u64,
251 report: &mut PinReport,
252) {
253 if !bytes.is_mmap() || bytes.is_empty() {
254 return;
255 }
256 let len = bytes.len() as u64;
257 report.intended_bytes += len;
258
259 if len > *remaining {
260 report.skipped_budget_bytes += len;
261 log::debug!(
262 "[pin] budget exhausted: skipping {} ({}, {} remaining)",
263 label,
264 crate::format_bytes(len),
265 crate::format_bytes(*remaining)
266 );
267 return;
268 }
269
270 match mode {
271 PinMode::Mlock => {
272 if bytes.mlock() {
273 *remaining -= len;
274 report.pinned_bytes += len;
275 } else {
276 report.failed_bytes += len;
277 log::warn!(
278 "[pin] mlock failed for {} ({}) — check RLIMIT_MEMLOCK; \
279 continuing unpinned",
280 label,
281 crate::format_bytes(len)
282 );
283 }
284 }
285 PinMode::Copy => {
286 *bytes = OwnedBytes::new(bytes.to_vec());
287 *remaining -= len;
288 report.pinned_bytes += len;
289 report.heap_copy_bytes += len;
290 }
291 }
292}