daqbuf/daqbuffer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP

Browse Source
This commit is contained in:
Dominik Werder
2024-10-24 09:55:31 +02:00
parent 9a35e2f4fa
commit 1fd41ed513
13 changed files with 183 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/disk/src/binned/prebinned.rs
View File

@@ -251,7 +251,7 @@ pub async fn pre_binned_bytes_for_http(
node_config,
)
.await?
.map(|item| match item.make_frame() {
.map(|item| match item.make_frame________() {
Ok(item) => Ok(item.freeze()),
Err(e) => Err(e),
});

46
crates/items_2/src/binning/aggregator.rs
View File

@@ -161,4 +161,48 @@ impl AggregatorTimeWeight<u64> for AggregatorNumeric {
}
}
// TODO do enum right from begin, using a SOA enum container.
impl AggregatorTimeWeight<bool> for AggregatorNumeric {
fn new() -> Self {
Self { sum: 0. }
}
fn ingest(&mut self, dt: DtNano, bl: DtNano, val: bool) {
let f = dt.ns() as f64 / bl.ns() as f64;
trace!("INGEST {} {}", f, val);
self.sum += f * val as u8 as f64;
}
fn reset_for_new_bin(&mut self) {
self.sum = 0.;
}
fn result_and_reset_for_new_bin(&mut self, filled_width_fraction: f32) -> f64 {
let sum = self.sum.clone();
trace!("result_and_reset_for_new_bin sum {} {}", sum, filled_width_fraction);
self.sum = 0.;
sum / filled_width_fraction as f64
}
}
impl AggregatorTimeWeight<String> for AggregatorNumeric {
fn new() -> Self {
Self { sum: 0. }
}
fn ingest(&mut self, dt: DtNano, bl: DtNano, val: String) {
let f = dt.ns() as f64 / bl.ns() as f64;
trace!("INGEST {} {}", f, val);
self.sum += f * val.len() as f64;
}
fn reset_for_new_bin(&mut self) {
self.sum = 0.;
}
fn result_and_reset_for_new_bin(&mut self, filled_width_fraction: f32) -> f64 {
let sum = self.sum.clone();
trace!("result_and_reset_for_new_bin sum {} {}", sum, filled_width_fraction);
self.sum = 0.;
sum / filled_width_fraction as f64
}
}

77
crates/items_2/src/binning/container_bins.rs
View File

@@ -10,6 +10,7 @@ use items_0::timebin::BinsBoxed;
use items_0::vecpreview::VecPreview;
use items_0::AsAnyMut;
use items_0::WithLen;
use netpod::EnumVariant;
use netpod::TsNano;
use serde::Deserialize;
use serde::Serialize;
@@ -339,6 +340,21 @@ where
}
}
macro_rules! try_to_old_time_binned {
($sty:ty, $this:expr, $lst:expr) => {
let this = $this;
let a = this as &dyn any::Any;
if let Some(src) = a.downcast_ref::<ContainerBins<$sty>>() {
use items_0::Empty;
let mut ret = crate::binsdim0::BinsDim0::<$sty>::empty();
for ((((((&ts1, &ts2), &cnt), min), max), avg), fnl) in src.zip_iter() {
ret.push(ts1.ns(), ts2.ns(), cnt, *min, *max, *avg as f32, $lst);
}
return Box::new(ret);
}
};
}
impl<EVT> BinningggContainerBinsDyn for ContainerBins<EVT>
where
EVT: EventValueType,
@@ -372,25 +388,56 @@ where
}
fn to_old_time_binned(&self) -> Box<dyn items_0::timebin::TimeBinned> {
try_to_old_time_binned!(u8, self, 0);
try_to_old_time_binned!(u16, self, 0);
try_to_old_time_binned!(u32, self, 0);
try_to_old_time_binned!(u64, self, 0);
try_to_old_time_binned!(i8, self, 0);
try_to_old_time_binned!(i16, self, 0);
try_to_old_time_binned!(i32, self, 0);
try_to_old_time_binned!(i64, self, 0);
try_to_old_time_binned!(f32, self, 0.);
try_to_old_time_binned!(f64, self, 0.);
try_to_old_time_binned!(bool, self, false);
// try_to_old_time_binned!(String, self, String::new());
let a = self as &dyn any::Any;
if let Some(src) = a.downcast_ref::<ContainerBins<f64>>() {
if let Some(src) = a.downcast_ref::<ContainerBins<EnumVariant>>() {
use items_0::Empty;
let mut ret = crate::binsdim0::BinsDim0::<f64>::empty();
for ((((((&ts1, &ts2), &cnt), min), max), avg), fnl) in src.zip_iter() {
ret.push(ts1.ns(), ts2.ns(), cnt, *min, *max, *avg as f32, 0.);
let mut ret = crate::binsdim0::BinsDim0::<EnumVariant>::empty();
for ((((((&ts1, &ts2), &cnt), min), max), avg), _fnl) in src.zip_iter() {
ret.push(
ts1.ns(),
ts2.ns(),
cnt,
min.clone(),
max.clone(),
*avg as f32,
EnumVariant::new(0, ""),
);
}
Box::new(ret)
} else if let Some(src) = a.downcast_ref::<ContainerBins<f32>>() {
use items_0::Empty;
let mut ret = crate::binsdim0::BinsDim0::<f32>::empty();
for ((((((&ts1, &ts2), &cnt), min), max), avg), fnl) in src.zip_iter() {
ret.push(ts1.ns(), ts2.ns(), cnt, *min, *max, *avg as f32, 0.);
}
Box::new(ret)
} else {
let styn = any::type_name::<EVT>();
todo!("TODO impl for {styn}")
return Box::new(ret);
}
let styn = any::type_name::<EVT>();
todo!("TODO impl for {styn}");
// let a = self as &dyn any::Any;
// if let Some(src) = a.downcast_ref::<ContainerBins<f64>>() {
// use items_0::Empty;
// let mut ret = crate::binsdim0::BinsDim0::<f64>::empty();
// for ((((((&ts1, &ts2), &cnt), min), max), avg), fnl) in src.zip_iter() {
// ret.push(ts1.ns(), ts2.ns(), cnt, *min, *max, *avg as f32, 0.);
// }
// Box::new(ret)
// } else if let Some(src) = a.downcast_ref::<ContainerBins<f32>>() {
// use items_0::Empty;
// let mut ret = crate::binsdim0::BinsDim0::<f32>::empty();
// for ((((((&ts1, &ts2), &cnt), min), max), avg), fnl) in src.zip_iter() {
// ret.push(ts1.ns(), ts2.ns(), cnt, *min, *max, *avg as f32, 0.);
// }
// Box::new(ret)
// } else {
// let styn = any::type_name::<EVT>();
// todo!("TODO impl for {styn}")
// }
}
}

76
crates/items_2/src/binning/container_events.rs
View File

@@ -9,6 +9,7 @@ use err::ThisError;
use items_0::timebin::BinningggContainerEventsDyn;
use items_0::vecpreview::PreviewRange;
use items_0::vecpreview::VecPreview;
use items_0::AsAnyRef;
use netpod::BinnedRange;
use netpod::TsNano;
use serde::Deserialize;
@@ -35,7 +36,7 @@ pub trait EventValueType: fmt::Debug + Clone + PartialOrd + Send + 'static {
type AggregatorTimeWeight: AggregatorTimeWeight<Self>;
type AggTimeWeightOutputAvg: AggTimeWeightOutputAvg;
fn identity_sum() -> Self;
// fn identity_sum() -> Self;
// fn add_weighted(&self, add: &Self, f: f32) -> Self;
}
@@ -57,48 +58,48 @@ where
}
macro_rules! impl_event_value_type {
($evt:ty, $zero:expr) => {
($evt:ty) => {
impl EventValueType for $evt {
type Container = VecDeque<Self>;
type AggregatorTimeWeight = AggregatorNumeric;
type AggTimeWeightOutputAvg = f64;
fn identity_sum() -> Self {
$zero
}
}
};
}
impl_event_value_type!(u8, 0);
impl_event_value_type!(u16, 0);
impl_event_value_type!(u32, 0);
impl_event_value_type!(u64, 0);
impl_event_value_type!(i8, 0);
impl_event_value_type!(i16, 0);
impl_event_value_type!(i32, 0);
impl_event_value_type!(i64, 0);
// impl_event_value_type!(f32, 0.);
// impl_event_value_type!(f64, 0.);
impl_event_value_type!(u8);
impl_event_value_type!(u16);
impl_event_value_type!(u32);
impl_event_value_type!(u64);
impl_event_value_type!(i8);
impl_event_value_type!(i16);
impl_event_value_type!(i32);
impl_event_value_type!(i64);
// impl_event_value_type!(f32);
// impl_event_value_type!(f64);
impl EventValueType for f32 {
type Container = VecDeque<Self>;
type AggregatorTimeWeight = AggregatorNumeric;
type AggTimeWeightOutputAvg = f32;
fn identity_sum() -> Self {
0.
}
}
impl EventValueType for f64 {
type Container = VecDeque<Self>;
type AggregatorTimeWeight = AggregatorNumeric;
type AggTimeWeightOutputAvg = f64;
}
fn identity_sum() -> Self {
0.
}
impl EventValueType for bool {
type Container = VecDeque<Self>;
type AggregatorTimeWeight = AggregatorNumeric;
type AggTimeWeightOutputAvg = f64;
}
impl EventValueType for String {
type Container = VecDeque<Self>;
type AggregatorTimeWeight = AggregatorNumeric;
type AggTimeWeightOutputAvg = f64;
}
#[derive(Debug, Clone)]
@@ -122,6 +123,20 @@ where
vals: <EVT as EventValueType>::Container,
}
macro_rules! try_to_events_dim0 {
($sty:ty, $this:expr) => {
let this = $this;
if let Some(evs) = this.as_any_ref().downcast_ref::<ContainerEvents<$sty>>() {
use crate::eventsdim0::EventsDim0;
let tss: VecDeque<_> = this.tss.iter().map(|x| x.ns()).collect();
let pulses = tss.iter().map(|_| 0).collect();
let values = evs.vals.clone();
let ret = EventsDim0::<$sty> { tss, pulses, values };
return Box::new(ret);
}
};
}
impl<EVT> ContainerEvents<EVT>
where
EVT: EventValueType,
@@ -178,6 +193,12 @@ where
self.tss.push_back(ts);
self.vals.push_back(val);
}
pub fn to_events_dim0(&self) -> Box<dyn items_0::Events> {
try_to_events_dim0!(f64, self);
let styn = any::type_name::<EVT>();
todo!("TODO to_container_events for {styn}")
}
}
impl<EVT> fmt::Debug for ContainerEvents<EVT>
@@ -196,6 +217,15 @@ where
}
}
impl<EVT> AsAnyRef for ContainerEvents<EVT>
where
EVT: EventValueType,
{
fn as_any_ref(&self) -> &dyn any::Any {
self
}
}
pub struct ContainerEventsTakeUpTo<'a, EVT>
where
EVT: EventValueType,

