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
2023-04-18 10:47:24 +02:00
parent 7f641f5971
commit 6c9f084623
9 changed files with 282 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
items_0/src/timebin.rs
View File

@@ -145,6 +145,18 @@ impl TimeBinnable for Box<dyn Events> {
}
}
impl TypeName for Box<dyn TimeBinnable> {
fn type_name(&self) -> String {
format!("Box<dyn TimeBinnable> TODO TypeName for Box<dyn TimeBinnable>")
}
}
impl Collectable for Box<dyn TimeBinnable> {
fn new_collector(&self) -> Box<dyn crate::collect_s::Collector> {
todo!()
}
}
#[derive(Debug)]
pub struct TimeBinnerDynStruct {
binrange: BinnedRangeEnum,

32
items_0/src/transform.rs
View File

@@ -106,17 +106,41 @@ impl EventTransform for TransformEvent {
}
}
pub struct EventStream(pub Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Events>>> + Send>>);
impl<T> WithTransformProperties for stream::Iter<T> {
fn query_transform_properties(&self) -> TransformProperties {
todo!()
}
}
impl<T> From<T> for EventStream
impl<T> EventStreamTrait for stream::Iter<T> where
T: core::iter::Iterator<Item = Sitemty<Box<(dyn Events + 'static)>>> + Send
{
}
pub struct EventStreamBox(pub Pin<Box<dyn EventStreamTrait>>);
impl<T> From<T> for EventStreamBox
where
T: Events,
{
fn from(value: T) -> Self {
let item = Ok(StreamItem::DataItem(RangeCompletableItem::Data(Box::new(value) as _)));
let x = stream::iter(vec![item]);
EventStream(Box::pin(x))
Self(Box::pin(x))
}
}
pub struct CollectableStream(pub Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Collectable>>> + Send>>);
pub struct CollectableStreamBox(pub Pin<Box<dyn CollectableStreamTrait>>);
impl<T> WithTransformProperties for stream::Empty<T> {
fn query_transform_properties(&self) -> TransformProperties {
todo!()
}
}
impl<T> CollectableStreamTrait for stream::Empty<T>
where
T: Send,
stream::Empty<T>: Stream<Item = Sitemty<Box<dyn Collectable>>>,
{
}

54
items_2/src/streams.rs
View File

@@ -2,9 +2,14 @@ use futures_util::Future;
use futures_util::FutureExt;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::Collectable;
use items_0::streamitem::Sitemty;
use items_0::transform::CollectableStreamTrait;
use items_0::transform::EventStreamTrait;
use items_0::transform::EventTransform;
use items_0::transform::TransformProperties;
use items_0::transform::WithTransformProperties;
use items_0::Events;
use std::collections::VecDeque;
use std::pin::Pin;
use std::task::Context;
@@ -217,3 +222,52 @@ impl<T> EventTransform for VecStream<T> {
todo!()
}
}
/// Wrap any event stream and provide transformation properties.
pub struct PlainEventStream<INP, T>
where
T: Events,
INP: Stream<Item = Sitemty<T>>,
{
inp: Pin<Box<INP>>,
}
impl<INP, T> PlainEventStream<INP, T>
where
T: Events,
INP: Stream<Item = Sitemty<T>>,
{
pub fn new(inp: INP) -> Self {
Self { inp: Box::pin(inp) }
}
}
impl<INP, T> Stream for PlainEventStream<INP, T>
where
T: Events,
INP: Stream<Item = Sitemty<T>>,
{
type Item = Sitemty<Box<dyn Events>>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
todo!()
}
}
impl<INP, T> WithTransformProperties for PlainEventStream<INP, T>
where
T: Events,
INP: Stream<Item = Sitemty<T>>,
{
fn query_transform_properties(&self) -> TransformProperties {
todo!()
}
}
impl<INP, T> EventStreamTrait for PlainEventStream<INP, T>
where
T: Events,
INP: Stream<Item = Sitemty<T>> + Send,
{
}

