alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::time::Duration;
2
3use knot_resource::DecayMs;
4use tokio_util::sync::CancellationToken;
5
6const SAMPLE_INTERVAL: Duration = Duration::from_secs(2);
7const BACKGROUND_THREAD: &[u8] = b"background_thread\0";
8const NARENAS: &[u8] = b"arenas.narenas\0";
9const DIRTY_DECAY_NEW_ARENAS: &[u8] = b"arenas.dirty_decay_ms\0";
10
11#[derive(Clone, Copy)]
12struct ArenaIndex(u32);
13
14impl ArenaIndex {
15 fn write_decay(self, decay: DecayMs) -> bool {
16 let key = format!("arena.{}.dirty_decay_ms\0", self.0);
17 unsafe { tikv_jemalloc_ctl::raw::write(key.as_bytes(), decay.ms()) }.is_ok()
18 }
19}
20
21#[derive(Clone, Copy)]
22struct DecayCoverage {
23 reached: usize,
24 skipped: usize,
25}
26
27impl DecayCoverage {
28 const EMPTY: Self = Self {
29 reached: 0,
30 skipped: 0,
31 };
32
33 fn record(self, reached: bool) -> Self {
34 match reached {
35 true => Self {
36 reached: self.reached + 1,
37 ..self
38 },
39 false => Self {
40 skipped: self.skipped + 1,
41 ..self
42 },
43 }
44 }
45}
46
47fn write_or_warn<T>(name: &[u8], value: T, control: &str) {
48 if let Err(error) = unsafe { tikv_jemalloc_ctl::raw::write(name, value) } {
49 tracing::warn!(%error, control, "jemalloc control unavailable");
50 }
51}
52
53fn apply_decay(decay: DecayMs) -> tikv_jemalloc_ctl::Result<DecayCoverage> {
54 unsafe { tikv_jemalloc_ctl::raw::write(DIRTY_DECAY_NEW_ARENAS, decay.ms())? };
55 let narenas: u32 = unsafe { tikv_jemalloc_ctl::raw::read(NARENAS)? };
56 Ok((0..narenas)
57 .map(ArenaIndex)
58 .map(|arena| arena.write_decay(decay))
59 .fold(DecayCoverage::EMPTY, DecayCoverage::record))
60}
61
62pub fn install() {
63 match apply_decay(knot_resource::target_decay()) {
64 Ok(coverage) => tracing::info!(
65 arenas_reached = coverage.reached,
66 arenas_skipped = coverage.skipped,
67 "jemalloc dirty_decay applied to reachable arenas"
68 ),
69 Err(error) => tracing::warn!(%error, "jemalloc dirty_decay governor unavailable"),
70 }
71 write_or_warn(BACKGROUND_THREAD, true, "background_thread");
72 let background: bool =
73 unsafe { tikv_jemalloc_ctl::raw::read(b"opt.background_thread\0") }.unwrap_or(false);
74 let retain: bool = unsafe { tikv_jemalloc_ctl::raw::read(b"opt.retain\0") }.unwrap_or(true);
75 let dirty_decay_ms: isize =
76 unsafe { tikv_jemalloc_ctl::raw::read(b"arena.0.dirty_decay_ms\0") }.unwrap_or(-1);
77 tracing::info!(
78 background_thread = background,
79 retain,
80 dirty_decay_ms,
81 "jemalloc page-return configured"
82 );
83}
84
85pub async fn govern_decay(shutdown: CancellationToken) {
86 let mut ticker = tokio::time::interval(SAMPLE_INTERVAL);
87 let mut applied = knot_resource::target_decay();
88 loop {
89 tokio::select! {
90 () = shutdown.cancelled() => return,
91 _ = ticker.tick() => {
92 let target = knot_resource::target_decay();
93 if knot_resource::decay_warrants_apply(applied, target) {
94 match apply_decay(target) {
95 Ok(coverage) => tracing::info!(
96 target_ms = target.ms(),
97 arenas_reached = coverage.reached,
98 arenas_skipped = coverage.skipped,
99 "jemalloc dirty_decay retuned"
100 ),
101 Err(error) => tracing::warn!(%error, "jemalloc dirty_decay retune failed"),
102 }
103 applied = target;
104 }
105 if knot_resource::cache_shed_warranted()
106 && let Some(freed) = knot_cache::reclaim_largest()
107 {
108 tracing::warn!(
109 freed_bytes = freed.get(),
110 "shedding largest cache under memory pressure"
111 );
112 }
113 }
114 }
115 }
116}
117
118#[cfg(test)]
119mod tests {
120 use super::*;
121
122 #[test]
123 fn apply_decay_reaches_reachable_arenas_and_persists() {
124 let narenas: u32 =
125 unsafe { tikv_jemalloc_ctl::raw::read(NARENAS) }.expect("arenas.narenas is readable");
126 assert!(narenas >= 1, "a live jemalloc has at least one arena");
127
128 let coverage = apply_decay(knot_resource::target_decay())
129 .expect("new-arena default write and narenas read succeed");
130 assert!(
131 coverage.reached >= 1,
132 "arena 0 is always initialized, so the sweep reaches at least one arena"
133 );
134 assert_eq!(
135 coverage.reached + coverage.skipped,
136 narenas as usize,
137 "every arena index is accounted as reached or skipped"
138 );
139
140 unsafe {
141 tikv_jemalloc_ctl::raw::write(b"arena.0.dirty_decay_ms\0", 5_000isize)
142 .expect("per-arena dirty_decay_ms is writable");
143 let back: isize = tikv_jemalloc_ctl::raw::read(b"arena.0.dirty_decay_ms\0")
144 .expect("per-arena dirty_decay_ms is readable");
145 assert_eq!(back, 5_000, "a per-arena write reads back");
146 }
147 }
148}