9
crates/items_2/src/binning/valuetype.rs
View File

@@ -65,7 +65,7 @@ impl AggregatorTimeWeight<EnumVariant> for EnumVariantAggregatorTimeWeight {
}
fn reset_for_new_bin(&mut self) {
self.sum = f32::identity_sum();
self.sum = 0.;
}
fn result_and_reset_for_new_bin(
@@ -73,7 +73,7 @@ impl AggregatorTimeWeight<EnumVariant> for EnumVariantAggregatorTimeWeight {
filled_width_fraction: f32,
) -> <EnumVariant as EventValueType>::AggTimeWeightOutputAvg {
let ret = self.sum.clone();
self.sum = f32::identity_sum();
self.sum = 0.;
ret / filled_width_fraction
}
}
@@ -82,9 +82,4 @@ impl EventValueType for EnumVariant {
type Container = EnumVariantContainer;
type AggregatorTimeWeight = EnumVariantAggregatorTimeWeight;
type AggTimeWeightOutputAvg = f32;
// TODO remove this from trait, only needed for common numeric cases but not in general.
fn identity_sum() -> Self {
todo!()
}
}

26
crates/items_2/src/eventsdim0.rs
View File

@@ -1128,8 +1128,8 @@ impl<STY: ScalarOps> Events for EventsDim0<STY> {
try_to_container_events!(i64, self);
try_to_container_events!(f32, self);
try_to_container_events!(f64, self);
// try_to_container_events!(bool, self);
// try_to_container_events!(String, self);
try_to_container_events!(bool, self);
try_to_container_events!(String, self);
let this = self;
if let Some(evs) = self.as_any_ref().downcast_ref::<EventsDim0<netpod::EnumVariant>>() {
use crate::binning::container_events::ContainerEvents;
@@ -1143,25 +1143,7 @@ impl<STY: ScalarOps> Events for EventsDim0<STY> {
return Box::new(ret);
}
let styn = any::type_name::<STY>();
todo!("TODO for {styn}")
}
}
fn try_to_container_events_fn<STY, EVT>(
this: &EventsDim0<STY>,
) -> Option<Box<dyn ::items_0::timebin::BinningggContainerEventsDyn>>
where
STY: ScalarOps,
EVT: crate::binning::container_events::EventValueType<Container = std::collections::VecDeque<STY>>,
{
use crate::binning::container_events::ContainerEvents;
if let Some(evs) = this.as_any_ref().downcast_ref::<EventsDim0<STY>>() {
let tss = this.tss.iter().map(|&x| TsNano::from_ns(x)).collect();
let vals = evs.values.clone();
let ret = ContainerEvents::<EVT>::from_constituents(tss, vals);
Some(Box::new(ret))
} else {
None
todo!("TODO to_container_events for {styn}")
}
}
@@ -1352,7 +1334,7 @@ mod test_frame {
let events: Box<dyn Events> = Box::new(events);
let item = ChannelEvents::Events(events);
let item = Ok::<_, Error>(StreamItem::DataItem(RangeCompletableItem::Data(item)));
let mut buf = item.make_frame().unwrap();
let mut buf = item.make_frame_dyn().unwrap();
let s = String::from_utf8_lossy(&buf[20..buf.len() - 4]);
eprintln!("[[{s}]]");
let buflen = buf.len();

12
crates/items_2/src/framable.rs
View File

@@ -59,7 +59,7 @@ impl FrameType for Box<dyn Events> {
}
pub trait Framable {
fn make_frame(&self) -> Result<BytesMut, Error>;
fn make_frame_dyn(&self) -> Result<BytesMut, Error>;
}
pub trait FramableInner: erased_serde::Serialize + FrameTypeInnerDyn + Send {
@@ -74,7 +74,7 @@ impl<T> Framable for Sitemty<T>
where
T: Sized + serde::Serialize + FrameType,
{
fn make_frame(&self) -> Result<BytesMut, Error> {
fn make_frame_dyn(&self) -> Result<BytesMut, Error> {
match self {
Ok(StreamItem::DataItem(RangeCompletableItem::Data(k))) => {
let frame_type_id = k.frame_type_id();
@@ -95,8 +95,8 @@ impl<T> Framable for Box<T>
where
T: Framable + ?Sized,
{
fn make_frame(&self) -> Result<BytesMut, Error> {
self.as_ref().make_frame()
fn make_frame_dyn(&self) -> Result<BytesMut, Error> {
self.as_ref().make_frame_dyn()
}
}
@@ -177,7 +177,7 @@ fn test_frame_log() {
msg: format!("test-log-message"),
};
let item: Sitemty<ChannelEvents> = Ok(StreamItem::Log(item));
let buf = Framable::make_frame(&item).unwrap();
let buf = Framable::make_frame_dyn(&item).unwrap();
let len = u32::from_le_bytes(buf[12..16].try_into().unwrap());
let item2: LogItem = decode_from_slice(&buf[20..20 + len as usize]).unwrap();
}
@@ -187,7 +187,7 @@ fn test_frame_error() {
use crate::channelevents::ChannelEvents;
use crate::frame::json_from_slice;
let item: Sitemty<ChannelEvents> = Err(Error::with_msg_no_trace(format!("dummy-error-message")));
let buf = Framable::make_frame(&item).unwrap();
let buf = Framable::make_frame_dyn(&item).unwrap();
let len = u32::from_le_bytes(buf[12..16].try_into().unwrap());
let tyid = u32::from_le_bytes(buf[8..12].try_into().unwrap());
if tyid != ERROR_FRAME_TYPE_ID {

16
crates/items_2/src/frame.rs
View File

@@ -27,22 +27,6 @@ use serde::Serialize;
use std::any;
use std::io;
trait EC {
fn ec(self) -> err::Error;
}
impl EC for rmp_serde::encode::Error {
fn ec(self) -> err::Error {
err::Error::with_msg_no_trace(format!("{self:?}"))
}
}
impl EC for rmp_serde::decode::Error {
fn ec(self) -> err::Error {
err::Error::with_msg_no_trace(format!("{self:?}"))
}
}
pub fn bincode_ser<W>(
w: W,
) -> bincode::Serializer<

2
crates/nodenet/src/client.rs
View File

@@ -28,7 +28,7 @@ async fn open_bytes_data_streams_http(
let mut streams = Vec::new();
for node in &cluster.nodes {
let item = sitem_data(frame1.clone());
let buf = item.make_frame()?;
let buf = item.make_frame_dyn()?;
let url = node.baseurl().join("/api/4/private/eventdata/frames").unwrap();
debug!("open_event_data_streams_http post {url}");

8
crates/nodenet/src/conn.rs
View File

@@ -139,7 +139,7 @@ pub async fn create_response_bytes_stream(
let fetch_info = evq.ch_conf().to_sf_databuffer()?;
let stream = disk::raw::conn::make_event_blobs_pipe(&evq, &fetch_info, reqctx, ncc)?;
// let stream = stream.map(|x| Box::new(x) as _);
let stream = stream.map(|x| x.make_frame().map(|x| x.freeze()));
let stream = stream.map(|x| x.make_frame_dyn().map(|x| x.freeze()));
let ret = Box::pin(stream);
Ok(ret)
} else {
@@ -161,7 +161,7 @@ pub async fn create_response_bytes_stream(
})
});
// let stream = stream.map(move |x| Box::new(x) as Box<dyn Framable + Send>);
let stream = stream.map(|x| x.make_frame().map(bytes::BytesMut::freeze));
let stream = stream.map(|x| x.make_frame_dyn().map(bytes::BytesMut::freeze));
let ret = Box::pin(stream);
Ok(ret)
}
@@ -204,7 +204,7 @@ async fn events_conn_handler_with_reqid(
msg: format!("buf_len_histo: {:?}", buf_len_histo),
};
let item: Sitemty<ChannelEvents> = Ok(StreamItem::Log(item));
let buf = match item.make_frame() {
let buf = match item.make_frame_dyn() {
Ok(k) => k,
Err(e) => return Err((e, netout))?,
};
@@ -331,7 +331,7 @@ where
Err(ce) => {
let mut out = ce.netout;
let item: Sitemty<ChannelEvents> = Err(err::Error::from_string(ce.err));
let buf = Framable::make_frame(&item)?;
let buf = Framable::make_frame_dyn(&item)?;
out.write_all(&buf).await?;
}
}

2
crates/nodenet/src/conn/test.rs
View File

@@ -89,7 +89,7 @@ fn raw_data_00() {
let qu = EventsSubQuery::from_parts(select, settings, "dummy".into(), log_level);
let frame1 = Frame1Parts::new(qu.clone());
let query = EventQueryJsonStringFrame(serde_json::to_string(&frame1).unwrap());
let frame = sitem_data(query).make_frame()?;
let frame = sitem_data(query).make_frame_dyn()?;
let scyqueue = err::todoval();
let jh = taskrun::spawn(events_conn_handler(client, addr, scyqueue, cfg));
con.write_all(&frame).await.unwrap();

2
crates/streams/src/generators.rs
View File

@@ -57,7 +57,7 @@ pub fn make_test_channel_events_bytes_stream(
}
})
});
let stream = stream.map(|x| x.make_frame().map(|x| x.freeze()));
let stream = stream.map(|x| x.make_frame_dyn().map(|x| x.freeze()));
let ret = Box::pin(stream);
Ok(ret)
}

6
crates/streams/src/tcprawclient.rs
View File

@@ -69,7 +69,7 @@ pub async fn x_processed_event_blobs_stream_from_node_tcp(
let net = TcpStream::connect(addr.clone()).await?;
let (netin, mut netout) = net.into_split();
let item = sitem_data(frame1);
let buf = item.make_frame()?;
let buf = item.make_frame_dyn()?;
netout.write_all(&buf).await?;
let buf = make_term_frame()?;
netout.write_all(&buf).await?;
@@ -95,7 +95,7 @@ pub async fn x_processed_event_blobs_stream_from_node_http(
let frame1 = make_node_command_frame(subq.clone())?;
let item = sitem_data(frame1.clone());
let buf = item.make_frame()?;
let buf = item.make_frame_dyn()?;
let url = node.baseurl().join("/api/4/private/eventdata/frames").unwrap();
debug!("open_event_data_streams_http post {url}");
@@ -168,7 +168,7 @@ where
let net = TcpStream::connect(addr.clone()).await?;
let (netin, mut netout) = net.into_split();
let item = sitem_data(frame1.clone());
let buf = item.make_frame()?;
let buf = item.make_frame_dyn()?;
netout.write_all(&buf).await?;
let buf = make_term_frame()?;
netout.write_all(&buf).await?;