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

Remove IntoBinnedT

Browse Source
This commit is contained in:
Dominik Werder
2021-06-09 20:28:51 +02:00
parent 88864435b2
commit f69523bace
3 changed files with 2 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
disk/src/agg/binnedt2.rs
View File

@@ -24,194 +24,6 @@ pub trait AggregatableTdim2: Sized {
fn aggregator_new_static(ts1: u64, ts2: u64) -> Self::Aggregator;
}
pub trait IntoBinnedT {
type StreamOut: Stream;
fn into_binned_t(self, spec: BinnedRange) -> Self::StreamOut;
}
impl<S, I> IntoBinnedT for S
where
S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
I: AggregatableTdim2 + Unpin,
I::Aggregator: Unpin,
{
type StreamOut = IntoBinnedTDefaultStream<S, I>;
fn into_binned_t(self, spec: BinnedRange) -> Self::StreamOut {
IntoBinnedTDefaultStream::new(self, spec)
}
}
pub struct IntoBinnedTDefaultStream<S, I>
where
S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>>,
I: AggregatableTdim2,
{
inp: S,
aggtor: Option<I::Aggregator>,
spec: BinnedRange,
curbin: u32,
inp_completed: bool,
all_bins_emitted: bool,
range_complete_observed: bool,
range_complete_emitted: bool,
left: Option<Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>>,
errored: bool,
completed: bool,
tmp_agg_results: VecDeque<I>,
}
impl<S, I> IntoBinnedTDefaultStream<S, I>
where
S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
I: AggregatableTdim2,
{
pub fn new(inp: S, spec: BinnedRange) -> Self {
let range = spec.get_range(0);
Self {
inp,
aggtor: Some(I::aggregator_new_static(range.beg, range.end)),
spec,
curbin: 0,
inp_completed: false,
all_bins_emitted: false,
range_complete_observed: false,
range_complete_emitted: false,
left: None,
errored: false,
completed: false,
tmp_agg_results: VecDeque::new(),
}
}
fn cur(&mut self, cx: &mut Context) -> Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>> {
if let Some(cur) = self.left.take() {
cur
} else if self.inp_completed {
Poll::Ready(None)
} else {
let inp_poll_span = span!(Level::TRACE, "into_t_inp_poll");
inp_poll_span.in_scope(|| self.inp.poll_next_unpin(cx))
}
}
fn cycle_current_bin(&mut self) {
self.curbin += 1;
let range = self.spec.get_range(self.curbin);
let ret = self
.aggtor
.replace(I::aggregator_new_static(range.beg, range.end))
// TODO handle None case, or remove Option if Agg is always present
.unwrap()
.result();
self.tmp_agg_results = ret.into();
if self.curbin >= self.spec.count as u32 {
self.all_bins_emitted = true;
}
}
fn handle(
&mut self,
cur: Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>,
) -> Option<Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>> {
use Poll::*;
match cur {
Ready(Some(Ok(item))) => match item {
StreamItem::Log(item) => Some(Ready(Some(Ok(StreamItem::Log(item))))),
StreamItem::Stats(item) => Some(Ready(Some(Ok(StreamItem::Stats(item))))),
StreamItem::DataItem(item) => match item {
RangeCompletableItem::RangeComplete => Some(Ready(Some(Ok(StreamItem::DataItem(
RangeCompletableItem::RangeComplete,
))))),
RangeCompletableItem::Data(item) => {
if self.all_bins_emitted {
// Just drop the item because we will not emit anymore data.
// Could also at least gather some stats.
None
} else {
let ag = self.aggtor.as_mut().unwrap();
if ag.ends_before(&item) {
None
} else if ag.starts_after(&item) {
self.left =
Some(Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))))));
self.cycle_current_bin();
// TODO cycle_current_bin enqueues the bin, can I return here instead?
None
} else {
let mut item = item;
ag.ingest(&mut item);
let item = item;
if ag.ends_after(&item) {
self.left =
Some(Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))))));
self.cycle_current_bin();
}
// TODO cycle_current_bin enqueues the bin, can I return here instead?
None
}
}
}
},
},
Ready(Some(Err(e))) => {
self.errored = true;
Some(Ready(Some(Err(e))))
}
Ready(None) => {
self.inp_completed = true;
if self.all_bins_emitted {
None
} else {
self.cycle_current_bin();
// TODO cycle_current_bin enqueues the bin, can I return here instead?
None
}
}
Pending => Some(Pending),
}
}
}
impl<S, I> Stream for IntoBinnedTDefaultStream<S, I>
where
S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
I: AggregatableTdim2 + Unpin,
I::Aggregator: Unpin,
{
type Item = Result<StreamItem<RangeCompletableItem<I>>, Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
'outer: loop {
break if self.completed {
panic!("IntoBinnedTDefaultStream poll_next on completed");
} else if self.errored {
self.completed = true;
Ready(None)
} else if let Some(item) = self.tmp_agg_results.pop_front() {
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))))
} else if self.range_complete_emitted {
self.completed = true;
Ready(None)
} else if self.inp_completed && self.all_bins_emitted {
self.range_complete_emitted = true;
if self.range_complete_observed {
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
} else {
continue 'outer;
}
} else {
let cur = self.cur(cx);
match self.handle(cur) {
Some(item) => item,
None => continue 'outer,
}
};
}
}
}
pub struct MinMaxAvgScalarBinBatchAgg {
ts1: u64,
ts2: u64,

15
disk/src/binned.rs
View File

@@ -8,7 +8,7 @@ use crate::agg::streams::{Appendable, Collectable, Collector, StreamItem, ToJson
use crate::agg::{Fits, FitsInside};
use crate::binned::binnedfrompbv::BinnedFromPreBinned;
use crate::binned::query::{BinnedQuery, PreBinnedQuery};
use crate::binnedstream::{BinnedScalarStreamFromPreBinnedPatches, BoxedStream};
use crate::binnedstream::BoxedStream;
use crate::cache::MergedFromRemotes;
use crate::decode::{
BigEndian, Endianness, EventValueFromBytes, EventValueShape, EventValues, EventValuesDim0Case, EventValuesDim1Case,
@@ -1283,18 +1283,7 @@ impl StreamKind for BinnedStreamKindScalar {
pre_range: PreBinnedPatchRange,
node_config: &NodeConfigCached,
) -> Result<Self::TBinnedStreamType, Error> {
let s = BinnedScalarStreamFromPreBinnedPatches::new(
PreBinnedPatchIterator::from_range(pre_range),
query.channel().clone(),
range.clone(),
query.agg_kind().clone(),
query.cache_usage().clone(),
node_config,
query.disk_stats_every().clone(),
query.report_error(),
self.clone(),
)?;
Ok(BoxedStream::new(Box::pin(s))?)
err::todoval()
}
fn new_binned_from_merged(

114
disk/src/binnedstream.rs
View File

@@ -12,120 +12,6 @@ use std::future::ready;
use std::pin::Pin;
use std::task::{Context, Poll};
// TODO remove after refactor.
pub struct BinnedScalarStreamFromPreBinnedPatches<SK>
where
SK: StreamKind,
{
inp: Pin<
Box<dyn Stream<Item = Result<StreamItem<RangeCompletableItem<<SK as StreamKind>::TBinnedBins>>, Error>> + Send>,
>,
_stream_kind: SK,
}
impl<SK> BinnedScalarStreamFromPreBinnedPatches<SK>
where
SK: StreamKind,
Result<StreamItem<RangeCompletableItem<<SK as StreamKind>::TBinnedBins>>, Error>: FrameType,
{
pub fn new(
patch_it: PreBinnedPatchIterator,
channel: Channel,
range: BinnedRange,
agg_kind: AggKind,
cache_usage: CacheUsage,
node_config: &NodeConfigCached,
disk_stats_every: ByteSize,
report_error: bool,
stream_kind: SK,
) -> Result<Self, Error> {
let patches: Vec<_> = patch_it.collect();
let mut sp = String::new();
if false {
// Convert this to a StreamLog message:
for (i, p) in patches.iter().enumerate() {
use std::fmt::Write;
write!(sp, " • patch {:2} {:?}\n", i, p)?;
}
info!("Using these pre-binned patches:\n{}", sp);
}
let pmax = patches.len();
let inp = futures_util::stream::iter(patches.into_iter().enumerate())
.map({
let node_config = node_config.clone();
let stream_kind = stream_kind.clone();
move |(pix, patch)| {
let query = PreBinnedQuery::new(
patch,
channel.clone(),
agg_kind.clone(),
cache_usage.clone(),
disk_stats_every.clone(),
report_error,
);
let ret: Pin<Box<dyn Stream<Item = _> + Send>> =
match PreBinnedScalarValueFetchedStream::new(&query, &node_config, &stream_kind) {
Ok(stream) => Box::pin(stream.map(move |q| (pix, q))),
Err(e) => {
error!("error from PreBinnedValueFetchedStream::new {:?}", e);
Box::pin(futures_util::stream::iter(vec![(pix, Err(e))]))
}
};
ret
}
})
.flatten()
.filter_map({
let range = range.clone();
move |(pix, k)| {
let fit_range = range.full_range();
let g = match k {
Ok(item) => match item {
StreamItem::Log(item) => Some(Ok(StreamItem::Log(item))),
StreamItem::Stats(item) => Some(Ok(StreamItem::Stats(item))),
StreamItem::DataItem(item) => match item {
RangeCompletableItem::RangeComplete => {
if pix + 1 == pmax {
Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete)))
} else {
None
}
}
RangeCompletableItem::Data(item) => {
match crate::binned::FilterFittingInside::filter_fitting_inside(item, fit_range) {
Some(item) => Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))),
None => None,
}
}
},
},
Err(e) => Some(Err(e)),
};
ready(g)
}
});
// TODO activate the T-binning via the bin-to-bin binning trait.
//err::todo();
let inp = crate::agg::binnedt2::IntoBinnedT::into_binned_t(inp, range);
Ok(Self {
inp: Box::pin(inp),
_stream_kind: stream_kind,
})
}
}
// TODO remove after SK no longer needed.
impl<SK> Stream for BinnedScalarStreamFromPreBinnedPatches<SK>
where
SK: StreamKind,
{
type Item = Result<StreamItem<RangeCompletableItem<SK::TBinnedBins>>, Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
self.inp.poll_next_unpin(cx)
}
}
// TODO remove after refactor.
pub struct BoxedStream<I> {
inp: Pin<Box<dyn Stream<Item = I> + Send>>,