1
items_2/src/transform.rs
View File

@@ -3,7 +3,6 @@
use crate::channelevents::ChannelEvents;
use crate::eventsdim0::EventsDim0;
use items_0::transform::EventStream;
use items_0::transform::EventTransform;
use items_0::transform::TransformEvent;
use items_0::transform::TransformProperties;

59
streams/src/boxed.rs
View File

@@ -1,20 +1,59 @@
use futures_util::stream::StreamExt;
use futures_util::Stream;
use items_0::on_sitemty_data;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::transform::EventStream;
use items_0::streamitem::StreamItem;
use items_0::transform::EventStreamBox;
use items_0::transform::TransformProperties;
use items_0::transform::WithTransformProperties;
use items_0::Events;
use std::pin::Pin;
use std::task::Context;
use std::task::Poll;
pub fn boxed_event_stream<T, S>(inp: S) -> EventStream
pub struct IntoBoxedEventStream<INP, T>
where
T: Events,
S: Stream<Item = Sitemty<T>> + Send + 'static,
INP: Stream<Item = Sitemty<T>> + WithTransformProperties,
{
let stream = inp.map(|x| {
let x = on_sitemty_data!(x, |x| Ok(StreamItem::DataItem(RangeCompletableItem::Data(
Box::new(x) as Box<dyn Events>
))));
x
});
EventStream(Box::pin(stream))
//inp: Pin<Box<dyn Stream<Item = Sitemty<T>>>>,
inp: Pin<Box<INP>>,
}
impl<INP, T> Stream for IntoBoxedEventStream<INP, T>
where
T: Events,
INP: Stream<Item = Sitemty<T>> + WithTransformProperties,
{
type Item = Sitemty<Box<dyn Events>>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
match self.inp.poll_next_unpin(cx) {
Ready(Some(item)) => Ready(Some(match item {
Ok(item) => Ok(match item {
StreamItem::DataItem(item) => StreamItem::DataItem(match item {
RangeCompletableItem::RangeComplete => RangeCompletableItem::RangeComplete,
RangeCompletableItem::Data(item) => RangeCompletableItem::Data(Box::new(item)),
}),
StreamItem::Log(item) => StreamItem::Log(item),
StreamItem::Stats(item) => StreamItem::Stats(item),
}),
Err(e) => Err(e),
})),
Ready(None) => Ready(None),
Pending => Pending,
}
}
}
impl<INP, T> WithTransformProperties for IntoBoxedEventStream<INP, T>
where
T: Events,
INP: Stream<Item = Sitemty<T>> + WithTransformProperties,
{
fn query_transform_properties(&self) -> TransformProperties {
self.inp.query_transform_properties()
}
}

34
streams/src/collect.rs
View File

