From 1ddf18276f67c0ae8acbb0c1b9f346e9349c37a8 Mon Sep 17 00:00:00 2001 From: Dominik Werder Date: Thu, 29 May 2025 06:41:23 +0200 Subject: [PATCH] Change debug out, test case --- log/src/log.rs | 30 +++++-- netfetch/src/ca/findioc.rs | 2 +- serieswriter/src/binwriter.rs | 40 +++------ serieswriter/src/binwriter/test.rs | 131 ++++++++++++++++++++++++++--- 4 files changed, 160 insertions(+), 43 deletions(-) diff --git a/log/src/log.rs b/log/src/log.rs index bf04f14..7c97b69 100644 --- a/log/src/log.rs +++ b/log/src/log.rs @@ -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); + }; +} diff --git a/netfetch/src/ca/findioc.rs b/netfetch/src/ca/findioc.rs index fa926c9..315dcc2 100644 --- a/netfetch/src/ca/findioc.rs +++ b/netfetch/src/ca/findioc.rs @@ -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; } diff --git a/serieswriter/src/binwriter.rs b/serieswriter/src/binwriter.rs index eb81ba1..012479b 100644 --- a/serieswriter/src/binwriter.rs +++ b/serieswriter/src/binwriter.rs @@ -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 = 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 { diff --git a/serieswriter/src/binwriter/test.rs b/serieswriter/src/binwriter/test.rs index e26d928..1d26d33 100644 --- a/serieswriter/src/binwriter/test.rs +++ b/serieswriter/src/binwriter/test.rs @@ -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::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) -> VecDeque { + // 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) { + 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); }