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}
108
109struct HeapPinGuard {
114 page_start: *mut libc::c_void,
115 page_len: usize,
116}
117
118unsafe impl Send for HeapPinGuard {}
121unsafe impl Sync for HeapPinGuard {}
122
123impl Drop for HeapPinGuard {
124 fn drop(&mut self) {
125 if unsafe { libc::munlock(self.page_start, self.page_len) } != 0 {
126 log::warn!(
127 "[pin] munlock failed for {} ANN heap bytes: {}",
128 self.page_len,
129 std::io::Error::last_os_error()
130 );
131 }
132 }
133}
134
135#[derive(Default)]
138pub(crate) struct HeapPinSet {
139 guards: Vec<HeapPinGuard>,
140 owners: Vec<Arc<dyn std::any::Any + Send + Sync>>,
143 report: PinReport,
144}
145
146impl HeapPinSet {
147 pub(crate) fn report(&self) -> PinReport {
148 self.report
149 }
150
151 pub(crate) fn retain_owner<T: std::any::Any + Send + Sync>(&mut self, owner: Arc<T>) {
152 self.owners.push(owner);
153 }
154
155 pub(crate) fn pin_slice<T>(
160 &mut self,
161 slice: &[T],
162 label: &str,
163 mode: PinMode,
164 remaining: &mut u64,
165 ) {
166 let len = std::mem::size_of_val(slice);
167 if len == 0 {
168 return;
169 }
170 let Ok(len_u64) = u64::try_from(len) else {
171 self.report.failed_bytes = u64::MAX;
172 log::warn!("[pin] ANN region {label} is too large to account");
173 return;
174 };
175 self.report.intended_bytes = self.report.intended_bytes.saturating_add(len_u64);
176 if len_u64 > *remaining {
177 self.report.skipped_budget_bytes =
178 self.report.skipped_budget_bytes.saturating_add(len_u64);
179 log::debug!(
180 "[pin] ANN budget exhausted: skipping {} ({} bytes, {} remaining)",
181 label,
182 len_u64,
183 *remaining
184 );
185 return;
186 }
187
188 if mode == PinMode::Copy {
189 *remaining -= len_u64;
190 self.report.pinned_bytes = self.report.pinned_bytes.saturating_add(len_u64);
191 return;
192 }
193
194 let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
195 let page_size = usize::try_from(page_size).ok().filter(|&size| size > 0);
196 let Some(page_size) = page_size else {
197 self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
198 log::warn!("[pin] cannot determine page size while locking {label}");
199 return;
200 };
201 let address = slice.as_ptr() as usize;
202 let page_start = address / page_size * page_size;
203 let Some(end) = address.checked_add(len) else {
204 self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
205 log::warn!("[pin] ANN region address overflow while locking {label}");
206 return;
207 };
208 let Some(rounded_end) = end
209 .checked_add(page_size - 1)
210 .map(|value| value / page_size * page_size)
211 else {
212 self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
213 log::warn!("[pin] ANN region page range overflow while locking {label}");
214 return;
215 };
216 let page_len = rounded_end - page_start;
217 let page_start = page_start as *mut libc::c_void;
218 if unsafe { libc::mlock(page_start.cast_const(), page_len) } == 0 {
219 self.guards.push(HeapPinGuard {
220 page_start,
221 page_len,
222 });
223 *remaining -= len_u64;
224 self.report.pinned_bytes = self.report.pinned_bytes.saturating_add(len_u64);
225 } else {
226 self.report.failed_bytes = self.report.failed_bytes.saturating_add(len_u64);
227 log::warn!(
228 "[pin] mlock failed for ANN {} ({} bytes): {} — check RLIMIT_MEMLOCK/CAP_IPC_LOCK; continuing unpinned",
229 label,
230 len_u64,
231 std::io::Error::last_os_error()
232 );
233 }
234 }
235}
236
237pub(crate) fn pin_section(
244 bytes: &mut OwnedBytes,
245 label: &str,
246 mode: PinMode,
247 remaining: &mut u64,
248 report: &mut PinReport,
249) {
250 if !bytes.is_mmap() || bytes.is_empty() {
251 return;
252 }
253 let len = bytes.len() as u64;
254 report.intended_bytes += len;
255
256 if len > *remaining {
257 report.skipped_budget_bytes += len;
258 log::debug!(
259 "[pin] budget exhausted: skipping {} ({} bytes, {} remaining)",
260 label,
261 len,
262 *remaining
263 );
264 return;
265 }
266
267 match mode {
268 PinMode::Mlock => {
269 if bytes.mlock() {
270 *remaining -= len;
271 report.pinned_bytes += len;
272 } else {
273 report.failed_bytes += len;
274 log::warn!(
275 "[pin] mlock failed for {} ({} bytes) — check RLIMIT_MEMLOCK; \
276 continuing unpinned",
277 label,
278 len
279 );
280 }
281 }
282 PinMode::Copy => {
283 *bytes = OwnedBytes::new(bytes.to_vec());
284 *remaining -= len;
285 report.pinned_bytes += len;
286 }
287 }
288}