Oregami
Repositories/oxedyne/fe2o3

oxedyne/fe2o3/fe2o3_o3db_sync/src/base/cfg.rs

16.8 KiB, 73 runs

created by r1870400018:719, which is this file's identity for as long as the history lasts, whatever it is later renamed to

download · who wrote it · its history

1use crate::{
2 prelude::*,
3 bots::worker::bot::WorkerType,
4};
5
6use oxedyne_fe2o3_jdat::{
7 prelude::*,
8 cfg::Config,
9 chunk::{
10 Chunker,
11 ChunkConfig,
12 },
13 id::NumIdDat,
14};
15
16use std::{
17 collections::BTreeMap,
18 io::Write,
19 path::{
20 Path,
21 PathBuf,
22 },
23 time::Duration,
24};
25
26use num_cpus;
27
28
29/// Use these as format strings in macros.
30#[macro_export]
31macro_rules! format_zones_dir { () => { "{:03}_zone" } }
32#[macro_export]
33macro_rules! format_zone_dir { () => { "zone_{:03}" } }
34#[macro_export]
35macro_rules! format_data_file { () => { "{:09}" } } // Further separated like 000_000_000 in file/zdir.rs
36#[macro_export]
37macro_rules! regex_data_file { () => { r"\d{3}_\d{3}_\d{3}" } }
38
39impl<
40 const UIDL: usize,
41 UID: NumIdDat<UIDL>,
42 ENC: Encrypter,
43 KH: Hasher,
44 PR: Hasher,
45 CS: Checksummer,
46>
47 O3db<UIDL, UID, ENC, KH, PR, CS>
48{
49 // System reflection.
50 pub fn num_cpus() -> usize {
51 num_cpus::get()
52 }
53}
54
55/// Holds primary configuration information for an ozone database in a Rust struct that can be
56/// read and written directly to file. The `zones` field must be a `Vec<Daticle>` because
57/// `FromDatMap` and `ToDatMap` currently do not handle recursion.
58#[derive(Clone, Debug, Eq, PartialEq, FromDatMap, ToDatMap)]
59pub struct OzoneConfig {
60 pub format_version: u8,
61 // Key hashing
62 pub bytes_before_hashing: u64, // applies only to keys
63 // Caches
64 pub cache_size_limit_bytes: u64,
65 pub init_load_caches: bool,
66 // Files
67 pub data_file_max_bytes: u64,
68 // Chunking
69 pub rest_chunk_threshold: u64, // applies only to values
70 pub rest_chunk_bytes: u64,
71 // Bots
72 pub num_cbots_per_zone: u16, // cache bots
73 pub num_fbots_per_zone: u16, // file bots
74 pub num_igbots_per_zone: u16, // init and garbage collecting bots
75 pub num_rbots_per_zone: u16, // reader bots
76 pub num_wbots_per_zone: u16, // writer bots
77 pub num_sbots: u16, // server bots
78 #[optional]
79 pub num_scbots_per_zone: u16, // scan bots
80 // Zones
81 pub num_zones: u16,
82 pub zone_state_update_secs: u8,
83 pub zone_overrides: BTreeMap<Dat, Dat>,
84 // ── Durability barrier ────────────────────────────────────────────────
85 //
86 // All three fields are `#[optional]` so an on-disk config written
87 // by a pre-durability-barrier Steel binary still loads. Missing
88 // fields fall through to the Default impl below, which reproduces
89 // the pre-feature behaviour (no fsync at all). Operators upgrade
90 // by editing the on-disk file or by letting Steel rewrite it on
91 // next clean start.
92
93 #[optional]
94 pub sync_on_write: bool,
95 #[optional]
96 pub sync_every_n_writes: u32,
97 #[optional]
98 pub sync_interval_ms: u64,
99}
100
101impl Config for OzoneConfig {
102
103 /// Performs a sequence of checks of the configuration data.
104 fn check_and_fix(&mut self) -> Outcome<()> {
105 res!(self.check_format_version());
106 res!(self.check_rest_chunk_config(&self.chunk_config()));
107 res!(self.check_file_size());
108 self.fix_scan_bot_count();
109 Ok(())
110 }
111}
112
113impl Default for OzoneConfig {
114 fn default() -> Self {
115 Self {
116 format_version: constant::CURRENT_FORMAT_VERSION,
117 // Key hashing
118 bytes_before_hashing: 32,
119 // Caches
120 cache_size_limit_bytes: 1_073_742_000, // 1 GiB
121 init_load_caches: true,
122 // Files
123 data_file_max_bytes: 1_048_576, // 1 MiB
124 // Chunking
125 rest_chunk_threshold: 716_800, // 700 KiB,
126 rest_chunk_bytes: 102_400, // 100 KiB
127 // Bots
128 num_cbots_per_zone: 2,
129 num_fbots_per_zone: 2,
130 num_igbots_per_zone: 2,
131 num_rbots_per_zone: 2,
132 num_wbots_per_zone: 2,
133 num_sbots: 2,
134 // One scan bot answers a zone's scans in the order they
135 // arrive. A second only helps when concurrent scans of the
136 // same zone are wanted, and each costs a thread.
137 num_scbots_per_zone: 1,
138 // Zones
139 num_zones: 2,
140 zone_state_update_secs: 5,
141 zone_overrides: mapdat!{ // use mapdat! for convenience.
142 1u16 => mapdat!{
143 "dir" => "",
144 "max_size" => 104_857_600u64,
145 },
146 2u16 => mapdat!{
147 "dir" => "",
148 "max_size" => 104_857_600u64,
149 },
150 }.get_map().unwrap(),
151 // Durability barrier. The default is a bounded group-commit: an
152 // acknowledged write is fsynced within `sync_interval_ms` of the
153 // last fsync, giving a durability floor without paying an fsync per
154 // write. 200 ms is the group-commit window: it bounds crash loss on
155 // the live file to roughly the writes of the last interval while
156 // leaving the hot path to ack into the page cache as before, so a
157 // wbot pays about one data+index fsync per 200 ms of sustained
158 // writing rather than one per write. This is independent of, and in
159 // addition to, the unconditional fsync barriers on seal and before
160 // the GC rename, which bound loss to the live file's tail under any
161 // policy. Operators wanting a stronger floor set `sync_on_write` (an
162 // fsync per write) or `sync_every_n_writes`; `sync_interval_ms = 0`
163 // restores the old never-fsync append behaviour. Each writer's
164 // syncer thread keeps the interval whether or not another write
165 // arrives, so the writes that end a burst are made durable too.
166 sync_on_write: false,
167 sync_every_n_writes: 0,
168 sync_interval_ms: 200,
169 }
170 }
171}
172
173impl OzoneConfig {
174 pub fn check_format_version(&self) -> Outcome<()> {
175 if self.format_version != constant::CURRENT_FORMAT_VERSION {
176 return Err(err!(
177 "Ozone database format version mismatch: config declares \
178 v{}, this binary expects v{}. No automatic migration path \
179 exists; the database must be re-created with the current \
180 ozone version.",
181 self.format_version, constant::CURRENT_FORMAT_VERSION;
182 Configuration, Invalid, Mismatch));
183 }
184 Ok(())
185 }
186
187 pub fn fix_scan_bot_count(&mut self) {
188 if self.num_scbots_per_zone == 0 {
189 warn!(sync_log::stream(),
190 "Configured scan bots per zone is zero, which leaves no bot to \
191 answer a scan; using one.");
192 self.num_scbots_per_zone = 1;
193 }
194 }
195
196 pub fn rest_chunk_size(&self) -> usize { self.rest_chunk_bytes as usize }
197 pub fn rest_chunking_threshold(&self) -> usize { self.rest_chunk_threshold as usize }
198 pub fn hashing_threshold(&self) -> usize { self.bytes_before_hashing as usize }
199
200 pub fn num_zones(&self) -> usize { self.num_zones as usize }
201 pub fn num_cbots_per_zone(&self) -> usize { self.num_cbots_per_zone as usize }
202 pub fn num_fbots_per_zone(&self) -> usize { self.num_fbots_per_zone as usize }
203 pub fn num_sbots(&self) -> usize { self.num_sbots as usize }
204
205 pub fn num_caches(&self) -> usize {
206 let nz = self.num_zones as usize;
207 let nbots = self.num_cbots_per_zone as usize;
208 nz * nbots
209 }
210
211 pub fn num_filemaps(&self) -> usize {
212 let nz = self.num_zones as usize;
213 let nbots = self.num_fbots_per_zone as usize;
214 nz * nbots
215 }
216
217 pub fn num_wbots(&self) -> usize {
218 let nz = self.num_zones as usize;
219 let nbots = self.num_wbots_per_zone as usize;
220 nz * nbots
221 }
222
223 pub fn num_bots_per_zone(&self, wtyp: &WorkerType) -> usize {
224 (match wtyp {
225 WorkerType::Cache => self.num_cbots_per_zone,
226 WorkerType::File => self.num_fbots_per_zone,
227 WorkerType::InitGarbage => self.num_igbots_per_zone,
228 WorkerType::Reader => self.num_rbots_per_zone,
229 WorkerType::Scan => self.num_scbots_per_zone,
230 WorkerType::Writer => self.num_wbots_per_zone,
231 }) as usize
232 }
233
234 pub fn zone_config(&self) -> ZoneConfig {
235 let nz = self.num_zones();
236 let nc = self.num_cbots_per_zone as usize;
237 let nf = self.num_fbots_per_zone as usize;
238
239 let total_num_caches = self.num_caches();
240 let cache_lim = self.cache_size_limit_bytes as usize;
241 let per_cache_lim = cache_lim / total_num_caches;
242 info!(sync_log::stream(),
243 "Total size limit for all caches, {} [B], will be used to set the size \
244 limit of {} [B] for {} caches across {} zones.",
245 cache_lim, per_cache_lim, total_num_caches, nz,
246 );
247 info!(sync_log::stream(),
248 "File data will {}be loaded into caches upon initialisation.",
249 if self.init_load_caches { "" } else { "not" },
250 );
251 ZoneConfig {
252 ncbots: nc,
253 nfbots: nf,
254 cache_size_lim: per_cache_lim,
255 init_load_caches: self.init_load_caches,
256 }
257 }
258
259 pub fn zone_overrides(&self) -> &BTreeMap<Dat, Dat> { &self.zone_overrides }
260
261 pub fn zone_state_update_interval(&self) -> Duration {
262 Duration::from_secs(self.zone_state_update_secs as u64)
263 }
264
265 pub fn zone_state_update_interval_secs(&self) -> u64 {
266 self.zone_state_update_secs as u64
267 }
268
269 pub fn check_zone_index(&self, z: usize) -> Outcome<()> {
270 let nz = self.num_zones as usize;
271 if z >= nz {
272 Err(err!(
273 "{} cannot be used to index a zone, with {} zone(s) at present.", z, nz;
274 Configuration))
275 } else {
276 Ok(())
277 }
278 }
279
280 pub fn write_config_file(&self, db_root: &Path) -> Outcome<()> {
281 let path = Self::config_path(db_root);
282 let mut file = res!(std::fs::File::create(&path));
283 let dat = Self::to_datmap(self.clone());
284 for mut line in dat.to_lines(" ", true) {
285 line.push_str("\n");
286 res!(file.write(line.as_bytes()));
287 }
288 debug!(sync_log::stream(), "O3db configuration written to {:?}: ", path);
289 for line in dat.to_lines(" ", true) {
290 debug!(sync_log::stream(), "{}", line);
291 }
292 Ok(())
293 }
294
295 // Path generation.
296
297 /// Ensures that the configured database container path is an absolute, comparable path.
298 pub fn canonicalize_path(pbuf: PathBuf) -> Outcome<String> {
299 let canonical = match (&pbuf).canonicalize() {
300 Ok(pbuf) => pbuf,
301 Err(e) => return Err(err!(e,
302 "The path {:?} does not exist and must be created.", pbuf;
303 Configuration, Invalid, Input)),
304 };
305 match canonical.into_os_string().into_string() {
306 Ok(s) => Ok(s),
307 _ => return Err(err!(
308 "Could not canonicalize {:?}, it may contain non-UTF-8 encoding.", pbuf;
309 Configuration, String, Conversion)),
310 }
311 }
312
313 pub fn zone_root(&self, container: &Path) -> PathBuf {
314 self.append_zone_root(PathBuf::from(container))
315 }
316
317 pub fn config_path(db_root: &Path) -> PathBuf {
318 let mut path = PathBuf::from(db_root);
319 path.push(constant::CONFIG_FILENAME);
320 path
321 }
322
323 pub fn append_zone_root(&self, mut path: PathBuf) -> PathBuf {
324 path.push(fmt!(format_zones_dir!(), self.num_zones));
325 path
326 }
327
328 // Chunking.
329
330 /// Return a rest chunk configuration from the database configuration.
331 pub fn chunk_config(&self) -> ChunkConfig {
332 ChunkConfig {
333 threshold_bytes: self.rest_chunk_threshold as usize,
334 chunk_size: self.rest_chunk_bytes as usize,
335 dat_wrap: true,
336 pad_last: true,
337 }
338 }
339
340 pub fn chunker(cfg: ChunkConfig) -> Chunker {
341 Chunker::default().set_config(cfg)
342 }
343
344 pub fn check_rest_chunk_config(&self, chunk_cfg: &ChunkConfig) -> Outcome<()> {
345 let max_file_size = self.data_file_max_bytes as f64;
346 let chunk_threshold = chunk_cfg.threshold_bytes as f64;
347 if chunk_threshold >= constant::MAX_FILE_TO_CHUNKING_THRESHOLD_RATIO * max_file_size {
348 return Err(err!(
349 "The size threshold before file keys and values are chunked, \
350 supplied in the configuration, of {} bytes, is too large compared \
351 to the specified maximum data file size of {} bytes. The ratio \
352 should not exceed {:.1}%.",
353 chunk_cfg.threshold_bytes, self.data_file_max_bytes,
354 constant::MAX_FILE_TO_CHUNKING_THRESHOLD_RATIO * 100.0;
355 TooBig, Configuration));
356 }
357 if chunk_cfg.chunk_size < constant::MIN_CHUNK_SIZE {
358 return Err(err!(
359 "Chunk size of {} is less than the current minimum of {}.",
360 chunk_cfg.chunk_size, constant::MIN_CHUNK_SIZE;
361 TooSmall, Configuration));
362 }
363 let chunk_size = chunk_cfg.chunk_size as f64;
364 if chunk_size * constant::MAX_FILE_TO_CHUNK_SIZE_RATIO >= max_file_size {
365 return Err(err!(
366 "The chunk size, supplied in the configuration, of {} bytes, is \
367 too large compared to the specified maximum data file size of {} \
368 bytes. The ratio should not exceed {:.1}%.",
369 chunk_cfg.chunk_size, self.data_file_max_bytes,
370 constant::MAX_FILE_TO_CHUNK_SIZE_RATIO * 100.0;
371 TooBig, Configuration));
372 }
373 Ok(())
374 }
375
376 pub fn check_file_size(&self) -> Outcome<()> {
377 if self.data_file_max_bytes > constant::MAX_FILE_BYTES {
378 return Err(err!(
379 "The configured maximum data file size in bytes of {} is too big, \
380 for safety it cannot exceed {}.",
381 self.data_file_max_bytes, constant::MAX_FILE_BYTES;
382 Invalid, Input));
383 }
384 Ok(())
385 }
386
387}
388
389#[derive(Clone, Debug, Default)]
390pub struct ZoneConfig {
391 pub ncbots: usize,
392 pub nfbots: usize,
393 pub cache_size_lim: usize,
394 pub init_load_caches: bool,
395}
396
397#[cfg(test)]
398mod tests {
399 use super::*;
400
401 #[test]
402 fn config_without_scan_bot_count_loads_with_one() -> Outcome<()> {
403 let mut map = match OzoneConfig::to_datmap(OzoneConfig::default()) {
404 Dat::Map(m) => m,
405 other => return Err(err!(
406 "Expected a Dat::Map from to_datmap, got {:?}.", other;
407 Test, Invalid)),
408 };
409 // Take the field out, leaving the file an older server would
410 // have written.
411 map.remove(&Dat::Str(fmt!("num_scbots_per_zone")));
412 if map.contains_key(&Dat::Str(fmt!("num_scbots_per_zone"))) {
413 return Err(err!(
414 "The scan bot count was still present after removal, so the \
415 test would prove nothing.";
416 Test, Bug));
417 }
418 let mut cfg = res!(OzoneConfig::from_datmap(map));
419 res!(cfg.check_and_fix());
420 if cfg.num_bots_per_zone(&WorkerType::Scan) != 1 {
421 return Err(err!(
422 "A config with no scan bot count loaded with {} scan bots per \
423 zone; a zone with none cannot answer a scan at all.",
424 cfg.num_bots_per_zone(&WorkerType::Scan);
425 Test, Mismatch));
426 }
427 Ok(())
428 }
429
430 #[test]
431 fn zero_scan_bot_count_is_raised_to_one() -> Outcome<()> {
432 let mut cfg = OzoneConfig::default();
433 cfg.num_scbots_per_zone = 0;
434 res!(cfg.check_and_fix());
435 if cfg.num_bots_per_zone(&WorkerType::Scan) != 1 {
436 return Err(err!(
437 "A zero scan bot count survived check_and_fix as {}.",
438 cfg.num_bots_per_zone(&WorkerType::Scan);
439 Test, Mismatch));
440 }
441 Ok(())
442 }
443}