@@ -1,4 +1,5 @@
use err::Error;
use futures_util::Future;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::Collectable;
@@ -8,13 +9,18 @@ use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StatsItem;
use items_0::streamitem::StreamItem;
use items_0::transform::EventStream;
use items_0::transform::CollectableStreamBox;
use items_0::transform::CollectableStreamTrait;
use items_0::transform::EventStreamTrait;
use items_0::WithLen;
use netpod::log::*;
use netpod::range::evrange::SeriesRange;
use netpod::BinnedRangeEnum;
use netpod::DiskStats;
use std::fmt;
use std::pin::Pin;
use std::task::Context;
use std::task::Poll;
use std::time::Duration;
use std::time::Instant;
use tracing::Instrument;
@@ -38,7 +44,31 @@ macro_rules! trace4 {
}
pub struct Collect {
inp: EventStream,
inp: CollectableStreamBox,
deadline: Instant,
}
impl Collect {
pub fn new<INP>(inp: INP, deadline: Instant) -> Self
where
INP: CollectableStreamTrait + 'static,
{
Self {
inp: CollectableStreamBox(Box::pin(inp)),
deadline,
}
}
}
impl Future for Collect {
type Output = Sitemty<Box<dyn Collected>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
use Poll::*;
let span = tracing::span!(Level::INFO, "Collect");
let _spg = span.enter();
todo!()
}
}
async fn collect_in_span<T, S>(

12
streams/src/plaineventsjson.rs
View File

@@ -1,5 +1,8 @@
use crate::collect::Collect;
use crate::rangefilter2::RangeFilter2;
use crate::tcprawclient::open_tcp_streams;
use crate::transform::EventsToTimeBinnable;
use crate::transform::TimeBinnableToCollectable;
use err::Error;
use futures_util::stream;
use futures_util::StreamExt;
@@ -7,6 +10,7 @@ use items_0::on_sitemty_data;
use items_0::streamitem::sitem_data;
use items_2::channelevents::ChannelEvents;
use items_2::merger::Merger;
use items_2::streams::PlainEventStream;
use netpod::log::*;
use netpod::ChConf;
use netpod::Cluster;
@@ -98,7 +102,13 @@ pub async fn plain_events_json(evq: &PlainEventsQuery, chconf: &ChConf, cluster:
item
});
let stream = stream::iter([empty]).chain(stream);
let collected = crate::collect::collect(stream, deadline, events_max, Some(evq.range().clone()), None).await?;
let stream = PlainEventStream::new(stream);
let stream = EventsToTimeBinnable::new(stream);
let stream = TimeBinnableToCollectable::new(stream);
// TODO allow Collect to respect events_max and give range to compute continue-at.
//let collected = crate::collect::collect(stream, deadline, events_max, Some(evq.range().clone()), None).await?;
let collected = Collect::new(stream, deadline).await;
let jsval = serde_json::to_value(&collected)?;
Ok(jsval)
}

6
streams/src/timebinnedjson.rs
View File

