Oregami
Repositories/oxedyne/fe2o3

oxedyne/fe2o3/fe2o3_o3db_sync/tests/perf.rs

10.3 KiB, 59 runs

created by r1870400018:840, 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

1// Be sure to run using release target, e.g.:
2// > clear;clear;cargo test -r --test perf -- --nocapture
3use oxedyne_fe2o3_core::{
4 prelude::*,
5 alt::Override,
6};
7use oxedyne_fe2o3_crypto::enc::EncryptionScheme;
8use oxedyne_fe2o3_hash::{
9 csum::ChecksumScheme,
10 hash::HashScheme,
11};
12use oxedyne_fe2o3_iop_db::api::{
13 Meta,
14 RestSchemesOverride,
15};
16use oxedyne_fe2o3_jdat::{
17 prelude::*,
18};
19use oxedyne_fe2o3_o3db_sync::{
20 data::core::RestSchemesInput,
21 test::{
22 data::{
23 find_unique,
24 },
25 dbapi::{
26 store,
27 fetch,
28 },
29 file::{
30 append_table,
31 save_multiple_files,
32 save_single_file,
33 },
34 setup,
35 },
36};
37use oxedyne_fe2o3_test::{
38 data::{
39 DataArrangement,
40 DataFill,
41 DataSize,
42 DataSpec,
43 },
44 error::delayed_error,
45};
46
47use std::{
48 path::{
49 Path,
50 PathBuf,
51 },
52 thread,
53 time::Duration,
54};
55
56use rand_core::{
57 RngCore,
58 OsRng,
59};
60
61
62fn generate_data(
63 keyspec: DataSpec,
64 valspec: DataSpec,
65)
66 -> Outcome<(Vec<Dat>, Vec<Dat>, usize, usize)>
67{
68 let n = res!(keyspec.len());
69
70 test!(sync_log::stream(), "Generating {} data pairs using:", n);
71 test!(sync_log::stream(), " keyspec: {:?}", keyspec);
72 test!(sync_log::stream(), " valspec: {:?}", valspec);
73 let k = res!(keyspec.generate());
74 let v = res!(valspec.generate());
75 //let n = k.len();
76 let mut kmaxlen = 0;
77 let mut kminlen = usize::MAX;
78 let mut vmaxlen = 0;
79 let mut vminlen = usize::MAX;
80 let mut kbyts = 0;
81 let mut vbyts = 0;
82 let mut kdats = Vec::new();
83 let mut vdats = Vec::new();
84 for i in 0..k.len() {
85 let x = &k[i];
86 if x.len() > kmaxlen { kmaxlen = x.len() }
87 if x.len() < kminlen { kminlen = x.len() }
88 kbyts += x.len();
89 kdats.push(Dat::wrap_dat(k[i].clone()));
90
91 let x = &v[i];
92 if x.len() > vmaxlen { vmaxlen = x.len() }
93 if x.len() < vminlen { vminlen = x.len() }
94 vbyts += x.len();
95 vdats.push(Dat::wrap_dat(v[i].clone()));
96 }
97 test!(sync_log::stream(), "Completed generation of {} pairs with metrics:", n);
98 test!(sync_log::stream(), " Key sizes: mean {} range {}..{}", kbyts / n, kminlen, kmaxlen);
99 test!(sync_log::stream(), " Val sizes: mean {} range {}..{}", vbyts / n, vminlen, vmaxlen);
100 Ok((kdats, vdats, kbyts, vbyts))
101}
102
103pub fn test_perf(_filter: &'static str) -> Outcome<()> {
104
105 let db_root = res!(Path::new("./test_db").canonicalize());
106 let mut enckey = [0u8; 32];
107 OsRng.fill_bytes(&mut enckey);
108 let aes_gcm = res!(EncryptionScheme::new_aes_256_gcm_with_key(&enckey[..]));
109 let sha3_256 = HashScheme::new_sha3_256();
110 let crc32 = ChecksumScheme::new_crc32();
111 let mut schms2: RestSchemesOverride<EncryptionScheme, HashScheme>;
112 let schms_input = RestSchemesInput::new(
113 Some(aes_gcm.clone()),
114 Some(sha3_256.clone()),
115 None::<HashScheme>,
116 Some(crc32.clone()),
117 );
118 let user = setup::Uid::default();
119 //let meta = Meta::<{ setup::UID_LEN }, setup::Uid>::new(setup::Uid::default());
120 let mut table = Vec::new();
121
122 // Generate data.
123 let keyspec = DataSpec {
124 size: DataSize::RandUniform { lo: 30, hi: 3000 },
125 fill: DataFill::Random,
126 arr: DataArrangement::RepeatFillAndSeq{
127 n: 60,
128 rep: 100,
129 specbox: Box::new(
130 DataSpec {
131 size: DataSize::RandUniform { lo: 30, hi: 3000 },
132 fill: DataFill::Random,
133 arr: DataArrangement::PlainFill(15),
134 }
135 ),
136 },
137 };
138 let valspec = DataSpec {
139 size: DataSize::RandUniform { lo: 30, hi: 10_000 },
140 fill: DataFill::Random,
141 arr: DataArrangement::PlainFill(res!(keyspec.len())),
142 };
143 let (kdats, vdats, kbyts, vbyts) = res!(generate_data(keyspec, valspec));
144 let mut mask_opt = None;
145
146 let error_delay = 2;
147 let data_file_max_bytes: u64 = 1_000_000;
148 let bytes_before_chunking: u64 = 100_000;
149
150 {
151 for case in &[
152 0,
153 100,
154 200,
155 ] {
156
157 let (note, cfg) = match case {
158 0 => {
159 schms2 = RestSchemesOverride::default()
160 .set_encrypter(Override::Default(aes_gcm.clone()));
161 let mut cfg = res!(setup::default_cfg());
162 cfg.data_file_max_bytes = data_file_max_bytes;
163 cfg.rest_chunk_threshold = bytes_before_chunking;
164 (
165 fmt!("The kitchen sink, everything on except gc."),
166 cfg,
167 )
168 },
169 100 => {
170 schms2 = RestSchemesOverride::default();
171 let mut cfg = res!(setup::default_cfg());
172 cfg.data_file_max_bytes = data_file_max_bytes;
173 cfg.rest_chunk_threshold = bytes_before_chunking;
174 (
175 fmt!("Value encryption off."),
176 cfg,
177 )
178 },
179 200 => {
180 schms2 = RestSchemesOverride::default();
181 let mut cfg = res!(setup::default_cfg());
182 cfg.data_file_max_bytes = data_file_max_bytes;
183 cfg.rest_chunk_threshold = bytes_before_chunking;
184 (
185 fmt!("Value encryption and key hashing off."),
186 cfg,
187 )
188 },
189 _ => unimplemented!(),
190 };
191
192 test!(sync_log::stream(), "+---------------------------------------------+");
193 test!(sync_log::stream(), "| NEW OZONE SESSION |");
194 test!(sync_log::stream(), "| Wipe all traces of previous test. |");
195 test!(sync_log::stream(), "| Start database. |");
196 test!(sync_log::stream(), "| No gc and no user filtering. |");
197 test!(sync_log::stream(), "| Store and fetch a standard data set. |");
198 test!(sync_log::stream(), "| Gracefully shut down the database. |");
199 test!(sync_log::stream(), "+---------------------------------------------+");
200 let mut db = match setup::start_db(
201 db_root.clone(),
202 Some(cfg.clone()),
203 schms_input.clone(),
204 None,
205 false,
206 true,
207 ) {
208 // These pauses on errors are needed to capture tardy messages from asynchronous
209 // logging.
210 Err(e) => return Err(delayed_error(e, error_delay)),
211 Ok(db) => db,
212 };
213
214 // Can use logger now.
215
216 if mask_opt.is_none() {
217 test!(sync_log::stream(), "Creating mask for unique keys...");
218 let mut kbufs = Vec::new();
219 for kdat in &kdats {
220 let (kbuf, _, _) = res!(db.api().ozone_key_dat(&kdat, Some(&schms2)));
221 kbufs.push(kbuf);
222 }
223 mask_opt = Some(find_unique(&kbufs));
224 test!(sync_log::stream(), " mask completed.");
225 }
226
227 thread::sleep(Duration::from_secs(1));
228 test!(sync_log::stream(), "Begin...");
229
230 if *case == 0 {
231
232 let mut path = db_root.clone();
233 path.push("control_single_file");
234 test!(sync_log::stream(), "Creating {:?}", path);
235 res!(std::fs::create_dir(&path));
236 match save_single_file(
237 path,
238 &mut db,
239 user,
240 Some(&schms2),
241 kdats.clone(),
242 vdats.clone(),
243 kbyts + vbyts,
244 ) {
245 Err(e) => return Err(delayed_error(e, error_delay)),
246 Ok((tps, bw)) => table.push((fmt!("Save directly to single file"), tps, bw)),
247 }
248
249 thread::sleep(Duration::from_secs(3));
250
251 let mut path = db_root.clone();
252 path.push("control_multiple_files");
253 test!(sync_log::stream(), "Creating {:?}", path);
254 res!(std::fs::create_dir(&path));
255 match save_multiple_files(
256 path,
257 res!(cfg.data_file_max_bytes.try_into()),
258 &mut db,
259 user,
260 Some(&schms2),
261 kdats.clone(),
262 vdats.clone(),
263 kbyts + vbyts,
264 ) {
265 Err(e) => return Err(delayed_error(e, error_delay)),
266 Ok((tps, bw)) => table.push((fmt!("Save directly to multiple files"), tps, bw)),
267 }
268
269 }
270
271 test!(sync_log::stream(), "Test {} {}", case, note);
272 thread::sleep(Duration::from_secs(3));
273
274 match store(
275 &mut db,
276 user,
277 Some(&schms2),
278 kdats.clone(),
279 vdats.clone(),
280 kbyts + vbyts,
281 ) {
282 Err(e) => return Err(delayed_error(e, error_delay)),
283 Ok((tps, bw)) => table.push((fmt!("Store: {}", note), tps, bw)),
284 }
285
286 thread::sleep(Duration::from_secs(3));
287
288 if let Some(mask) = mask_opt.as_ref() {
289 match fetch(
290 &mut db,
291 Some(&schms2),
292 &kdats,
293 &mask,
294 &vdats,
295 kbyts + vbyts,
296 ) {
297 Err(e) => return Err(delayed_error(e, error_delay)),
298 Ok(result) => {
299 for (s, tps, bw) in result {
300 table.push((fmt!("Fetch {}: {}", s, note), tps, bw));
301 }
302 },
303 }
304 }
305
306 thread::sleep(Duration::from_secs(3));
307
308 test!(sync_log::stream(), "Shutting db down...");
309 res!(db.shutdown());
310 }
311 }
312
313 let path = PathBuf::from("./perf_record.txt");
314 res!(append_table(path, table));
315
316 Ok(())
317}