From c9f39d55743a63742c8e04742c89130253fbfe9a Mon Sep 17 00:00:00 2001 From: Dominik Werder Date: Fri, 3 Feb 2023 17:46:01 +0100 Subject: [PATCH] Remove stats agg kind --- daqbufp2/src/nodes.rs | 2 +- disk/src/agg/enp.rs | 41 +--- disk/src/decode.rs | 50 ++++- disk/src/raw/conn.rs | 5 - httpret/src/proxy.rs | 57 +++-- httpret/src/proxy/api1.rs | 60 ++++++ items/src/items.rs | 1 - items/src/statsevents.rs | 434 -------------------------------------- items/src/waveevents.rs | 5 +- netpod/src/netpod.rs | 9 - netpod/src/query.rs | 5 - 11 files changed, 158 insertions(+), 511 deletions(-) create mode 100644 httpret/src/proxy/api1.rs delete mode 100644 items/src/statsevents.rs diff --git a/daqbufp2/src/nodes.rs b/daqbufp2/src/nodes.rs index 8b3c98e..e4855cf 100644 --- a/daqbufp2/src/nodes.rs +++ b/daqbufp2/src/nodes.rs @@ -3,7 +3,7 @@ use err::Error; use netpod::log::*; use netpod::Cluster; use std::sync::{Arc, Mutex}; -use std::{thread}; +use std::thread; use std::time::Duration; use tokio::task::JoinHandle; diff --git a/disk/src/agg/enp.rs b/disk/src/agg/enp.rs index cfc77d6..c4fee51 100644 --- a/disk/src/agg/enp.rs +++ b/disk/src/agg/enp.rs @@ -1,8 +1,8 @@ +use items::numops::NumOps; use items::scalarevents::ScalarEvents; -use items::waveevents::WaveEvents; use items::EventsNodeProcessor; -use items::{numops::NumOps, statsevents::StatsEvents}; -use netpod::{AggKind, Shape}; +use netpod::AggKind; +use netpod::Shape; use std::marker::PhantomData; pub struct Identity { @@ -24,38 +24,3 @@ where inp } } - -pub struct Stats1Scalar {} - -impl EventsNodeProcessor for Stats1Scalar { - type Input = StatsEvents; - type Output = StatsEvents; - - fn create(_shape: Shape, _agg_kind: AggKind) -> Self { - Self {} - } - - fn process(&self, inp: Self::Input) -> Self::Output { - inp - } -} - -pub struct Stats1Wave { - _m1: PhantomData, -} - -impl EventsNodeProcessor for Stats1Wave -where - NTY: NumOps, -{ - type Input = WaveEvents; - type Output = StatsEvents; - - fn create(_shape: Shape, _agg_kind: AggKind) -> Self { - Self { _m1: PhantomData } - } - - fn process(&self, _inp: Self::Input) -> Self::Output { - err::todoval() - } -} diff --git a/disk/src/decode.rs b/disk/src/decode.rs index c0876be..79b0870 100644 --- a/disk/src/decode.rs +++ b/disk/src/decode.rs @@ -170,7 +170,6 @@ where type NumXAggToSingleBin: EventsNodeProcessor>::Batch>; type NumXAggToNBins: EventsNodeProcessor>::Batch>; type NumXAggPlain: EventsNodeProcessor>::Batch>; - type NumXAggToStats1: EventsNodeProcessor>::Batch>; } pub struct EventValuesDim0Case { @@ -191,7 +190,6 @@ where // TODO is this sufficient? type NumXAggToNBins = Identity; type NumXAggPlain = Identity; - type NumXAggToStats1 = Identity; } pub struct EventValuesDim1Case { @@ -212,7 +210,6 @@ where type NumXAggToSingleBin = WaveXBinner; type NumXAggToNBins = WaveNBinner; type NumXAggPlain = WavePlainProc; - type NumXAggToStats1 = crate::agg::enp::Stats1Wave; } pub struct EventsDecodedStream @@ -341,6 +338,53 @@ where } } +pub struct EventsDynStream { + events_full: EventChunkerMultifile, + done: bool, + complete: bool, +} + +impl EventsDynStream { + pub fn new(events_full: EventChunkerMultifile) -> Self { + Self { + events_full, + done: false, + complete: false, + } + } +} + +impl Stream for EventsDynStream { + type Item = Sitemty>; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll> { + use Poll::*; + loop { + break if self.complete { + panic!("poll_next on complete") + } else if self.done { + self.complete = true; + Ready(None) + } else { + match self.events_full.poll_next_unpin(cx) { + Ready(Some(Ok(k))) => { + todo!() + } + Ready(Some(Err(e))) => { + self.done = true; + Ready(Some(Err(e))) + } + Ready(None) => { + self.done = true; + continue; + } + Pending => Pending, + } + }; + } + } +} + pub struct EventsItemStream { inp: Pin>>>, done: bool, diff --git a/disk/src/raw/conn.rs b/disk/src/raw/conn.rs index 9f7be4b..f12df8f 100644 --- a/disk/src/raw/conn.rs +++ b/disk/src/raw/conn.rs @@ -135,11 +135,6 @@ macro_rules! pipe4 { <$evs<$nty> as EventValueShape<$nty, $end>>::NumXAggPlain::create($shape, $agg_kind), $event_blobs, ), - AggKind::Stats1 => make_num_pipeline_stream_evs::<$nty, $end, $evs<$nty>, _>( - $evsv, - <$evs<$nty> as EventValueShape<$nty, $end>>::NumXAggToStats1::create($shape, $agg_kind), - $event_blobs, - ), } }; } diff --git a/httpret/src/proxy.rs b/httpret/src/proxy.rs index c63011a..953470d 100644 --- a/httpret/src/proxy.rs +++ b/httpret/src/proxy.rs @@ -1,29 +1,58 @@ +pub mod api1; pub mod api4; -use crate::api1::{channel_search_configs_v1, channel_search_list_v1, gather_json_2_v1}; +use crate::api1::channel_search_configs_v1; +use crate::api1::channel_search_list_v1; +use crate::api1::gather_json_2_v1; +use crate::api_1_docs; +use crate::api_4_docs; use crate::err::Error; -use crate::gather::{gather_get_json_generic, SubRes}; +use crate::gather::gather_get_json_generic; +use crate::gather::SubRes; use crate::pulsemap::MapPulseQuery; -use crate::{api_1_docs, api_4_docs, response, response_err, Cont, ReqCtx, PSI_DAQBUFFER_SERVICE_MARK}; -use futures_util::{pin_mut, Stream}; -use http::{Method, StatusCode}; -use hyper::service::{make_service_fn, service_fn}; -use hyper::{Body, Request, Response, Server}; +use crate::response; +use crate::response_err; +use crate::Cont; +use crate::ReqCtx; +use crate::PSI_DAQBUFFER_SERVICE_MARK; +use futures_util::pin_mut; +use futures_util::Stream; +use http::Method; +use http::StatusCode; +use hyper::service::make_service_fn; +use hyper::service::service_fn; +use hyper::Body; +use hyper::Request; +use hyper::Response; +use hyper::Server; use itertools::Itertools; use netpod::log::*; -use netpod::query::{BinnedQuery, ChannelStateEventsQuery, PlainEventsQuery}; -use netpod::{AppendToUrl, ChannelConfigQuery, FromUrl, HasBackend, HasTimeout, ProxyConfig}; -use netpod::{ChannelSearchQuery, ChannelSearchResult, ChannelSearchSingleResult}; -use netpod::{ACCEPT_ALL, APP_JSON}; -use serde::{Deserialize, Serialize}; +use netpod::query::BinnedQuery; +use netpod::query::ChannelStateEventsQuery; +use netpod::query::PlainEventsQuery; +use netpod::AppendToUrl; +use netpod::ChannelConfigQuery; +use netpod::ChannelSearchQuery; +use netpod::ChannelSearchResult; +use netpod::ChannelSearchSingleResult; +use netpod::FromUrl; +use netpod::HasBackend; +use netpod::HasTimeout; +use netpod::ProxyConfig; +use netpod::ACCEPT_ALL; +use netpod::APP_JSON; +use serde::Deserialize; +use serde::Serialize; use serde_json::Value as JsonValue; use std::future::Future; use std::net::SocketAddr; use std::pin::Pin; -use std::task::{Context, Poll}; +use std::task::Context; +use std::task::Poll; use std::time::Duration; use tokio::fs::File; -use tokio::io::{AsyncRead, ReadBuf}; +use tokio::io::AsyncRead; +use tokio::io::ReadBuf; use url::Url; const DISTRI_PRE: &str = "/distri/"; diff --git a/httpret/src/proxy/api1.rs b/httpret/src/proxy/api1.rs new file mode 100644 index 0000000..ac02bdf --- /dev/null +++ b/httpret/src/proxy/api1.rs @@ -0,0 +1,60 @@ +use crate::bodystream::response; +use crate::err::Error; +use crate::ReqCtx; +use http::Method; +use http::Request; +use http::Response; +use http::StatusCode; +use hyper::Body; +use netpod::log::*; +use netpod::query::api1::Api1Query; +use netpod::ProxyConfig; +use netpod::ACCEPT_ALL; + +pub struct PythonDataApi1Query {} + +impl PythonDataApi1Query { + pub fn path() -> &'static str { + "/api/1/query" + } + + pub fn handler(req: &Request) -> Option { + if req.uri().path() == Self::path() { + Some(Self {}) + } else { + None + } + } + + pub async fn handle( + &self, + req: Request, + ctx: &ReqCtx, + proxy_config: &ProxyConfig, + ) -> Result, Error> { + if req.method() != Method::POST { + return Ok(response(StatusCode::METHOD_NOT_ALLOWED).body(Body::empty())?); + } + let (head, body) = req.into_parts(); + let accept = head + .headers + .get(http::header::ACCEPT) + .map_or(Ok(ACCEPT_ALL), |k| k.to_str()) + .map_err(|e| Error::with_msg_no_trace(format!("{e:?}")))? + .to_owned(); + let body_data = hyper::body::to_bytes(body).await?; + if body_data.len() < 512 && body_data.first() == Some(&"{".as_bytes()[0]) { + info!("request body_data string: {}", String::from_utf8_lossy(&body_data)); + } + let qu = match serde_json::from_slice::(&body_data) { + Ok(qu) => qu, + Err(e) => { + error!("got body_data: {:?}", String::from_utf8_lossy(&body_data[..])); + error!("can not parse: {e}"); + return Err(Error::with_msg_no_trace("can not parse query")); + } + }; + info!("Proxy sees request: {qu:?}"); + Ok(response(StatusCode::OK).body(Body::empty())?) + } +} diff --git a/items/src/items.rs b/items/src/items.rs index 8288da9..0c7b195 100644 --- a/items/src/items.rs +++ b/items/src/items.rs @@ -8,7 +8,6 @@ pub mod inmem; pub mod numops; pub mod plainevents; pub mod scalarevents; -pub mod statsevents; pub mod streams; pub mod waveevents; pub mod xbinnedscalarevents; diff --git a/items/src/statsevents.rs b/items/src/statsevents.rs deleted file mode 100644 index cfd07f8..0000000 --- a/items/src/statsevents.rs +++ /dev/null @@ -1,434 +0,0 @@ -use crate::streams::{Collectable, Collector}; -use crate::{ - ts_offs_from_abs, Appendable, ByteEstimate, Clearable, EventAppendable, EventsNodeProcessorOutput, - FilterFittingInside, Fits, FitsInside, FrameType, FrameTypeInnerStatic, NewEmpty, PushableIndex, RangeOverlapInfo, - ReadPbv, ReadableFromFile, TimeBinnableType, TimeBinnableTypeAggregator, WithLen, WithTimestamps, -}; -use err::Error; -use netpod::log::*; -use netpod::{NanoRange, Shape}; -use serde::{Deserialize, Serialize}; -use std::any::Any; -use std::collections::VecDeque; -use std::fmt; -use tokio::fs::File; - -#[derive(Serialize, Deserialize)] -pub struct StatsEvents { - pub tss: Vec, - pub pulses: Vec, -} - -impl FrameTypeInnerStatic for StatsEvents { - const FRAME_TYPE_ID: u32 = crate::STATS_EVENTS_FRAME_TYPE_ID; -} - -impl FrameType for StatsEvents { - fn frame_type_id(&self) -> u32 { - ::FRAME_TYPE_ID - } -} - -impl StatsEvents { - pub fn empty() -> Self { - Self { - tss: vec![], - pulses: vec![], - } - } -} - -impl fmt::Debug for StatsEvents { - fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { - write!( - fmt, - "count {} tss {:?} .. {:?} pulses {:?} .. {:?}", - self.tss.len(), - self.tss.first(), - self.tss.last(), - self.pulses.first(), - self.pulses.last(), - ) - } -} - -impl WithLen for StatsEvents { - fn len(&self) -> usize { - self.tss.len() - } -} - -impl WithTimestamps for StatsEvents { - fn ts(&self, ix: usize) -> u64 { - self.tss[ix] - } -} - -impl ByteEstimate for StatsEvents { - fn byte_estimate(&self) -> u64 { - if self.tss.len() == 0 { - 0 - } else { - // TODO improve via a const fn on NTY - self.tss.len() as u64 * 16 - } - } -} - -impl RangeOverlapInfo for StatsEvents { - fn ends_before(&self, range: NanoRange) -> bool { - match self.tss.last() { - Some(&ts) => ts < range.beg, - None => true, - } - } - - fn ends_after(&self, range: NanoRange) -> bool { - match self.tss.last() { - Some(&ts) => ts >= range.end, - None => panic!(), - } - } - - fn starts_after(&self, range: NanoRange) -> bool { - match self.tss.first() { - Some(&ts) => ts >= range.end, - None => panic!(), - } - } -} - -impl FitsInside for StatsEvents { - fn fits_inside(&self, range: NanoRange) -> Fits { - if self.tss.is_empty() { - Fits::Empty - } else { - let t1 = *self.tss.first().unwrap(); - let t2 = *self.tss.last().unwrap(); - if t2 < range.beg { - Fits::Lower - } else if t1 > range.end { - Fits::Greater - } else if t1 < range.beg && t2 > range.end { - Fits::PartlyLowerAndGreater - } else if t1 < range.beg { - Fits::PartlyLower - } else if t2 > range.end { - Fits::PartlyGreater - } else { - Fits::Inside - } - } - } -} - -impl FilterFittingInside for StatsEvents { - fn filter_fitting_inside(self, fit_range: NanoRange) -> Option { - match self.fits_inside(fit_range) { - Fits::Inside | Fits::PartlyGreater | Fits::PartlyLower | Fits::PartlyLowerAndGreater => Some(self), - _ => None, - } - } -} - -impl PushableIndex for StatsEvents { - fn push_index(&mut self, src: &Self, ix: usize) { - self.tss.push(src.tss[ix]); - self.pulses.push(src.pulses[ix]); - } -} - -impl NewEmpty for StatsEvents { - fn empty(_shape: Shape) -> Self { - Self { - tss: Vec::new(), - pulses: Vec::new(), - } - } -} - -impl Appendable for StatsEvents { - fn empty_like_self(&self) -> Self { - Self::empty() - } - - fn append(&mut self, src: &Self) { - self.tss.extend_from_slice(&src.tss); - self.pulses.extend_from_slice(&src.pulses); - } - - fn append_zero(&mut self, ts1: u64, _ts2: u64) { - self.tss.push(ts1); - self.pulses.push(0); - } -} - -impl Clearable for StatsEvents { - fn clear(&mut self) { - self.tss.clear(); - self.pulses.clear(); - } -} - -impl ReadableFromFile for StatsEvents { - fn read_from_file(_file: File) -> Result, Error> { - // TODO refactor types such that this can be removed. - panic!() - } - - fn from_buf(_buf: &[u8]) -> Result { - panic!() - } -} - -impl TimeBinnableType for StatsEvents { - type Output = StatsEvents; - type Aggregator = StatsEventsAggregator; - - fn aggregator(range: NanoRange, _x_bin_count: usize, do_time_weight: bool) -> Self::Aggregator { - Self::Aggregator::new(range, do_time_weight) - } -} - -pub struct StatsEventsCollector { - vals: StatsEvents, - range_complete: bool, - timed_out: bool, -} - -impl StatsEventsCollector { - pub fn new() -> Self { - Self { - vals: StatsEvents::empty(), - range_complete: false, - timed_out: false, - } - } -} - -impl WithLen for StatsEventsCollector { - fn len(&self) -> usize { - self.vals.tss.len() - } -} - -#[derive(Serialize)] -pub struct StatsEventsCollectorOutput { - #[serde(rename = "tsAnchor")] - ts_anchor_sec: u64, - #[serde(rename = "tsMs")] - ts_off_ms: Vec, - #[serde(rename = "tsNs")] - ts_off_ns: Vec, - // TODO what to collect? pulse min/max - #[serde(skip_serializing_if = "crate::bool_is_false", rename = "rangeFinal")] - range_complete: bool, - #[serde(skip_serializing_if = "crate::bool_is_false", rename = "timedOut")] - timed_out: bool, -} - -impl Collector for StatsEventsCollector { - type Input = StatsEvents; - type Output = StatsEventsCollectorOutput; - - fn ingest(&mut self, src: &Self::Input) { - self.vals.append(src); - } - - fn set_range_complete(&mut self) { - self.range_complete = true; - } - - fn set_timed_out(&mut self) { - self.timed_out = true; - } - - fn result(self) -> Result { - 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, - range_complete: self.range_complete, - timed_out: self.timed_out, - }; - Ok(ret) - } -} - -impl Collectable for StatsEvents { - type Collector = StatsEventsCollector; - - fn new_collector(_bin_count_exp: u32) -> Self::Collector { - Self::Collector::new() - } -} - -pub struct StatsEventsAggregator { - range: NanoRange, - count: u64, - sumc: u64, - sum: f32, - int_ts: u64, - last_ts: u64, - do_time_weight: bool, -} - -impl StatsEventsAggregator { - pub fn new(range: NanoRange, do_time_weight: bool) -> Self { - let int_ts = range.beg; - Self { - range, - count: 0, - sum: 0f32, - sumc: 0, - int_ts, - last_ts: 0, - do_time_weight, - } - } - - fn apply_min_max(&mut self, _val: f32) { - // TODO currently no values to min/max - } - - fn apply_event_unweight(&mut self, val: f32) { - self.apply_min_max(val); - let vf = val; - if vf.is_nan() { - } else { - self.sum += vf; - self.sumc += 1; - } - } - - fn apply_event_time_weight(&mut self, _ts: u64) { - // TODO currently no value to weight. - } - - fn ingest_unweight(&mut self, item: &::Input) { - for i1 in 0..item.tss.len() { - let ts = item.tss[i1]; - let val = 0.0; - if ts < self.range.beg { - } else if ts >= self.range.end { - } else { - self.apply_event_unweight(val); - self.count += 1; - } - } - } - - fn ingest_time_weight(&mut self, item: &::Input) { - for i1 in 0..item.tss.len() { - let ts = item.tss[i1]; - //let val = 0.0; - if ts < self.int_ts { - self.last_ts = ts; - //self.last_val = Some(val); - } else if ts >= self.range.end { - return; - } else { - self.apply_event_time_weight(ts); - self.count += 1; - self.last_ts = ts; - //self.last_val = Some(val); - } - } - } - - fn reset(&mut self, range: NanoRange) { - self.int_ts = range.beg; - self.range = range; - self.count = 0; - //self.min = None; - //self.max = None; - self.sum = 0f32; - self.sumc = 0; - } - - fn result_reset_unweight( - &mut self, - range: NanoRange, - _expand: bool, - ) -> ::Output { - let _avg = if self.sumc == 0 { - None - } else { - Some(self.sum / self.sumc as f32) - }; - // TODO return some meaningful value - let ret = StatsEvents::empty(); - self.reset(range); - ret - } - - fn result_reset_time_weight( - &mut self, - range: NanoRange, - expand: bool, - ) -> ::Output { - // TODO check callsite for correct expand status. - if true || expand { - debug!("result_reset_time_weight calls apply_event_time_weight"); - self.apply_event_time_weight(self.range.end); - } else { - debug!("result_reset_time_weight NO EXPAND"); - } - let _avg = { - let sc = self.range.delta() as f32 * 1e-9; - Some(self.sum / sc) - }; - // TODO return some meaningful value - let ret = StatsEvents::empty(); - self.reset(range); - ret - } -} - -impl TimeBinnableTypeAggregator for StatsEventsAggregator { - type Input = StatsEvents; - type Output = StatsEvents; - - fn range(&self) -> &NanoRange { - &self.range - } - - fn ingest(&mut self, item: &Self::Input) { - if self.do_time_weight { - self.ingest_time_weight(item) - } else { - self.ingest_unweight(item) - } - } - - fn result_reset(&mut self, range: NanoRange, expand: bool) -> Self::Output { - if self.do_time_weight { - self.result_reset_time_weight(range, expand) - } else { - self.result_reset_unweight(range, expand) - } - } -} - -impl EventAppendable for StatsEvents { - type Value = f32; - - fn append_event(ret: Option, _ts: u64, _pulse: u64, _value: Self::Value) -> Self { - let ret = if let Some(ret) = ret { ret } else { Self::empty() }; - // TODO - error!("TODO statsevents append_event"); - err::todo(); - ret - } -} - -impl EventsNodeProcessorOutput for StatsEvents { - fn as_any_mut(&mut self) -> &mut dyn Any { - self - } - - fn into_parts(self) -> (Box, VecDeque, VecDeque) { - todo!() - } -} diff --git a/items/src/waveevents.rs b/items/src/waveevents.rs index a55545b..52bb2d1 100644 --- a/items/src/waveevents.rs +++ b/items/src/waveevents.rs @@ -68,7 +68,10 @@ where } } -impl AsAnyRef for WaveEvents where NTY:NumOps { +impl AsAnyRef for WaveEvents +where + NTY: NumOps, +{ fn as_any_ref(&self) -> &dyn Any { self } diff --git a/netpod/src/netpod.rs b/netpod/src/netpod.rs index 65f1b97..a5d903b 100644 --- a/netpod/src/netpod.rs +++ b/netpod/src/netpod.rs @@ -1564,7 +1564,6 @@ pub enum AggKind { DimXBinsN(u32), Plain, TimeWeightedScalar, - Stats1, } impl AggKind { @@ -1575,7 +1574,6 @@ impl AggKind { Self::DimXBins1 => false, Self::DimXBinsN(_) => false, Self::Plain => false, - Self::Stats1 => false, } } @@ -1586,7 +1584,6 @@ impl AggKind { Self::DimXBins1 => false, Self::DimXBinsN(_) => false, Self::Plain => false, - Self::Stats1 => false, } } } @@ -1612,7 +1609,6 @@ pub fn x_bin_count(shape: &Shape, agg_kind: &AggKind) -> usize { Shape::Wave(n) => *n as usize, Shape::Image(j, k) => *j as usize * *k as usize, }, - AggKind::Stats1 => 0, } } @@ -1634,9 +1630,6 @@ impl fmt::Display for AggKind { Self::TimeWeightedScalar => { write!(fmt, "TimeWeightedScalar") } - Self::Stats1 => { - write!(fmt, "Stats1") - } } } } @@ -1658,8 +1651,6 @@ impl FromStr for AggKind { Ok(AggKind::DimXBins1) } else if s == "TimeWeightedScalar" { Ok(AggKind::TimeWeightedScalar) - } else if s == "Stats1" { - Ok(AggKind::Stats1) } else if s.starts_with(nmark) { let nbins: u32 = s[nmark.len()..].parse()?; Ok(AggKind::DimXBinsN(nbins)) diff --git a/netpod/src/query.rs b/netpod/src/query.rs index 60b484e..b842ea5 100644 --- a/netpod/src/query.rs +++ b/netpod/src/query.rs @@ -512,9 +512,6 @@ pub fn binning_scheme_append_to_url(agg_kind: &AggKind, url: &mut Url) { g.append_pair("binningScheme", "binnedX"); g.append_pair("binnedXcount", &format!("{}", n)); } - AggKind::Stats1 => { - g.append_pair("binningScheme", "stats1"); - } } } @@ -534,8 +531,6 @@ pub fn agg_kind_from_binning_scheme(pairs: &BTreeMap) -> Result< } else if s == "binnedX" { let u = pairs.get("binnedXcount").map_or("1", |k| k).parse()?; AggKind::DimXBinsN(u) - } else if s == "stats1" { - AggKind::Stats1 } else { return Err(Error::with_msg("can not extract binningScheme")); };