@@ -28,6 +28,12 @@ pub async fn timebinned_json(query: &BinnedQuery, chconf: &ChConf, cluster: &Clu
let empty = ChannelEvents::Events(empty);
let empty = sitem_data(empty);
// TODO this part is supposed to be done one the event nodes, if sharded:
//crate::transform::build_event_transform(tr, inp);
query.transform();
crate::transform::build_event_transform;
// TODO
let evquery = PlainEventsQuery::new(query.channel().clone(), query.range().clone()).for_time_weighted_scalar();
let inps = open_tcp_streams::<_, items_2::channelevents::ChannelEvents>(&evquery, cluster).await?;

97
streams/src/transform.rs
View File

@@ -1,16 +1,19 @@
use err::Error;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::Collectable;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StreamItem;
use items_0::timebin::TimeBinnable;
use items_0::transform::CollectableStream;
use items_0::transform::EventStream;
use items_0::transform::CollectableStreamBox;
use items_0::transform::CollectableStreamTrait;
use items_0::transform::EventStreamBox;
use items_0::transform::EventStreamTrait;
use items_0::transform::TimeBinnableStreamTrait;
use items_0::transform::TransformEvent;
use items_0::transform::TransformProperties;
use items_0::transform::WithTransformProperties;
use items_0::Events;
use items_2::transform::make_transform_identity;
use items_2::transform::make_transform_min_max_avg;
use items_2::transform::make_transform_pulse_id_diff;
@@ -21,7 +24,7 @@ use std::pin::Pin;
use std::task::Context;
use std::task::Poll;
pub fn build_event_transform(tr: &TransformQuery, inp: EventStream) -> Result<TransformEvent, Error> {
pub fn build_event_transform(tr: &TransformQuery, inp: EventStreamBox) -> Result<TransformEvent, Error> {
let trev = tr.get_tr_event();
match trev {
EventTransformQuery::ValueFull => Ok(make_transform_identity()),
@@ -43,6 +46,15 @@ pub struct EventsToTimeBinnable {
inp: Pin<Box<dyn EventStreamTrait>>,
}
impl EventsToTimeBinnable {
pub fn new<INP>(inp: INP) -> Self
where
INP: EventStreamTrait + 'static,
{
Self { inp: Box::pin(inp) }
}
}
impl Stream for EventsToTimeBinnable {
type Item = Sitemty<Box<dyn TimeBinnable>>;
@@ -67,12 +79,82 @@ impl Stream for EventsToTimeBinnable {
}
impl WithTransformProperties for EventsToTimeBinnable {
fn query_transform_properties(&self) -> items_0::transform::TransformProperties {
fn query_transform_properties(&self) -> TransformProperties {
self.inp.query_transform_properties()
}
}
pub fn build_full_transform_collectable(tr: &TransformQuery, inp: EventStream) -> Result<CollectableStream, Error> {
impl TimeBinnableStreamTrait for EventsToTimeBinnable {}
pub struct TimeBinnableToCollectable {
inp: Pin<Box<dyn TimeBinnableStreamTrait>>,
}
impl TimeBinnableToCollectable {
pub fn new<INP>(inp: INP) -> Self
where
INP: TimeBinnableStreamTrait + 'static,
{
Self { inp: Box::pin(inp) }
}
}
impl Stream for TimeBinnableToCollectable {
type Item = Sitemty<Box<dyn Collectable>>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
match self.inp.poll_next_unpin(cx) {
Ready(Some(item)) => Ready(Some(match item {
Ok(item) => Ok(match item {
StreamItem::DataItem(item) => StreamItem::DataItem(match item {
RangeCompletableItem::RangeComplete => RangeCompletableItem::RangeComplete,
RangeCompletableItem::Data(item) => RangeCompletableItem::Data(Box::new(item)),
}),
StreamItem::Log(item) => StreamItem::Log(item),
StreamItem::Stats(item) => StreamItem::Stats(item),
}),
Err(e) => Err(e),
})),
Ready(None) => Ready(None),
Pending => Pending,
}
}
}
impl WithTransformProperties for TimeBinnableToCollectable {
fn query_transform_properties(&self) -> TransformProperties {
self.inp.query_transform_properties()
}
}
impl CollectableStreamTrait for TimeBinnableToCollectable {}
pub fn build_time_binning_transform(
tr: &TransformQuery,
inp: Pin<Box<dyn TimeBinnableStreamTrait>>,
) -> Result<TransformEvent, Error> {
let trev = tr.get_tr_event();
match trev {
EventTransformQuery::ValueFull => Ok(make_transform_identity()),
EventTransformQuery::MinMaxAvgDev => Ok(make_transform_min_max_avg()),
EventTransformQuery::ArrayPick(..) => Err(Error::with_msg_no_trace(format!(
"build_event_transform don't know what to do {trev:?}"
))),
EventTransformQuery::PulseIdDiff => Ok(make_transform_pulse_id_diff()),
EventTransformQuery::EventBlobsVerbatim => Err(Error::with_msg_no_trace(format!(
"build_event_transform don't know what to do {trev:?}"
))),
EventTransformQuery::EventBlobsUncompressed => Err(Error::with_msg_no_trace(format!(
"build_event_transform don't know what to do {trev:?}"
))),
}
}
pub fn build_full_transform_collectable(
tr: &TransformQuery,
inp: EventStreamBox,
) -> Result<CollectableStreamBox, Error> {
// TODO this must return a Stream!
//let evs = build_event_transform(tr, inp)?;
let trtb = tr.get_tr_time_binning();
@@ -98,8 +180,9 @@ pub fn build_full_transform_collectable(tr: &TransformQuery, inp: EventStream) -
}));
let stream: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Collectable>>> + Send>> =
Box::pin(futures_util::stream::empty());
let stream = Box::pin(futures_util::stream::empty()) as _;
match trtb {
TimeBinningTransformQuery::None => Ok(CollectableStream(stream)),
TimeBinningTransformQuery::None => Ok(CollectableStreamBox(stream)),
TimeBinningTransformQuery::TimeWeighted => todo!(),
TimeBinningTransformQuery::Unweighted => todo!(),
}