daqbuf/daqingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Change debug out, test case

Browse Source
This commit is contained in:
Dominik Werder
2025-05-29 06:41:23 +02:00
parent f16353f21f
commit 1ddf18276f
4 changed files with 160 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
log/src/log.rs
+25 -5
View File
@@ -3,12 +3,13 @@
// pub use tracing::error;
// pub use tracing::info;
// pub use tracing::trace;
pub use tracing::warn;
// pub use tracing::warn;
pub use direct_debug as debug;
pub use direct_error as error;
pub use direct_info as info;
pub use direct_trace as trace;
pub use direct_warn as warn;
pub mod log_macros_direct {
#[allow(unused)]
@@ -25,20 +26,24 @@ pub mod log_macros_direct {
#[macro_export]
macro_rules! direct_debug {
($fmt:expr) => {
eprintln!(concat!("DEBUG ", $fmt));
// eprintln!(concat!("DEBUG ", $fmt));
// eprintln!("DEBUG {}", format_args!($fmt));
eprintln!("{}", format_args!(concat!("DEBUG ", $fmt)));
};
($fmt:expr, $($arg:expr),*) => {
eprintln!(concat!("DEBUG ", $fmt), $($arg),*);
// eprintln!(concat!("DEBUG ", $fmt), $($arg),*);
// eprintln!("DEBUG {}", format_args!($fmt, $($arg),*));
eprintln!("{}", format_args!(concat!("DEBUG ", $fmt), $($arg),*));
};
}
#[allow(unused)]
#[macro_export]
macro_rules! direct_info {
($fmt:expr) => {
eprintln!(concat!("INFO ", $fmt));
eprintln!("INFO {}", format_args!($fmt));
};
($fmt:expr, $($arg:expr),*) => {
eprintln!(concat!("INFO ", $fmt), $($arg),*);
eprintln!("INFO {}", format_args!($fmt, $($arg),*));
};
}
#[allow(unused)]
@@ -62,3 +67,18 @@ pub mod log_macros_direct {
};
}
}
#[allow(unused)]
#[macro_export]
macro_rules! log_v2_trace {
// ($fmt:expr) => {
// let h = format_args!();
// eprintln!(concat!("TRACE V2 ", $fmt));
// };
($fmt:expr, $($arg:expr),*) => {
// let fmt2 = concat!("", $fmt);
// let fmt2 = concat!("TRACE V2 ", $fmt, $($arg),*);
// let h = format_args!($fmt, $($arg),*);
// eprintln!("h: {:?}", h);
};
}
netfetch/src/ca/findioc.rs
+1 -1
View File
@@ -658,7 +658,7 @@ impl Stream for FindIocStream {
// Already answered from another target
//trace!("bid {bid:?} from batch send queue not in flight AND all done");
} else {
warn!("bid {bid:?} from batch send queue not in flight NOT done");
warn!("bid {:?} from batch send queue not in flight NOT done", bid);
}
have_progress = true;
}
serieswriter/src/binwriter.rs
+13 -27
View File
@@ -35,6 +35,8 @@ macro_rules! trace_tick_verbose { ($t:expr, $($arg:expr),*) => ( if false { if $
macro_rules! trace_bin { ($t:expr, $($arg:expr),*) => ( if true { if $t { log::info!($($arg),*); } } ) }
macro_rules! debug_rebin_ingest { ($t:expr, $($arg:expr),*) => ( if true { if $t { log::debug!($($arg),*); } } ) }
autoerr::create_error_v1!(
name(Error, "SerieswriterBinwriter"),
enum variants {
@@ -161,7 +163,7 @@ impl BinWriter {
const DUR_MAX: DtMs = DtMs::from_ms_u64(1000 * 60 * 60 * 24 * 123);
let rts = [RetentionTime::Short, RetentionTime::Medium, RetentionTime::Long];
let quiets = [min_quiets.st.clone(), min_quiets.mt.clone(), min_quiets.lt.clone()];
let cnt_zero_disable = WriteCntZero::Disable;
let cnt_zero_mode = WriteCntZero::Enable;
let mut binner_1st = None;
let mut binner_others = Vec::new();
let mut has_monitor = None;
@@ -174,7 +176,7 @@ impl BinWriter {
}
})
.filter(|x| x.1 > DUR_ZERO && x.1 < DUR_MAX)
.map(|x| (x.0, bin_len_clamp(x.1), cnt_zero_disable.clone()))
.map(|x| (x.0, bin_len_clamp(x.1), cnt_zero_mode.clone()))
.collect();
let has_monitor = has_monitor;
debug_init!(trd, "has_monitor {:?} is_polled {:?}", has_monitor, is_polled);
@@ -190,39 +192,27 @@ impl BinWriter {
combs.push((RetentionTime::Long, PrebinnedPartitioning::Day1, WriteCntZero::Enable));
}
_ => {
combs.push((
RetentionTime::Long,
PrebinnedPartitioning::Hour1,
cnt_zero_disable.clone(),
));
combs.push((RetentionTime::Long, PrebinnedPartitioning::Hour1, cnt_zero_mode.clone()));
combs.push((RetentionTime::Long, PrebinnedPartitioning::Day1, WriteCntZero::Enable));
}
}
} else {
match &has_monitor {
Some(RetentionTime::Short) => {
combs.push((
RetentionTime::Short,
PrebinnedPartitioning::Min1,
cnt_zero_disable.clone(),
));
combs.push((RetentionTime::Short, PrebinnedPartitioning::Min1, cnt_zero_mode.clone()));
combs.push((
RetentionTime::Medium,
PrebinnedPartitioning::Hour1,
cnt_zero_disable.clone(),
cnt_zero_mode.clone(),
));
combs.push((RetentionTime::Long, PrebinnedPartitioning::Day1, WriteCntZero::Enable));
}
Some(RetentionTime::Medium) => {
combs.push((
RetentionTime::Short,
PrebinnedPartitioning::Min1,
cnt_zero_disable.clone(),
));
combs.push((RetentionTime::Short, PrebinnedPartitioning::Min1, cnt_zero_mode.clone()));
combs.push((
RetentionTime::Medium,
PrebinnedPartitioning::Hour1,
cnt_zero_disable.clone(),
cnt_zero_mode.clone(),
));
combs.push((RetentionTime::Long, PrebinnedPartitioning::Day1, WriteCntZero::Enable));
}
@@ -230,13 +220,9 @@ impl BinWriter {
combs.push((
RetentionTime::Medium,
PrebinnedPartitioning::Min1,
cnt_zero_disable.clone(),
));
combs.push((
RetentionTime::Long,
PrebinnedPartitioning::Hour1,
cnt_zero_disable.clone(),
cnt_zero_mode.clone(),
));
combs.push((RetentionTime::Long, PrebinnedPartitioning::Hour1, cnt_zero_mode.clone()));
combs.push((RetentionTime::Long, PrebinnedPartitioning::Day1, WriteCntZero::Enable));
}
None => {
@@ -407,6 +393,7 @@ impl BinWriter {
let mut bins2: BinsBoxed = Box::new(bins);
for i in 0..self.binner_others.len() {
let st = &mut self.binner_others[i];
debug_rebin_ingest!(self.trd, "rt {:?} len {}", st.rt, bins2.len());
st.binner.ingest(&bins2)?;
let bb: Option<BinsBoxed> = st.binner.output()?;
match bb {
@@ -478,7 +465,6 @@ impl BinWriter {
}
let bins_len = bins.len();
for (ts1, ts2, cnt, min, max, avg, lst, fnl) in bins.zip_iter_2() {
eprintln!("cnt {}", cnt);
info!("cnt {}", cnt);
let bin_len = DtMs::from_ms_u64(ts2.delta(ts1).ms_u64());
if fnl == false {
@@ -494,7 +480,7 @@ impl BinWriter {
let e = Error::UnexpectedBinLen(bin_len, pbp);
return Err(e);
}
if *discard_front < 1 {
if false && *discard_front < 1 {
*discard_front += 1;
debug_bin!(trd, "handle_output_ready discard_front {:?}", rt);
} else {
serieswriter/src/binwriter/test.rs
+121 -10
View File
@@ -1,18 +1,90 @@
use super::BinWriter;
use crate::rtwriter::MinQuiets;
use log::*;
use netpod::DtMs;
use netpod::ScalarType;
use netpod::Shape;
use netpod::TsMs;
use netpod::TsNano;
use scywr::insertqueues::InsertDeques;
use scywr::iteminsertqueue::QueryItem;
use scywr::iteminsertqueue::TimeBinSimpleF32V02;
use series::ChannelStatusSeriesId;
use series::SeriesId;
use series::msp::PrebinnedPartitioning;
use std::collections::VecDeque;
const T0: TsMs = TsMs::from_ms_u64(1000 * 60 * 60 * 36);
macro_rules! debug_item {
($fmt:expr) => {
eprintln!("{0} {0:?}", format_args!($fmt));
};
($fmt:expr, $arg1:expr) => {
// let h = format_args!($fmt, $arg1);
eprintln!("h: {:?}", format_args!($fmt, $arg1));
};
($fmt:expr, $($args:expr),*) => {
// let h = format_args!($fmt, $arg1);
// eprintln!("h: {:?}", format_args!($fmt, $arg1));
eprintln!("h: {:?}", format_args!($fmt, $($args),*));
};
}
fn def<T: Default>() -> T {
Default::default()
}
fn sec(sec: f32) -> TsNano {
TsNano::from_ms((1e3 * sec) as u64)
T0.add_dt_ms(DtMs::from_ms_u64((1e3 * sec) as u64)).ns()
// TsNano::from_ms((1e3 * sec) as u64)
}
fn collect_bin_write_only(qu: &VecDeque<QueryItem>) -> VecDeque<TimeBinSimpleF32V02> {
// TODO
// take the scylla insert items and store them in hashmap according to binlen
// in a deque for more easy comparison later.
// let mut ret = def();
let mut ret = VecDeque::new();
for item in qu {
match item {
QueryItem::TimeBinSimpleF32V02(bin) => {
ret.push_back(bin.clone());
}
_ => {}
}
}
ret
}
fn print_binscol(rt: &str, binscol: &VecDeque<TimeBinSimpleF32V02>) {
for bin in binscol {
debug!(
"{} bl {:5} msp {:4} off {:4} cnt {:3} min {:7.2} max {:7.2} lst {:7.2}",
rt,
// min = bin.min,
bin.binlen / 1000,
bin.msp,
bin.off,
bin.cnt,
bin.min,
bin.max,
bin.lst
);
}
}
#[test]
fn binwriter_nest01_00() {
let _ = PrebinnedPartitioning::Day1;
let _ = format_args!("");
let _ = format_args!(concat!("", ""));
debug_item!("plain fmt string");
debug_item!("plain fmt string with comma");
debug_item!("param as param: {:?}", "some--string123");
let some_string = "some--string123";
debug_item!("param interpolated: {some_string:?}");
debug_item!("param interpolated: {some_string}");
let beg = TsNano::from_ms(1000 * 40);
let min_quiets = MinQuiets::test_1_10_60();
let is_polled = false;
@@ -20,24 +92,63 @@ fn binwriter_nest01_00() {
let sid = SeriesId::new(51);
let scalar_type = ScalarType::F32;
let shape = Shape::Scalar;
let chname2 = String::from("DUMMY");
let chname2 = String::from("daqbuftest");
let mut iqdqs = InsertDeques::new();
let mut binwriter = BinWriter::new(beg, min_quiets, is_polled, cssid, sid, scalar_type, shape, chname2).unwrap();
binwriter.ingest(sec(39.9), 2., &mut iqdqs).unwrap();
binwriter.ingest(sec(39.9), 2.2, &mut iqdqs).unwrap();
binwriter.ingest(sec(40.0), 2., &mut iqdqs).unwrap();
binwriter.ingest(sec(40.1), 2., &mut iqdqs).unwrap();
binwriter.ingest(sec(50.0), 2., &mut iqdqs).unwrap();
binwriter.ingest(sec(50.1), 2., &mut iqdqs).unwrap();
binwriter.ingest(sec(51.0), 2., &mut iqdqs).unwrap();
binwriter.ingest(sec(51.1), 2., &mut iqdqs).unwrap();
binwriter.ingest(sec(60.0), 2., &mut iqdqs).unwrap();
// binwriter.ingest(sec(70.0), 2., &mut iqdqs).unwrap();
// binwriter.ingest(sec(70.1), 2., &mut iqdqs).unwrap();
binwriter.ingest(sec(80.0), 2., &mut iqdqs).unwrap();
binwriter.tick(&mut iqdqs).unwrap();
eprintln!("iqdqs summary {}", iqdqs.summary());
for x in iqdqs.st_rf3_qu {
eprintln!("ST {:?}", x);
debug!("iqdqs summary {}", iqdqs.summary());
for x in &iqdqs.st_rf3_qu {
debug!("ST {:?}", x);
}
for x in iqdqs.mt_rf3_qu {
eprintln!("MT {:?}", x);
for x in &iqdqs.mt_rf3_qu {
debug!("MT {:?}", x);
}
for x in iqdqs.lt_rf3_qu {
eprintln!("LT {:?}", x);
for x in &iqdqs.lt_rf3_qu {
debug!("LT {:?}", x);
}
{
let rt = "ST";
let binscol = collect_bin_write_only(&iqdqs.st_rf3_qu);
print_binscol(rt, &binscol);
let pbp = PrebinnedPartitioning::Sec1;
let x = pbp.msp_lsp(T0);
let msp_exp = T0.ms() / pbp.bin_len().ms() / pbp.patch_len() as u64;
assert_eq!(x.0 as u64, msp_exp);
let x = pbp.msp_lsp(sec(39.9).to_ts_ms());
assert_eq!(binscol[0].msp as u32, x.0);
assert_eq!(binscol[0].off as u32, x.1);
let x = pbp.msp_lsp(sec(40.0).to_ts_ms());
assert_eq!(binscol[1].msp as u32, x.0);
assert_eq!(binscol[1].off as u32, x.1);
assert_eq!(binscol[1].max, 2.0);
}
{
let rt = "MT";
let binscol = collect_bin_write_only(&iqdqs.mt_rf3_qu);
print_binscol(rt, &binscol);
}
{
let rt = "LT";
let binscol = collect_bin_write_only(&iqdqs.lt_rf3_qu);
print_binscol(rt, &binscol);
}
{
let pbp = PrebinnedPartitioning::Sec10;
debug!(
"expect MT msp at T0 {}",
T0.ms() / pbp.bin_len().ms() / pbp.patch_len() as u64
);
}
log_v2_trace!("ARG-1 {}", 42);
}