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

Get X-binned dim-1 with N X-bins as json

Browse Source
This commit is contained in:
Dominik Werder
2021-06-16 13:57:45 +02:00
parent edafc610c2
commit 99d0a97a69
8 changed files with 405 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files

248
disk/src/agg/enp.rs
View File

@@ -1,14 +1,15 @@
use crate::agg::binnedt::{TimeBinnableType, TimeBinnableTypeAggregator};
use crate::agg::streams::Appendable;
use crate::agg::streams::{Appendable, Collectable, Collector};
use crate::agg::{Fits, FitsInside};
use crate::binned::dim1::MinMaxAvgDim1Bins;
use crate::binned::{
EventsNodeProcessor, FilterFittingInside, MinMaxAvgBins, MinMaxAvgWaveBins, NumOps, PushableIndex,
Bool, EventsNodeProcessor, FilterFittingInside, MinMaxAvgBins, MinMaxAvgWaveBins, NumOps, PushableIndex,
RangeOverlapInfo, ReadPbv, ReadableFromFile, WithLen, WithTimestamps,
};
use crate::decode::EventValues;
use err::Error;
use netpod::log::*;
use netpod::timeunits::{MS, SEC};
use netpod::{x_bin_count, AggKind, NanoRange, Shape};
use serde::{Deserialize, Serialize};
use std::marker::PhantomData;
@@ -35,7 +36,7 @@ where
}
// TODO rename Scalar -> Dim0
#[derive(Serialize, Deserialize)]
#[derive(Debug, Serialize, Deserialize)]
pub struct XBinnedScalarEvents<NTY> {
tss: Vec<u64>,
mins: Vec<NTY>,
@@ -275,8 +276,108 @@ where
}
}
// TODO rename Wave -> Dim1
#[derive(Serialize, Deserialize)]
pub struct XBinnedScalarEventsCollectedResult<NTY> {
#[serde(rename = "tsAnchor")]
ts_anchor_sec: u64,
#[serde(rename = "tsMs")]
ts_off_ms: Vec<u64>,
#[serde(rename = "tsNs")]
ts_off_ns: Vec<u64>,
mins: Vec<NTY>,
maxs: Vec<NTY>,
avgs: Vec<f32>,
#[serde(skip_serializing_if = "Bool::is_false", rename = "finalisedRange")]
finalised_range: bool,
#[serde(skip_serializing_if = "Bool::is_false", rename = "timedOut")]
timed_out: bool,
}
pub struct XBinnedScalarEventsCollector<NTY> {
vals: XBinnedScalarEvents<NTY>,
finalised_range: bool,
timed_out: bool,
#[allow(dead_code)]
bin_count_exp: u32,
}
impl<NTY> XBinnedScalarEventsCollector<NTY> {
pub fn new(bin_count_exp: u32) -> Self {
Self {
finalised_range: false,
timed_out: false,
vals: XBinnedScalarEvents::empty(),
bin_count_exp,
}
}
}
impl<NTY> WithLen for XBinnedScalarEventsCollector<NTY> {
fn len(&self) -> usize {
self.vals.tss.len()
}
}
pub fn ts_offs_from_abs(tss: &[u64]) -> (u64, Vec<u64>, Vec<u64>) {
let ts_anchor_sec = tss.first().map_or(0, |&k| k) / SEC;
let ts_anchor_ns = ts_anchor_sec * SEC;
let ts_off_ms: Vec<_> = tss.iter().map(|&k| (k - ts_anchor_ns) / MS).collect();
let ts_off_ns = tss
.iter()
.zip(ts_off_ms.iter().map(|&k| k * MS))
.map(|(&j, k)| (j - ts_anchor_ns - k))
.collect();
(ts_anchor_sec, ts_off_ms, ts_off_ns)
}
impl<NTY> Collector for XBinnedScalarEventsCollector<NTY>
where
NTY: NumOps,
{
type Input = XBinnedScalarEvents<NTY>;
type Output = XBinnedScalarEventsCollectedResult<NTY>;
fn ingest(&mut self, src: &Self::Input) {
self.vals.append(src);
}
fn set_range_complete(&mut self) {
self.finalised_range = true;
}
fn set_timed_out(&mut self) {
self.timed_out = true;
}
fn result(self) -> Result<Self::Output, Error> {
let tst = ts_offs_from_abs(&self.vals.tss);
let ret = Self::Output {
ts_anchor_sec: tst.0,
ts_off_ms: tst.1,
ts_off_ns: tst.2,
mins: self.vals.mins,
maxs: self.vals.maxs,
avgs: self.vals.avgs,
finalised_range: self.finalised_range,
timed_out: self.timed_out,
};
Ok(ret)
}
}
impl<NTY> Collectable for XBinnedScalarEvents<NTY>
where
NTY: NumOps,
{
type Collector = XBinnedScalarEventsCollector<NTY>;
fn new_collector(bin_count_exp: u32) -> Self::Collector {
Self::Collector::new(bin_count_exp)
}
}
// TODO rename Wave -> Dim1
#[derive(Debug, Serialize, Deserialize)]
pub struct XBinnedWaveEvents<NTY> {
tss: Vec<u64>,
mins: Vec<Vec<NTY>>,
@@ -435,11 +536,15 @@ where
NTY: NumOps,
{
pub fn new(range: NanoRange, bin_count: usize) -> Self {
if bin_count == 0 {
panic!("bin_count == 0");
}
Self {
range,
count: 0,
min: vec![NTY::min_or_nan(); bin_count],
max: vec![NTY::max_or_nan(); bin_count],
//min: vec![NTY::fourty_two(); bin_count],
max: vec![NTY::fourty_two(); bin_count],
sum: vec![0f32; bin_count],
sumc: 0,
}
@@ -458,6 +563,7 @@ where
}
fn ingest(&mut self, item: &Self::Input) {
//info!("XBinnedWaveEventsAggregator ingest item {:?}", item);
for i1 in 0..item.tss.len() {
let ts = item.tss[i1];
if ts < self.range.beg {
@@ -465,17 +571,17 @@ where
} else if ts >= self.range.end {
continue;
} else {
for (i2, v) in item.mins[i1].iter().enumerate() {
if *v < self.min[i2] || self.min[i2].is_nan() {
self.min[i2] = *v;
for (i2, &v) in item.mins[i1].iter().enumerate() {
if v < self.min[i2] || self.min[i2].is_nan() {
self.min[i2] = v;
}
}
for (i2, v) in item.maxs[i1].iter().enumerate() {
if *v > self.max[i2] || self.max[i2].is_nan() {
self.max[i2] = *v;
for (i2, &v) in item.maxs[i1].iter().enumerate() {
if v > self.max[i2] || self.max[i2].is_nan() {
self.max[i2] = v;
}
}
for (i2, v) in item.avgs[i1].iter().enumerate() {
for (i2, &v) in item.avgs[i1].iter().enumerate() {
if v.is_nan() {
} else {
self.sum[i2] += v;
@@ -499,19 +605,120 @@ where
}
} else {
let avg = self.sum.iter().map(|k| *k / self.sumc as f32).collect();
Self::Output {
let ret = Self::Output {
ts1s: vec![self.range.beg],
ts2s: vec![self.range.end],
counts: vec![self.count],
mins: vec![Some(self.min)],
maxs: vec![Some(self.max)],
avgs: vec![Some(avg)],
};
if ret.ts1s[0] < 1300 {
info!("XBinnedWaveEventsAggregator result {:?}", ret);
}
ret
}
}
}
#[derive(Serialize, Deserialize)]
pub struct XBinnedWaveEventsCollectedResult<NTY> {
#[serde(rename = "tsAnchor")]
ts_anchor_sec: u64,
#[serde(rename = "tsMs")]
ts_off_ms: Vec<u64>,
#[serde(rename = "tsNs")]
ts_off_ns: Vec<u64>,
mins: Vec<Vec<NTY>>,
maxs: Vec<Vec<NTY>>,
avgs: Vec<Vec<f32>>,
#[serde(skip_serializing_if = "Bool::is_false", rename = "finalisedRange")]
finalised_range: bool,
#[serde(skip_serializing_if = "Bool::is_false", rename = "timedOut")]
timed_out: bool,
}
pub struct XBinnedWaveEventsCollector<NTY> {
vals: XBinnedWaveEvents<NTY>,
finalised_range: bool,
timed_out: bool,
#[allow(dead_code)]
bin_count_exp: u32,
}
impl<NTY> XBinnedWaveEventsCollector<NTY> {
pub fn new(bin_count_exp: u32) -> Self {
Self {
finalised_range: false,
timed_out: false,
vals: XBinnedWaveEvents::empty(),
bin_count_exp,
}
}
}
impl<NTY> WithLen for XBinnedWaveEventsCollector<NTY> {
fn len(&self) -> usize {
self.vals.tss.len()
}
}
impl<NTY> Collector for XBinnedWaveEventsCollector<NTY>
where
NTY: NumOps,
{
type Input = XBinnedWaveEvents<NTY>;
type Output = XBinnedWaveEventsCollectedResult<NTY>;
fn ingest(&mut self, src: &Self::Input) {
self.vals.append(src);
}
fn set_range_complete(&mut self) {
self.finalised_range = true;
}
fn set_timed_out(&mut self) {
self.timed_out = true;
}
fn result(self) -> Result<Self::Output, Error> {
let ts_anchor_sec = self.vals.tss.first().map_or(0, |&k| k) / SEC;
let ts_anchor_ns = ts_anchor_sec * SEC;
let ts_off_ms: Vec<_> = self.vals.tss.iter().map(|&k| (k - ts_anchor_ns) / MS).collect();
let ts_off_ns = self
.vals
.tss
.iter()
.zip(ts_off_ms.iter().map(|&k| k * MS))
.map(|(&j, k)| (j - ts_anchor_ns - k))
.collect();
let ret = Self::Output {
finalised_range: self.finalised_range,
timed_out: self.timed_out,
ts_anchor_sec,
ts_off_ms,
ts_off_ns,
mins: self.vals.mins,
maxs: self.vals.maxs,
avgs: self.vals.avgs,
};
Ok(ret)
}
}
impl<NTY> Collectable for XBinnedWaveEvents<NTY>
where
NTY: NumOps,
{
type Collector = XBinnedWaveEventsCollector<NTY>;
fn new_collector(bin_count_exp: u32) -> Self::Collector {
Self::Collector::new(bin_count_exp)
}
}
#[derive(Debug, Serialize, Deserialize)]
pub struct WaveEvents<NTY> {
pub tss: Vec<u64>,
pub vals: Vec<Vec<NTY>>,
@@ -861,22 +1068,23 @@ where
fn process(&self, inp: EventValues<Self::Input>) -> Self::Output {
let nev = inp.tss.len();
let mut ret = Self::Output {
tss: inp.tss,
// TODO get rid of this clone:
tss: inp.tss.clone(),
mins: Vec::with_capacity(nev),
maxs: Vec::with_capacity(nev),
avgs: Vec::with_capacity(nev),
};
for i1 in 0..nev {
let mut min = vec![NTY::min_or_nan(); self.x_bin_count];
let mut max = vec![NTY::max_or_nan(); self.x_bin_count];
let mut min = vec![NTY::max_or_nan(); self.x_bin_count];
let mut max = vec![NTY::min_or_nan(); self.x_bin_count];
let mut sum = vec![0f32; self.x_bin_count];
let mut sumc = vec![0u64; self.x_bin_count];
for (i2, &v) in inp.values[i1].iter().enumerate() {
let i3 = i2 * self.x_bin_count / self.shape_bin_count;
if v < min[i3] {
if v < min[i3] || min[i3].is_nan() {
min[i3] = v;
}
if v > max[i3] {
if v > max[i3] || max[i3].is_nan() {
max[i3] = v;
}
if v.is_nan() {
@@ -885,6 +1093,10 @@ where
sumc[i3] += 1;
}
}
// TODO
if false && inp.tss[0] < 1300 {
info!("WaveNBinner process push min {:?}", min);
}
ret.mins.push(min);
ret.maxs.push(max);
let avg = sum

199
disk/src/binned.rs
View File

@@ -1,5 +1,5 @@
use crate::agg::binnedt::{TBinnerStream, TimeBinnableType, TimeBinnableTypeAggregator};
use crate::agg::enp::{Identity, WaveXBinner};
use crate::agg::enp::{ts_offs_from_abs, Identity, WaveXBinner};
use crate::agg::eventbatch::MinMaxAvgScalarEventBatch;
use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
use crate::agg::streams::{Appendable, Collectable, Collector, StreamItem, ToJsonBytes, ToJsonResult};
@@ -7,7 +7,7 @@ use crate::agg::{Fits, FitsInside};
use crate::binned::binnedfrompbv::BinnedFromPreBinned;
use crate::binned::query::BinnedQuery;
use crate::binnedstream::BoxedStream;
use crate::channelexec::{channel_exec, collect_plain_events_json, ChannelExecFunction, PlainEventsAggMethod};
use crate::channelexec::{channel_exec, collect_plain_events_json, ChannelExecFunction};
use crate::decode::{
BigEndian, Endianness, EventValueFromBytes, EventValueShape, EventValues, EventValuesDim0Case, EventValuesDim1Case,
LittleEndian, NumFromBytes,
@@ -24,7 +24,7 @@ use futures_util::{FutureExt, StreamExt};
use netpod::log::*;
use netpod::timeunits::SEC;
use netpod::{
AggKind, BinnedRange, ByteOrder, Channel, NanoRange, NodeConfigCached, PerfOpts, PreBinnedPatchIterator,
x_bin_count, AggKind, BinnedRange, ByteOrder, NanoRange, NodeConfigCached, PerfOpts, PreBinnedPatchIterator,
PreBinnedPatchRange, ScalarType, Shape,
};
use num_traits::{AsPrimitive, Bounded, Float, Zero};
@@ -269,6 +269,7 @@ where
}
}
#[allow(dead_code)]
fn make_num_pipeline_nty_end_old<PPP, NTY, END>(
shape: Shape,
query: BinnedQuery,
@@ -446,6 +447,7 @@ struct CollectForJson {
}
impl CollectForJson {
#[allow(dead_code)]
pub fn new(timeout: Duration, abort_after_bin_count: u32) -> Self {
Self {
timeout,
@@ -637,6 +639,12 @@ impl Serialize for IsoDateTime {
}
}
pub fn make_iso_ts(tss: &[u64]) -> Vec<IsoDateTime> {
tss.iter()
.map(|&k| IsoDateTime(Utc.timestamp_nanos(k as i64)))
.collect()
}
pub async fn collect_all<T, S>(
stream: S,
bin_count_exp: u32,
@@ -704,6 +712,10 @@ pub struct BinnedJsonChannelExec {
impl BinnedJsonChannelExec {
pub fn new(query: BinnedQuery, node_config: NodeConfigCached) -> Self {
info!(
"BinnedJsonChannelExec AggKind: {:?}\n--------------------------------------------------------------",
query.agg_kind()
);
Self {
query,
node_config,
@@ -717,7 +729,7 @@ impl ChannelExecFunction for BinnedJsonChannelExec {
fn exec<NTY, END, EVS, ENP>(
self,
byte_order: END,
_byte_order: END,
shape: Shape,
event_value_shape: EVS,
_events_node_proc: ENP,
@@ -725,14 +737,14 @@ impl ChannelExecFunction for BinnedJsonChannelExec {
where
NTY: NumOps + NumFromBytes<NTY, END> + 'static,
END: Endianness + 'static,
EVS: EventValueShape<NTY, END> + EventValueFromBytes<NTY, END> + PlainEventsAggMethod + 'static,
EVS: EventValueShape<NTY, END> + EventValueFromBytes<NTY, END> + 'static,
ENP: EventsNodeProcessor<Input = <EVS as EventValueFromBytes<NTY, END>>::Output> + 'static,
Sitemty<<<EVS as PlainEventsAggMethod>::Method as EventsNodeProcessor>::Output>: FrameType,
<<EVS as PlainEventsAggMethod>::Method as EventsNodeProcessor>::Output: Collectable + PushableIndex,
<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output:
TimeBinnableType<Output = <<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output> + Unpin,
// TODO require these things in general?
<ENP as EventsNodeProcessor>::Output: PushableIndex,
<ENP as EventsNodeProcessor>::Output: Collectable + PushableIndex,
<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output: Debug
+ TimeBinnableType<Output = <<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output>
+ Collectable
+ Unpin,
Sitemty<<ENP as EventsNodeProcessor>::Output>: FrameType + Framable + 'static,
Sitemty<<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output>:
FrameType + Framable + DeserializeOwned,
@@ -742,8 +754,9 @@ impl ChannelExecFunction for BinnedJsonChannelExec {
BinnedRange::covering_range(self.query.range().clone(), self.query.bin_count())?.ok_or(Error::with_msg(
format!("binned_bytes_for_http BinnedRange::covering_range returned None"),
))?;
let t_bin_count = range.count as u32;
let perf_opts = PerfOpts { inmem_bufcap: 512 };
match PreBinnedPatchRange::covering_range(self.query.range().clone(), self.query.bin_count()) {
let souter = match PreBinnedPatchRange::covering_range(self.query.range().clone(), self.query.bin_count()) {
Ok(Some(pre_range)) => {
info!("binned_bytes_for_http found pre_range: {:?}", pre_range);
if range.grid_spec.bin_t_len() < pre_range.grid_spec.bin_t_len() {
@@ -763,74 +776,37 @@ impl ChannelExecFunction for BinnedJsonChannelExec {
&self.node_config,
self.query.disk_stats_every().clone(),
self.query.report_error(),
)?
.map(|item| match item.make_frame() {
Ok(item) => Ok(item.freeze()),
Err(e) => Err(e),
)?;
let f = collect_plain_events_json(s, self.timeout, t_bin_count);
let s = futures_util::stream::once(f).map(|item| match item {
Ok(item) => Ok(Bytes::from(serde_json::to_vec(&item)?)),
Err(e) => Err(e.into()),
});
// TODO remove?
/*let ret = BinnedResponseStat {
stream: Box::pin(s),
bin_count: range.count as u32,
};*/
Ok(Box::pin(s))
Ok(Box::pin(s) as Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send>>)
}
Ok(None) => {
info!(
"binned_bytes_for_http no covering range for prebinned, merge from remotes instead {:?}",
range
);
let bin_count = range.count as u32;
let evq = EventsQuery {
channel: self.query.channel().clone(),
range: self.query.range().clone(),
agg_kind: self.query.agg_kind().clone(),
};
let x_bin_count = if let AggKind::DimXBinsN(n) = self.query.agg_kind() {
*n as usize
} else {
0
};
let x_bin_count = x_bin_count(&shape, self.query.agg_kind());
let s = MergedFromRemotes::<ENP>::new(evq, perf_opts, self.node_config.node_config.cluster.clone());
let s =
TBinnerStream::<_, <ENP as EventsNodeProcessor>::Output>::new(s, range, x_bin_count).map(|item| {
match item.make_frame() {
Ok(item) => Ok(item.freeze()),
Err(e) => Err(e),
}
});
/*let ret = BinnedResponseStat {
stream: Box::pin(s),
bin_count,
};*/
Ok(Box::pin(s))
let s = TBinnerStream::<_, <ENP as EventsNodeProcessor>::Output>::new(s, range, x_bin_count);
let f = collect_plain_events_json(s, self.timeout, t_bin_count);
let s = futures_util::stream::once(f).map(|item| match item {
Ok(item) => Ok(Bytes::from(serde_json::to_vec(&item)?)),
Err(e) => Err(e.into()),
});
Ok(Box::pin(s) as Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send>>)
}
Err(e) => Err(e),
}
/*let perf_opts = PerfOpts { inmem_bufcap: 4096 };
let evq = EventsQuery {
channel: self.channel,
range: self.range,
agg_kind: self.agg_kind,
};
let s = MergedFromRemotes::<<EVS as PlainEventsAggMethod>::Method>::new(
evq,
perf_opts,
self.node_config.node_config.cluster,
);
let f = collect_plain_events_json(s, self.timeout);
let f = FutureExt::map(f, |item| match item {
Ok(item) => {
// TODO add channel entry info here?
//let obj = item.as_object_mut().unwrap();
//obj.insert("channelName", JsonValue::String(en));
Ok(Bytes::from(serde_json::to_vec(&item)?))
}
Err(e) => Err(e.into()),
});
let s = futures_util::stream::once(f);
Ok(Box::pin(s))*/
}?;
Ok(souter)
}
fn empty() -> Self::Output {
@@ -842,11 +818,6 @@ pub async fn binned_json(
query: &BinnedQuery,
node_config: &NodeConfigCached,
) -> Result<Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send>>, Error> {
// TODO try the channelexec approach.
// TODO why does channel_exec need the range, and what does it use it for?
// do I want there the user-requested range or the bin-edge-adjusted range?
// TODO currently, channel_exec resolves NTY, END, EVS but not ENP!
// can I add that or does that break other things?
let ret = channel_exec(
BinnedJsonChannelExec::new(query.clone(), node_config.clone()),
query.channel(),
@@ -855,19 +826,6 @@ pub async fn binned_json(
node_config,
)
.await?;
/*let pl = make_num_pipeline(
query,
CollectForJson::new(query.timeout(), query.abort_after_bin_count()),
node_config,
)
.await?;
let ret = pl.stream.map(|item| {
let fr = item.to_json_result()?;
let buf = fr.to_json_bytes()?;
Ok(Bytes::from(buf))
});*/
Ok(Box::pin(ret))
}
@@ -1027,6 +985,7 @@ pub trait NumOps:
fn min_or_nan() -> Self;
fn max_or_nan() -> Self;
fn is_nan(&self) -> bool;
fn fourty_two() -> Self;
}
macro_rules! impl_num_ops {
@@ -1041,6 +1000,9 @@ macro_rules! impl_num_ops {
fn is_nan(&self) -> bool {
$is_nan(self)
}
fn fourty_two() -> Self {
42 as Self
}
}
};
}
@@ -1268,8 +1230,12 @@ impl<NTY> MinMaxAvgBinsCollected<NTY> {
#[derive(Serialize)]
pub struct MinMaxAvgBinsCollectedResult<NTY> {
ts0: u64,
tsoff: Vec<u64>,
#[serde(rename = "tsAnchor")]
ts_anchor_sec: u64,
#[serde(rename = "tsMs")]
ts_off_ms: Vec<u64>,
#[serde(rename = "tsNs")]
ts_off_ns: Vec<u64>,
//ts_bin_edges: Vec<IsoDateTime>,
counts: Vec<u64>,
mins: Vec<Option<NTY>>,
@@ -1352,9 +1318,16 @@ where
} else {
None
};
// TODO could save the copy:
let mut ts_all = self.vals.ts1s.clone();
if self.vals.ts2s.len() > 0 {
ts_all.push(*self.vals.ts2s.last().unwrap());
}
let tst = ts_offs_from_abs(&ts_all);
let ret = MinMaxAvgBinsCollectedResult::<NTY> {
ts0,
tsoff,
ts_anchor_sec: tst.0,
ts_off_ms: tst.1,
ts_off_ns: tst.2,
counts: self.vals.counts,
mins: self.vals.mins,
maxs: self.vals.maxs,
@@ -1573,7 +1546,7 @@ pub enum RangeCompletableItem<T> {
Data(T),
}
#[derive(Clone, Serialize, Deserialize)]
#[derive(Serialize, Deserialize)]
pub struct MinMaxAvgWaveBins<NTY> {
pub ts1s: Vec<u64>,
pub ts2s: Vec<u64>,
@@ -1590,7 +1563,7 @@ where
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
write!(
fmt,
"MinMaxAvgBins count {} ts1s {:?} ts2s {:?} counts {:?} mins {:?} maxs {:?} avgs {:?}",
"MinMaxAvgWaveBins count {} ts1s {:?} ts2s {:?} counts {:?} mins {:?} maxs {:?} avgs {:?}",
self.ts1s.len(),
self.ts1s.iter().map(|k| k / SEC).collect::<Vec<_>>(),
self.ts2s.iter().map(|k| k / SEC).collect::<Vec<_>>(),
@@ -1758,9 +1731,12 @@ impl<NTY> MinMaxAvgWaveBinsCollected<NTY> {
#[derive(Serialize)]
pub struct MinMaxAvgWaveBinsCollectedResult<NTY> {
ts0: u64,
tsoff: Vec<u64>,
//ts_bin_edges: Vec<IsoDateTime>,
#[serde(rename = "tsAnchor")]
ts_anchor_sec: u64,
#[serde(rename = "tsMs")]
ts_off_ms: Vec<u64>,
#[serde(rename = "tsNs")]
ts_off_ns: Vec<u64>,
counts: Vec<u64>,
mins: Vec<Option<Vec<NTY>>>,
maxs: Vec<Option<Vec<NTY>>>,
@@ -1770,8 +1746,7 @@ pub struct MinMaxAvgWaveBinsCollectedResult<NTY> {
#[serde(skip_serializing_if = "Zero::is_zero", rename = "missingBins")]
missing_bins: u32,
#[serde(skip_serializing_if = "Option::is_none", rename = "continueAt")]
//continue_at: Option<IsoDateTime>,
continue_at: Option<u64>,
continue_at: Option<IsoDateTime>,
}
pub struct MinMaxAvgWaveBinsCollector<NTY> {
@@ -1823,34 +1798,34 @@ where
}
fn result(self) -> Result<Self::Output, Error> {
let ts0 = self.vals.ts1s.first().map_or(0, |k| *k / SEC);
let bin_count = self.vals.ts1s.len() as u32;
let mut tsoff: Vec<_> = self.vals.ts1s.iter().map(|k| *k - ts0 * SEC).collect();
if let Some(&k) = self.vals.ts2s.last() {
tsoff.push(k - ts0 * SEC);
let t_bin_count = self.vals.counts.len();
// TODO could save the copy:
let mut ts_all = self.vals.ts1s.clone();
if self.vals.ts2s.len() > 0 {
ts_all.push(*self.vals.ts2s.last().unwrap());
}
let tsoff = tsoff;
let _iso: Vec<_> = tsoff
.iter()
.map(|&k| IsoDateTime(Utc.timestamp_nanos(k as i64)))
.collect();
let continue_at = if self.vals.ts1s.len() < self.bin_count_exp as usize {
match tsoff.last() {
Some(k) => Some(k.clone()),
match ts_all.last() {
Some(&k) => {
let iso = IsoDateTime(Utc.timestamp_nanos(k as i64));
Some(iso)
}
None => Err(Error::with_msg("partial_content but no bin in result"))?,
}
} else {
None
};
let tst = ts_offs_from_abs(&ts_all);
let ret = MinMaxAvgWaveBinsCollectedResult {
ts0,
tsoff,
ts_anchor_sec: tst.0,
ts_off_ms: tst.1,
ts_off_ns: tst.2,
counts: self.vals.counts,
mins: self.vals.mins,
maxs: self.vals.maxs,
avgs: self.vals.avgs,
finalised_range: self.range_complete,
missing_bins: self.bin_count_exp - bin_count,
missing_bins: self.bin_count_exp - t_bin_count as u32,
continue_at,
};
Ok(ret)
@@ -1885,8 +1860,8 @@ where
Self {
range,
count: 0,
min: vec![NTY::min_or_nan(); x_bin_count],
max: vec![NTY::max_or_nan(); x_bin_count],
min: vec![NTY::max_or_nan(); x_bin_count],
max: vec![NTY::min_or_nan(); x_bin_count],
sum: vec![0f32; x_bin_count],
sumc: 0,
}
@@ -1916,7 +1891,7 @@ where
None => {}
Some(inp) => {
for (a, b) in self.min.iter_mut().zip(inp.iter()) {
if *b < *a {
if *b < *a || a.is_nan() {
*a = *b;
}
}
@@ -1926,7 +1901,7 @@ where
None => {}
Some(inp) => {
for (a, b) in self.max.iter_mut().zip(inp.iter()) {
if *b > *a {
if *b > *a || a.is_nan() {
*a = *b;
}
}

2
disk/src/binned/dim1.rs
View File

@@ -17,7 +17,7 @@ use std::fmt;
use std::marker::PhantomData;
use tokio::fs::File;
#[derive(Clone, Serialize, Deserialize)]
#[derive(Serialize, Deserialize)]
pub struct MinMaxAvgDim1Bins<NTY> {
pub ts1s: Vec<u64>,
pub ts2s: Vec<u64>,

5
disk/src/binned/query.rs
View File

@@ -311,6 +311,7 @@ fn binning_scheme_string(agg_kind: &AggKind) -> String {
fn agg_kind_from_binning_scheme(params: &BTreeMap<String, String>) -> Result<AggKind, Error> {
let key = "binningScheme";
let tok1 = "binnedXcount";
let s = params
.get(key)
.map_or(Err(Error::with_msg(format!("can not find {}", key))), |k| Ok(k))?;
@@ -318,8 +319,8 @@ fn agg_kind_from_binning_scheme(params: &BTreeMap<String, String>) -> Result<Agg
AggKind::Plain
} else if s == "toScalarX" {
AggKind::DimXBins1
} else if s.starts_with("binnedXcount") {
AggKind::DimXBinsN(s[12..].parse()?)
} else if s.starts_with(tok1) {
AggKind::DimXBinsN(s[tok1.len()..].parse()?)
} else {
return Err(Error::with_msg("can not extract binningScheme"));
};

96
disk/src/channelexec.rs
View File

@@ -1,5 +1,5 @@
use crate::agg::binnedt::TimeBinnableType;
use crate::agg::enp::{Identity, WavePlainProc};
use crate::agg::enp::Identity;
use crate::agg::streams::{Collectable, Collector, StreamItem};
use crate::binned::{EventsNodeProcessor, NumOps, PushableIndex, RangeCompletableItem};
use crate::decode::{
@@ -15,10 +15,12 @@ use err::Error;
use futures_core::Stream;
use futures_util::future::FutureExt;
use futures_util::StreamExt;
use netpod::log::*;
use netpod::{AggKind, ByteOrder, Channel, NanoRange, NodeConfigCached, PerfOpts, ScalarType, Shape};
use parse::channelconfig::{extract_matching_config_entry, read_local_config, MatchingConfigEntry};
use serde::de::DeserializeOwned;
use serde_json::Value as JsonValue;
use std::fmt::Debug;
use std::pin::Pin;
use std::time::Duration;
use tokio::time::timeout_at;
@@ -36,14 +38,14 @@ pub trait ChannelExecFunction {
where
NTY: NumOps + NumFromBytes<NTY, END> + 'static,
END: Endianness + 'static,
EVS: EventValueShape<NTY, END> + EventValueFromBytes<NTY, END> + PlainEventsAggMethod + 'static,
EVS: EventValueShape<NTY, END> + EventValueFromBytes<NTY, END> + 'static,
ENP: EventsNodeProcessor<Input = <EVS as EventValueFromBytes<NTY, END>>::Output> + 'static,
Sitemty<<<EVS as PlainEventsAggMethod>::Method as EventsNodeProcessor>::Output>: FrameType,
<<EVS as PlainEventsAggMethod>::Method as EventsNodeProcessor>::Output: Collectable + PushableIndex,
<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output:
TimeBinnableType<Output = <<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output> + Unpin,
// TODO require these things in general?
<ENP as EventsNodeProcessor>::Output: PushableIndex,
<ENP as EventsNodeProcessor>::Output: Debug + Collectable + PushableIndex,
<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output: Debug
+ TimeBinnableType<Output = <<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output>
+ Collectable
+ Unpin,
Sitemty<<ENP as EventsNodeProcessor>::Output>: FrameType + Framable + 'static,
Sitemty<<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output>:
FrameType + Framable + DeserializeOwned;
@@ -62,21 +64,17 @@ where
F: ChannelExecFunction,
NTY: NumOps + NumFromBytes<NTY, END> + 'static,
END: Endianness + 'static,
// TODO
// TODO
// TODO
// TODO
// Can I replace the PlainEventsAggMethod by EventsNodeProcessor?
EVS: EventValueShape<NTY, END> + EventValueFromBytes<NTY, END> + PlainEventsAggMethod + 'static,
EVS: EventValueShape<NTY, END> + EventValueFromBytes<NTY, END> + 'static,
ENP: EventsNodeProcessor<Input = <EVS as EventValueFromBytes<NTY, END>>::Output> + 'static,
Sitemty<<<EVS as PlainEventsAggMethod>::Method as EventsNodeProcessor>::Output>: FrameType,
<<EVS as PlainEventsAggMethod>::Method as EventsNodeProcessor>::Output: Collectable + PushableIndex,
// TODO require these things in general?
<ENP as EventsNodeProcessor>::Output: Debug + Collectable + PushableIndex,
<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output: Debug
+ TimeBinnableType<Output = <<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output>
+ Collectable
+ Unpin,
Sitemty<<ENP as EventsNodeProcessor>::Output>: FrameType + Framable + 'static,
Sitemty<<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output>:
FrameType + Framable + DeserializeOwned,
{
Ok(f.exec(byte_order, shape, event_value_shape, events_node_proc)?)
}
@@ -238,9 +236,9 @@ impl ChannelExecFunction for PlainEvents {
fn exec<NTY, END, EVS, ENP>(
self,
byte_order: END,
shape: Shape,
_shape: Shape,
event_value_shape: EVS,
events_node_proc: ENP,
_events_node_proc: ENP,
) -> Result<Self::Output, Error>
where
NTY: NumOps + NumFromBytes<NTY, END> + 'static,
@@ -294,16 +292,20 @@ impl PlainEventsJson {
}
}
pub async fn collect_plain_events_json<T, S>(stream: S, timeout: Duration) -> Result<JsonValue, Error>
pub async fn collect_plain_events_json<T, S>(
stream: S,
timeout: Duration,
bin_count_exp: u32,
) -> Result<JsonValue, Error>
where
S: Stream<Item = Sitemty<T>> + Unpin,
T: Collectable,
T: Collectable + Debug,
{
let deadline = tokio::time::Instant::now() + timeout;
// TODO in general a Collector does not need to know about the expected number of bins.
// It would make more sense for some specific Collector kind to know.
// Therefore introduce finer grained types.
let mut collector = <T as Collectable>::new_collector(0);
let mut collector = <T as Collectable>::new_collector(bin_count_exp);
let mut i1 = 0;
let mut stream = stream;
loop {
@@ -333,6 +335,7 @@ where
collector.set_range_complete();
}
RangeCompletableItem::Data(item) => {
info!("collect_plain_events_json GOT ITEM {:?}", item);
collector.ingest(&item);
i1 += 1;
}
@@ -351,41 +354,30 @@ where
Ok(ret)
}
pub trait PlainEventsAggMethod {
type Method: EventsNodeProcessor;
}
impl<NTY> PlainEventsAggMethod for EventValuesDim0Case<NTY>
where
NTY: NumOps,
{
type Method = Identity<NTY>;
}
impl<NTY> PlainEventsAggMethod for EventValuesDim1Case<NTY>
where
NTY: NumOps,
{
type Method = WavePlainProc<NTY>;
}
impl ChannelExecFunction for PlainEventsJson {
type Output = Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send>>;
fn exec<NTY, END, EVS, ENP>(
self,
byte_order: END,
shape: Shape,
_shape: Shape,
event_value_shape: EVS,
_events_node_proc: ENP,
) -> Result<Self::Output, Error>
where
NTY: NumOps + NumFromBytes<NTY, END> + 'static,
END: Endianness + 'static,
EVS: EventValueShape<NTY, END> + EventValueFromBytes<NTY, END> + PlainEventsAggMethod + 'static,
EVS: EventValueShape<NTY, END> + EventValueFromBytes<NTY, END> + 'static,
ENP: EventsNodeProcessor<Input = <EVS as EventValueFromBytes<NTY, END>>::Output> + 'static,
Sitemty<<<EVS as PlainEventsAggMethod>::Method as EventsNodeProcessor>::Output>: FrameType,
<<EVS as PlainEventsAggMethod>::Method as EventsNodeProcessor>::Output: Collectable + PushableIndex,
// TODO require these things in general?
<ENP as EventsNodeProcessor>::Output: Debug + Collectable + PushableIndex,
<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output: Debug
+ TimeBinnableType<Output = <<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output>
+ Collectable
+ Unpin,
Sitemty<<ENP as EventsNodeProcessor>::Output>: FrameType + Framable + 'static,
Sitemty<<<ENP as EventsNodeProcessor>::Output as TimeBinnableType>::Output>:
FrameType + Framable + DeserializeOwned,
{
let _ = byte_order;
let _ = event_value_shape;
@@ -395,12 +387,8 @@ impl ChannelExecFunction for PlainEventsJson {
range: self.range,
agg_kind: self.agg_kind,
};
let s = MergedFromRemotes::<<EVS as PlainEventsAggMethod>::Method>::new(
evq,
perf_opts,
self.node_config.node_config.cluster,
);
let f = collect_plain_events_json(s, self.timeout);
let s = MergedFromRemotes::<ENP>::new(evq, perf_opts, self.node_config.node_config.cluster);
let f = collect_plain_events_json(s, self.timeout, 0);
let f = FutureExt::map(f, |item| match item {
Ok(item) => {
// TODO add channel entry info here?

20
disk/src/frame/makeframe.rs
View File

@@ -2,6 +2,7 @@ use crate::agg::enp::{WaveEvents, XBinnedScalarEvents, XBinnedWaveEvents};
use crate::agg::eventbatch::MinMaxAvgScalarEventBatch;
use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
use crate::agg::streams::StreamItem;
use crate::binned::dim1::MinMaxAvgDim1Bins;
use crate::binned::{MinMaxAvgBins, MinMaxAvgWaveBins, NumOps, RangeCompletableItem};
use crate::decode::EventValues;
use crate::frame::inmem::InMemoryFrame;
@@ -118,6 +119,13 @@ where
const FRAME_TYPE_ID: u32 = 0xa00 + NTY::SUB;
}
impl<NTY> FrameType for Sitemty<MinMaxAvgDim1Bins<NTY>>
where
NTY: SubFrId,
{
const FRAME_TYPE_ID: u32 = 0xb00 + NTY::SUB;
}
pub trait ProvidesFrameType {
fn frame_type_id(&self) -> u32;
}
@@ -226,6 +234,18 @@ where
}
}
impl<NTY> Framable for Sitemty<MinMaxAvgDim1Bins<NTY>>
where
NTY: NumOps + Serialize,
{
fn typeid(&self) -> u32 {
Self::FRAME_TYPE_ID
}
fn make_frame(&self) -> Result<BytesMut, Error> {
make_frame(self)
}
}
pub fn make_frame<FT>(item: &FT) -> Result<BytesMut, Error>
where
FT: FrameType + Serialize,

4
disk/src/gen.rs
View File

@@ -370,7 +370,7 @@ async fn gen_event(
let ele_size = 8;
let mut vals = vec![0; (ele_size * ele_count) as usize];
for i1 in 0..ele_count {
let v = evix as f64;
let v = (evix as f64) * 100.0 + i1 as f64;
let a = if config.byte_order.is_be() {
v.to_be_bytes()
} else {
@@ -393,7 +393,7 @@ async fn gen_event(
let ele_size = 2;
let mut vals = vec![0; (ele_size * ele_count) as usize];
for i1 in 0..ele_count {
let v = evix as u16;
let v = (evix as u16).wrapping_mul(100).wrapping_add(i1 as u16);
let a = if config.byte_order.is_be() {
v.to_be_bytes()
} else {