WIP on adding stats to the streams

dbconn/src/lib.rs

@@ -1,13 +1,14 @@
 use err::Error;
 use netpod::log::*;
-use netpod::{Channel, NodeConfig};
+use netpod::{Channel, NodeConfigCached};
 use tokio_postgres::NoTls;
 
-pub async fn channel_exists(channel: &Channel, node_config: &NodeConfig) -> Result<bool, Error> {
+pub async fn channel_exists(channel: &Channel, node_config: &NodeConfigCached) -> Result<bool, Error> {
-    let d = &node_config.cluster.database;
+    let d = &node_config.node_config.cluster.database;
     let uri = format!("postgresql://{}:{}@{}:{}/{}", d.user, d.pass, d.host, 5432, d.name);
     let (cl, conn) = tokio_postgres::connect(&uri, NoTls).await?;
-    let cjh = tokio::spawn(async move {
+    // TODO monitor connection drop.
+    let _cjh = tokio::spawn(async move {
         if let Err(e) = conn.await {
             error!("connection error: {}", e);
         }
@@ -25,6 +26,5 @@ pub async fn channel_exists(channel: &Channel, node_config: &NodeConfig) -> Resu
             row.get::<_, i64>(0)
         );
     }
-    drop(cjh);
     Ok(true)
 }

disk/src/agg.rs

@@ -7,10 +7,11 @@ use crate::agg::eventbatch::MinMaxAvgScalarEventBatch;
 use err::Error;
 use futures_core::Stream;
 use futures_util::StreamExt;
-use netpod::NanoRange;
+use netpod::{EventDataReadStats, NanoRange};
 use netpod::{Node, ScalarType};
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use std::time::{Duration, Instant};
 #[allow(unused_imports)]
 use tracing::{debug, error, info, span, trace, warn, Level};
 
@@ -97,10 +98,28 @@ impl AggregatableXdim1Bin for ValuesDim1 {
     }
 }
 
+pub struct ValuesDim1ExtractStats {
+    pub dur: Duration,
+}
+
+impl ValuesDim1ExtractStats {
+    pub fn new() -> Self {
+        Self {
+            dur: Duration::default(),
+        }
+    }
+    pub fn trans(self: &mut Self, k: &mut Self) {
+        self.dur += k.dur;
+        k.dur = Duration::default();
+    }
+}
+
 /// Batch of events with a numeric one-dimensional (i.e. array) value.
 pub struct ValuesDim1 {
     pub tss: Vec<u64>,
     pub values: Vec<Vec<f32>>,
+    pub event_data_read_stats: EventDataReadStats,
+    pub values_dim_1_extract_stats: ValuesDim1ExtractStats,
 }
 
 impl ValuesDim1 {
@@ -108,6 +127,8 @@ impl ValuesDim1 {
         Self {
             tss: vec![],
             values: vec![],
+            event_data_read_stats: EventDataReadStats::new(),
+            values_dim_1_extract_stats: ValuesDim1ExtractStats::new(),
         }
     }
 }
@@ -216,7 +237,8 @@ where
         }
         match self.inp.poll_next_unpin(cx) {
             Ready(Some(Ok(k))) => {
-                let mut ret = ValuesDim1 {
+                // TODO implement here for dim-0
+                let mut ret = ValuesDim0 {
                     tss: vec![],
                     values: vec![],
                 };
@@ -328,10 +350,8 @@ where
         }
         match self.inp.poll_next_unpin(cx) {
             Ready(Some(Ok(k))) => {
-                let mut ret = ValuesDim1 {
+                let inst1 = Instant::now();
-                    tss: vec![],
+                let mut ret = ValuesDim1::empty();
-                    values: vec![],
-                };
                 use ScalarType::*;
                 for i1 in 0..k.tss.len() {
                     // TODO iterate sibling arrays after single bounds check
@@ -389,6 +409,10 @@ where
                         }
                     }
                 }
+                let inst2 = Instant::now();
+                let mut k = k;
+                ret.event_data_read_stats.trans(&mut k.event_data_read_stats);
+                ret.values_dim_1_extract_stats.dur += inst2.duration_since(inst1);
                 Ready(Some(Ok(ret)))
             }
             Ready(Some(Err(e))) => {

disk/src/aggtest.rs

@@ -1,8 +1,7 @@
-use super::agg::{AggregatableXdim1Bin, IntoDim1F32Stream, ValuesDim1};
+use super::agg::{make_test_node, IntoDim1F32Stream};
 use super::merge::MergeDim1F32Stream;
 use crate::agg::binnedt::IntoBinnedT;
 use crate::agg::binnedx::IntoBinnedXBins1;
-use crate::agg::make_test_node;
 use futures_util::StreamExt;
 use netpod::timeunits::*;
 use netpod::{BinnedRange, Channel, ChannelConfig, NanoRange, Nanos, ScalarType, Shape};
@@ -194,15 +193,10 @@ async fn merge_0_inner() {
 }
 
 pub fn tmp_some_older_things() {
-    let vals = ValuesDim1 {
-        tss: vec![0, 1, 2, 3],
-        values: vec![vec![0., 0., 0.], vec![1., 1., 1.], vec![2., 2., 2.], vec![3., 3., 3.]],
-    };
     // I want to distinguish already in the outer part between dim-0 and dim-1 and generate
     // separate code for these cases...
     // That means that also the reading chain itself needs to be typed on that.
     // Need to supply some event-payload converter type which has that type as Output type.
-    let _vals2 = vals.into_agg();
     // Now the T-binning:
 
     /*

disk/src/binnedstream.rs

@@ -1,11 +1,11 @@
 use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
-use crate::cache::pbvfs::PreBinnedValueFetchedStream;
+use crate::cache::pbvfs::{PreBinnedItem, PreBinnedValueFetchedStream};
 use err::Error;
 use futures_core::Stream;
 use futures_util::StreamExt;
 #[allow(unused_imports)]
 use netpod::log::*;
-use netpod::{AggKind, BinnedRange, Channel, NodeConfig, PreBinnedPatchIterator};
+use netpod::{AggKind, BinnedRange, Channel, NodeConfigCached, PreBinnedPatchIterator};
 use std::future::ready;
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -20,7 +20,7 @@ impl BinnedStream {
         channel: Channel,
         range: BinnedRange,
         agg_kind: AggKind,
-        node_config: &NodeConfig,
+        node_config: &NodeConfigCached,
     ) -> Self {
         let patches: Vec<_> = patch_it.collect();
         warn!("BinnedStream::new");
@@ -36,10 +36,10 @@ impl BinnedStream {
             .flatten()
             .filter_map({
                 let range = range.clone();
-                move |k: Result<MinMaxAvgScalarBinBatch, Error>| {
+                move |k| {
                     let fit_range = range.full_range();
-                    let g = match k {
+                    let g = match k.0 {
-                        Ok(k) => {
+                        Ok(PreBinnedItem::Batch(k)) => {
                             use super::agg::{Fits, FitsInside};
                             match k.fits_inside(fit_range) {
                                 Fits::Inside

disk/src/cache.rs

@@ -12,7 +12,7 @@ use futures_core::Stream;
 use futures_util::{pin_mut, StreamExt};
 use hyper::Response;
 use netpod::{
-    AggKind, BinnedRange, Channel, Cluster, NanoRange, Node, NodeConfig, PreBinnedPatchCoord, PreBinnedPatchIterator,
+    AggKind, BinnedRange, Channel, Cluster, NanoRange, NodeConfigCached, PreBinnedPatchCoord, PreBinnedPatchIterator,
     PreBinnedPatchRange, ToNanos,
 };
 use serde::{Deserialize, Serialize};
@@ -65,12 +65,11 @@ impl Query {
 }
 
 pub async fn binned_bytes_for_http(
-    node_config: &NodeConfig,
+    node_config: &NodeConfigCached,
-    node: &Node,
     query: &Query,
 ) -> Result<BinnedBytesForHttpStream, Error> {
     let range = &query.range;
-    let channel_config = read_local_config(&query.channel, node).await?;
+    let channel_config = read_local_config(&query.channel, &node_config.node).await?;
     let entry = extract_matching_config_entry(range, &channel_config);
     info!("found config entry {:?}", entry);
     let range = BinnedRange::covering_range(range.clone(), query.count)
@@ -180,11 +179,10 @@ impl PreBinnedQuery {
 // A user must first make sure that the grid spec is valid, and that this node is responsible for it.
 // Otherwise it is an error.
 pub fn pre_binned_bytes_for_http(
-    node_config: &NodeConfig,
+    node_config: &NodeConfigCached,
-    node: &Node,
     query: &PreBinnedQuery,
 ) -> Result<PreBinnedValueByteStream, Error> {
-    info!("pre_binned_bytes_for_http  {:?}  {:?}", query, node);
+    info!("pre_binned_bytes_for_http  {:?}  {:?}", query, node_config.node);
     let ret = super::cache::pbv::pre_binned_value_byte_stream_new(
         query.patch.clone(),
         query.channel.clone(),

disk/src/cache/pbv.rs

@@ -1,6 +1,5 @@
 use crate::agg::binnedt::IntoBinnedT;
-use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
+use crate::cache::pbvfs::{PreBinnedFrame, PreBinnedItem, PreBinnedValueFetchedStream};
-use crate::cache::pbvfs::{PreBinnedHttpFrame, PreBinnedValueFetchedStream};
 use crate::cache::{node_ix_for_patch, MergedFromRemotes};
 use crate::frame::makeframe::make_frame;
 use crate::raw::EventsQuery;
@@ -11,7 +10,7 @@ use futures_util::{FutureExt, StreamExt};
 use netpod::log::*;
 use netpod::streamext::SCC;
 use netpod::{
-    AggKind, BinnedRange, Channel, NanoRange, NodeConfig, PreBinnedPatchCoord, PreBinnedPatchIterator,
+    AggKind, BinnedRange, Channel, NanoRange, NodeConfigCached, PreBinnedPatchCoord, PreBinnedPatchIterator,
     PreBinnedPatchRange,
 };
 use std::future::Future;
@@ -28,7 +27,7 @@ pub fn pre_binned_value_byte_stream_new(
     patch: PreBinnedPatchCoord,
     channel: Channel,
     agg_kind: AggKind,
-    node_config: &NodeConfig,
+    node_config: &NodeConfigCached,
 ) -> PreBinnedValueByteStream {
     let s1 = PreBinnedValueStream::new(patch, channel, agg_kind, node_config);
     let s2 = PreBinnedValueByteStreamInner { inp: s1 };
@@ -41,7 +40,7 @@ impl Stream for PreBinnedValueByteStreamInner {
     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)) => match make_frame::<PreBinnedHttpFrame>(&item) {
+            Ready(Some(item)) => match make_frame::<PreBinnedFrame>(&item) {
                 Ok(buf) => Ready(Some(Ok(buf.freeze()))),
                 Err(e) => Ready(Some(Err(e.into()))),
             },
@@ -55,9 +54,9 @@ pub struct PreBinnedValueStream {
     patch_coord: PreBinnedPatchCoord,
     channel: Channel,
     agg_kind: AggKind,
-    node_config: NodeConfig,
+    node_config: NodeConfigCached,
     open_check_local_file: Option<Pin<Box<dyn Future<Output = Result<tokio::fs::File, std::io::Error>> + Send>>>,
-    fut2: Option<Pin<Box<dyn Stream<Item = Result<MinMaxAvgScalarBinBatch, Error>> + Send>>>,
+    fut2: Option<Pin<Box<dyn Stream<Item = PreBinnedFrame> + Send>>>,
     errored: bool,
     completed: bool,
 }
@@ -67,10 +66,10 @@ impl PreBinnedValueStream {
         patch_coord: PreBinnedPatchCoord,
         channel: Channel,
         agg_kind: AggKind,
-        node_config: &NodeConfig,
+        node_config: &NodeConfigCached,
     ) -> Self {
         // TODO check that we are the correct node.
-        let _node_ix = node_ix_for_patch(&patch_coord, &channel, &node_config.cluster);
+        let _node_ix = node_ix_for_patch(&patch_coord, &channel, &node_config.node_config.cluster);
         Self {
             patch_coord,
             channel,
@@ -129,7 +128,7 @@ impl PreBinnedValueStream {
                 error!("try_setup_fetch_prebinned_higher_res  apply all requested transformations and T-binning");
                 let count = self.patch_coord.patch_t_len() / self.patch_coord.bin_t_len();
                 let range = BinnedRange::covering_range(evq.range.clone(), count).unwrap();
-                let s1 = MergedFromRemotes::new(evq, self.node_config.cluster.clone());
+                let s1 = MergedFromRemotes::new(evq, self.node_config.node_config.cluster.clone());
                 let s2 = s1
                     .map(|k| {
                         if k.is_err() {
@@ -139,7 +138,11 @@ impl PreBinnedValueStream {
                         }
                         k
                     })
-                    .into_binned_t(range);
+                    .into_binned_t(range)
+                    .map(|k| match k {
+                        Ok(k) => PreBinnedFrame(Ok(PreBinnedItem::Batch(k))),
+                        Err(e) => PreBinnedFrame(Err(e)),
+                    });
                 self.fut2 = Some(Box::pin(s2));
             }
         }
@@ -148,7 +151,7 @@ impl PreBinnedValueStream {
 
 impl Stream for PreBinnedValueStream {
     // TODO need this generic for scalar and array (when wave is not binned down to a single scalar point)
-    type Item = Result<MinMaxAvgScalarBinBatch, Error>;
+    type Item = PreBinnedFrame;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
@@ -162,31 +165,33 @@ impl Stream for PreBinnedValueStream {
         'outer: loop {
             break if let Some(fut) = self.fut2.as_mut() {
                 match fut.poll_next_unpin(cx) {
-                    Ready(Some(Ok(k))) => Ready(Some(Ok(k))),
+                    Ready(Some(k)) => match k.0 {
-                    Ready(Some(Err(e))) => {
+                        Ok(k) => Ready(Some(PreBinnedFrame(Ok(k)))),
-                        self.errored = true;
+                        Err(e) => {
-                        Ready(Some(Err(e)))
+                            self.errored = true;
-                    }
+                            Ready(Some(PreBinnedFrame(Err(e))))
+                        }
+                    },
                     Ready(None) => Ready(None),
                     Pending => Pending,
                 }
             } else if let Some(fut) = self.open_check_local_file.as_mut() {
                 match fut.poll_unpin(cx) {
                     Ready(Ok(_file)) => {
+                        let e = Err(Error::with_msg(format!("TODO use the cached data from file")));
+                        let e = PreBinnedFrame(e);
                         self.errored = true;
-                        Ready(Some(Err(Error::with_msg(format!(
+                        Ready(Some(e))
-                            "TODO use the cached data from file"
-                        )))))
                     }
                     Ready(Err(e)) => match e.kind() {
                         std::io::ErrorKind::NotFound => {
                             error!("TODO LOCAL CACHE FILE NOT FOUND");
                             self.try_setup_fetch_prebinned_higher_res();
                             if self.fut2.is_none() {
+                                let e = Err(Error::with_msg(format!("try_setup_fetch_prebinned_higher_res  failed")));
+                                let e = PreBinnedFrame(e);
                                 self.errored = true;
-                                Ready(Some(Err(Error::with_msg(format!(
+                                Ready(Some(e))
-                                    "try_setup_fetch_prebinned_higher_res  failed"
-                                )))))
                             } else {
                                 continue 'outer;
                             }
@@ -194,7 +199,8 @@ impl Stream for PreBinnedValueStream {
                         _ => {
                             error!("File I/O error: {:?}", e);
                             self.errored = true;
-                            Ready(Some(Err(e.into())))
+                            let e = PreBinnedFrame(Err(e.into()));
+                            Ready(Some(e))
                         }
                     },
                     Pending => Pending,

disk/src/cache/pbvfs.rs

@@ -7,7 +7,8 @@ use futures_core::Stream;
 use futures_util::{pin_mut, FutureExt};
 #[allow(unused_imports)]
 use netpod::log::*;
-use netpod::{AggKind, Channel, NodeConfig, PreBinnedPatchCoord};
+use netpod::{AggKind, Channel, NodeConfigCached, PreBinnedPatchCoord};
+use serde::{Deserialize, Serialize};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
@@ -24,10 +25,10 @@ impl PreBinnedValueFetchedStream {
         patch_coord: PreBinnedPatchCoord,
         channel: Channel,
         agg_kind: AggKind,
-        node_config: &NodeConfig,
+        node_config: &NodeConfigCached,
     ) -> Self {
-        let nodeix = node_ix_for_patch(&patch_coord, &channel, &node_config.cluster);
+        let nodeix = node_ix_for_patch(&patch_coord, &channel, &node_config.node_config.cluster);
-        let node = &node_config.cluster.nodes[nodeix as usize];
+        let node = &node_config.node_config.cluster.nodes[nodeix as usize];
         warn!("TODO defining property of a PreBinnedPatchCoord? patchlen + ix? binsize + patchix? binsize + patchsize + patchix?");
         // TODO encapsulate uri creation, how to express aggregation kind?
         let uri: hyper::Uri = format!(
@@ -52,13 +53,26 @@ impl PreBinnedValueFetchedStream {
     }
 }
 
-// TODO use a newtype here to use a different FRAME_TYPE_ID compared to
+#[derive(Serialize, Deserialize)]
-// impl FrameType for BinnedBytesForHttpStreamFrame
+pub struct PreBinnedFrame(pub Result<PreBinnedItem, Error>);
-pub type PreBinnedHttpFrame = Result<MinMaxAvgScalarBinBatch, Error>;
+
+impl<T> From<T> for PreBinnedFrame
+where
+    T: Into<Error>,
+{
+    fn from(k: T) -> Self {
+        PreBinnedFrame(Err(k.into()))
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+pub enum PreBinnedItem {
+    Batch(MinMaxAvgScalarBinBatch),
+}
 
 impl Stream for PreBinnedValueFetchedStream {
     // TODO need this generic for scalar and array (when wave is not binned down to a single scalar point)
-    type Item = PreBinnedHttpFrame;
+    type Item = PreBinnedFrame;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
@@ -73,16 +87,19 @@ impl Stream for PreBinnedValueFetchedStream {
             break if let Some(res) = self.res.as_mut() {
                 pin_mut!(res);
                 match res.poll_next(cx) {
-                    Ready(Some(Ok(frame))) => match decode_frame::<PreBinnedHttpFrame>(&frame) {
+                    Ready(Some(Ok(frame))) => match decode_frame::<PreBinnedFrame>(&frame) {
-                        Ok(item) => Ready(Some(item)),
+                        Ok(item) => match item.0 {
+                            Ok(item) => Ready(Some(PreBinnedFrame(Ok(item)))),
+                            Err(e) => Ready(Some(PreBinnedFrame(Err(e)))),
+                        },
                         Err(e) => {
                             self.errored = true;
-                            Ready(Some(Err(e.into())))
+                            Ready(Some(e.into()))
                         }
                     },
                     Ready(Some(Err(e))) => {
                         self.errored = true;
-                        Ready(Some(Err(e.into())))
+                        Ready(Some(e.into()))
                     }
                     Ready(None) => {
                         self.completed = true;
@@ -103,20 +120,28 @@ impl Stream for PreBinnedValueFetchedStream {
                         Err(e) => {
                             error!("PreBinnedValueStream  error in stream {:?}", e);
                             self.errored = true;
-                            Ready(Some(Err(e.into())))
+                            Ready(Some(PreBinnedFrame(Err(e.into()))))
                         }
                     },
                     Pending => Pending,
                 }
             } else {
-                let req = hyper::Request::builder()
+                match hyper::Request::builder()
                     .method(http::Method::GET)
                     .uri(&self.uri)
-                    .body(hyper::Body::empty())?;
+                    .body(hyper::Body::empty())
-                let client = hyper::Client::new();
+                {
-                info!("PreBinnedValueFetchedStream  START REQUEST FOR {:?}", req);
+                    Ok(req) => {
-                self.resfut = Some(client.request(req));
+                        let client = hyper::Client::new();
-                continue 'outer;
+                        info!("PreBinnedValueFetchedStream  START REQUEST FOR {:?}", req);
+                        self.resfut = Some(client.request(req));
+                        continue 'outer;
+                    }
+                    Err(e) => {
+                        self.errored = true;
+                        Ready(Some(e.into()))
+                    }
+                }
             };
         }
     }

disk/src/eventchunker.rs

@@ -1,19 +1,18 @@
-use crate::NeedMinBuffer;
+use crate::{FileChunkRead, NeedMinBuffer};
 use bitshuffle::bitshuffle_decompress;
 use bytes::{Buf, BytesMut};
 use err::Error;
 use futures_core::Stream;
-use futures_util::pin_mut;
+use futures_util::StreamExt;
 #[allow(unused_imports)]
 use netpod::log::*;
 use netpod::timeunits::SEC;
-use netpod::{ChannelConfig, NanoRange, ScalarType, Shape};
+use netpod::{ChannelConfig, EventDataReadStats, NanoRange, ScalarType, Shape};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
 pub struct EventChunker {
     inp: NeedMinBuffer,
-    polled: u32,
     state: DataFileState,
     need_min: u32,
     channel_config: ChannelConfig,
@@ -30,7 +29,7 @@ enum DataFileState {
 
 impl EventChunker {
     pub fn from_start(
-        inp: Pin<Box<dyn Stream<Item = Result<BytesMut, Error>> + Send>>,
+        inp: Pin<Box<dyn Stream<Item = Result<FileChunkRead, Error>> + Send>>,
         channel_config: ChannelConfig,
         range: NanoRange,
     ) -> Self {
@@ -38,7 +37,6 @@ impl EventChunker {
         inp.set_need_min(6);
         Self {
             inp,
-            polled: 0,
             state: DataFileState::FileHeader,
             need_min: 6,
             channel_config,
@@ -50,7 +48,7 @@ impl EventChunker {
     }
 
     pub fn from_event_boundary(
-        inp: Pin<Box<dyn Stream<Item = Result<BytesMut, Error>> + Send>>,
+        inp: Pin<Box<dyn Stream<Item = Result<FileChunkRead, Error>> + Send>>,
         channel_config: ChannelConfig,
         range: NanoRange,
     ) -> Self {
@@ -58,7 +56,6 @@ impl EventChunker {
         inp.set_need_min(4);
         Self {
             inp,
-            polled: 0,
             state: DataFileState::Event,
             need_min: 4,
             channel_config,
@@ -103,6 +100,7 @@ impl EventChunker {
                         self.state = DataFileState::Event;
                         self.need_min = 4;
                         buf.advance(totlen);
+                        ret.event_data_read_stats.parsed_bytes += totlen as u64;
                     }
                 }
                 DataFileState::Event => {
@@ -110,24 +108,12 @@ impl EventChunker {
                     let len = sl.read_i32::<BE>().unwrap();
                     assert!(len >= 20 && len < 1024 * 1024 * 10);
                     let len = len as u32;
-                    trace!(
-                        "+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-  event len {}",
-                        len,
-                    );
                     if (buf.len() as u32) < 20 {
                         // TODO gather stats about how often we find not enough input
                         //info!("parse_buf not enough B");
                         self.need_min = len as u32;
                         break;
                     } else if (buf.len() as u32) < len {
-                        {
-                            // TODO this is just for testing
-                            let mut sl = std::io::Cursor::new(buf.as_ref());
-                            sl.read_i32::<BE>().unwrap();
-                            sl.read_i64::<BE>().unwrap();
-                            let ts = sl.read_i64::<BE>().unwrap();
-                            trace!("parse_buf not enough C   len {}  have {}  ts {}", len, buf.len(), ts);
-                        }
                         self.need_min = len as u32;
                         break;
                     } else {
@@ -239,6 +225,7 @@ impl EventChunker {
                         }
                         trace!("advance and reset need_min");
                         buf.advance(len as usize);
+                        ret.event_data_read_stats.parsed_bytes += len as u64;
                         self.need_min = 4;
                     }
                 }
@@ -289,25 +276,15 @@ impl Stream for EventChunker {
             self.completed = true;
             return Ready(None);
         }
-        self.polled += 1;
+        match self.inp.poll_next_unpin(cx) {
-        if self.polled >= 20000 {
+            Ready(Some(Ok(mut fcr))) => {
-            warn!("EventChunker poll limit reached");
+                trace!("EventChunker got buffer  len {}", fcr.buf.len());
-            self.errored = true;
+                let r = self.parse_buf(&mut fcr.buf);
-            return Poll::Ready(Some(Err(Error::with_msg(format!("EventChunker poll limit reached")))));
-        }
-        let g = &mut self.inp;
-        pin_mut!(g);
-        trace!("EventChunker  call input poll_next");
-        match g.poll_next(cx) {
-            Ready(Some(Ok(mut buf))) => {
-                trace!("EventChunker got buffer  len {}", buf.len());
-                let r = self.parse_buf(&mut buf);
                 match r {
                     Ok(res) => {
-                        if buf.len() > 0 {
+                        if fcr.buf.len() > 0 {
                             // TODO gather stats about this:
-                            //info!("parse_buf returned {} leftover bytes to me", buf.len());
+                            self.inp.put_back(fcr);
-                            self.inp.put_back(buf);
                         }
                         if self.need_min > 1024 * 8 {
                             let msg = format!("spurious EventChunker asks for need_min {}", self.need_min);
@@ -345,6 +322,7 @@ pub struct EventFull {
     pub pulses: Vec<u64>,
     pub decomps: Vec<Option<BytesMut>>,
     pub scalar_types: Vec<ScalarType>,
+    pub event_data_read_stats: EventDataReadStats,
 }
 
 impl EventFull {
@@ -354,6 +332,7 @@ impl EventFull {
             pulses: vec![],
             decomps: vec![],
             scalar_types: vec![],
+            event_data_read_stats: EventDataReadStats::new(),
         }
     }
 

disk/src/frame/makeframe.rs

@@ -1,3 +1,4 @@
+use crate::cache::pbvfs::PreBinnedFrame;
 use crate::cache::BinnedBytesForHttpStreamFrame;
 use crate::frame::inmem::InMemoryFrame;
 use crate::raw::conn::RawConnOut;
@@ -26,6 +27,10 @@ impl FrameType for RawConnOut {
     const FRAME_TYPE_ID: u32 = 0x04;
 }
 
+impl FrameType for PreBinnedFrame {
+    const FRAME_TYPE_ID: u32 = 0x05;
+}
+
 pub fn make_frame<FT>(item: &FT) -> Result<BytesMut, Error>
 where
     FT: FrameType + Serialize,

disk/src/lib.rs

@@ -10,6 +10,7 @@ use std::future::Future;
 use std::path::PathBuf;
 use std::pin::Pin;
 use std::task::{Context, Poll};
+use std::time::{Duration, Instant};
 use tokio::fs::{File, OpenOptions};
 use tokio::io::AsyncRead;
 #[allow(unused_imports)]
@@ -300,21 +301,32 @@ fn unused_raw_concat_channel_read_stream_file_pipe(
     }
 }
 
+pub struct FileChunkRead {
+    buf: BytesMut,
+    duration: Duration,
+}
+
 pub fn file_content_stream(
     mut file: tokio::fs::File,
     buffer_size: usize,
-) -> impl Stream<Item = Result<BytesMut, Error>> + Send {
+) -> impl Stream<Item = Result<FileChunkRead, Error>> + Send {
     async_stream::stream! {
         use tokio::io::AsyncReadExt;
         loop {
             let mut buf = BytesMut::with_capacity(buffer_size);
+            let inst1 = Instant::now();
             let n1 = file.read_buf(&mut buf).await?;
+            let inst2 = Instant::now();
             if n1 == 0 {
                 info!("file EOF");
                 break;
             }
             else {
-                yield Ok(buf);
+                let ret = FileChunkRead {
+                    buf,
+                    duration: inst2.duration_since(inst1),
+                };
+                yield Ok(ret);
             }
         }
     }
@@ -357,22 +369,27 @@ pub fn parsed1(query: &netpod::AggQuerySingleChannel, node: &Node) -> impl Strea
         }
     }
 }
+
 pub struct NeedMinBuffer {
-    inp: Pin<Box<dyn Stream<Item = Result<BytesMut, Error>> + Send>>,
+    inp: Pin<Box<dyn Stream<Item = Result<FileChunkRead, Error>> + Send>>,
     need_min: u32,
-    left: Option<BytesMut>,
+    left: Option<FileChunkRead>,
+    errored: bool,
+    completed: bool,
 }
 
 impl NeedMinBuffer {
-    pub fn new(inp: Pin<Box<dyn Stream<Item = Result<BytesMut, Error>> + Send>>) -> Self {
+    pub fn new(inp: Pin<Box<dyn Stream<Item = Result<FileChunkRead, Error>> + Send>>) -> Self {
         Self {
             inp: inp,
             need_min: 1,
             left: None,
+            errored: false,
+            completed: false,
         }
     }
 
-    pub fn put_back(&mut self, buf: BytesMut) {
+    pub fn put_back(&mut self, buf: FileChunkRead) {
         assert!(self.left.is_none());
         self.left = Some(buf);
     }
@@ -383,46 +400,56 @@ impl NeedMinBuffer {
 }
 
 impl Stream for NeedMinBuffer {
-    type Item = Result<BytesMut, Error>;
+    type Item = Result<FileChunkRead, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        use Poll::*;
+        if self.completed {
+            panic!("NeedMinBuffer  poll_next on completed");
+        }
+        if self.errored {
+            self.completed = true;
+            return Ready(None);
+        }
         loop {
             let mut again = false;
             let g = &mut self.inp;
             pin_mut!(g);
             let z = match g.poll_next(cx) {
-                Poll::Ready(Some(Ok(buf))) => {
+                Ready(Some(Ok(fcr))) => {
                     //info!("NeedMin got buf  len {}", buf.len());
                     match self.left.take() {
-                        Some(mut left) => {
+                        Some(mut lfcr) => {
-                            left.unsplit(buf);
+                            // TODO measure:
-                            let buf = left;
+                            lfcr.buf.unsplit(fcr.buf);
-                            if buf.len() as u32 >= self.need_min {
+                            lfcr.duration += fcr.duration;
+                            let fcr = lfcr;
+                            if fcr.buf.len() as u32 >= self.need_min {
                                 //info!("with left ready  len {}  need_min {}", buf.len(), self.need_min);
-                                Poll::Ready(Some(Ok(buf)))
+                                Ready(Some(Ok(fcr)))
                             } else {
                                 //info!("with left not enough  len {}  need_min {}", buf.len(), self.need_min);
-                                self.left.replace(buf);
+                                self.left.replace(fcr);
                                 again = true;
-                                Poll::Pending
+                                Pending
                             }
                         }
                         None => {
-                            if buf.len() as u32 >= self.need_min {
+                            if fcr.buf.len() as u32 >= self.need_min {
                                 //info!("simply ready  len {}  need_min {}", buf.len(), self.need_min);
-                                Poll::Ready(Some(Ok(buf)))
+                                Ready(Some(Ok(fcr)))
                             } else {
                                 //info!("no previous leftover, need more  len {}  need_min {}", buf.len(), self.need_min);
-                                self.left.replace(buf);
+                                self.left.replace(fcr);
                                 again = true;
-                                Poll::Pending
+                                Pending
                             }
                         }
                     }
                 }
-                Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e.into()))),
+                Ready(Some(Err(e))) => Ready(Some(Err(e.into()))),
-                Poll::Ready(None) => Poll::Ready(None),
+                Ready(None) => Ready(None),
-                Poll::Pending => Poll::Pending,
+                Pending => Pending,
             };
             if !again {
                 break z;

disk/src/raw/conn.rs

@@ -11,36 +11,30 @@ use futures_util::StreamExt;
 #[allow(unused_imports)]
 use netpod::log::*;
 use netpod::timeunits::SEC;
-use netpod::{Node, NodeConfig, Shape};
+use netpod::{NodeConfigCached, Shape};
 use std::net::SocketAddr;
 use tokio::io::AsyncWriteExt;
 use tokio::net::tcp::OwnedWriteHalf;
 use tokio::net::TcpStream;
 use tracing::Instrument;
 
-pub async fn raw_service(node_config: NodeConfig, node: Node) -> Result<(), Error> {
+pub async fn raw_service(node_config: NodeConfigCached) -> Result<(), Error> {
-    let addr = format!("{}:{}", node.listen, node.port_raw);
+    let addr = format!("{}:{}", node_config.node.listen, node_config.node.port_raw);
     let lis = tokio::net::TcpListener::bind(addr).await?;
     loop {
         match lis.accept().await {
             Ok((stream, addr)) => {
-                let node = node.clone();
+                taskrun::spawn(raw_conn_handler(stream, addr, node_config.clone()));
-                taskrun::spawn(raw_conn_handler(stream, addr, node_config.clone(), node));
             }
             Err(e) => Err(e)?,
         }
     }
 }
 
-async fn raw_conn_handler(
+async fn raw_conn_handler(stream: TcpStream, addr: SocketAddr, node_config: NodeConfigCached) -> Result<(), Error> {
-    stream: TcpStream,
-    addr: SocketAddr,
-    node_config: NodeConfig,
-    node: Node,
-) -> Result<(), Error> {
     //use tracing_futures::Instrument;
     let span1 = span!(Level::INFO, "raw::raw_conn_handler");
-    let r = raw_conn_handler_inner(stream, addr, &node_config, &node)
+    let r = raw_conn_handler_inner(stream, addr, &node_config)
         .instrument(span1)
         .await;
     match r {
@@ -57,10 +51,9 @@ pub type RawConnOut = Result<MinMaxAvgScalarEventBatch, Error>;
 async fn raw_conn_handler_inner(
     stream: TcpStream,
     addr: SocketAddr,
-    node_config: &NodeConfig,
+    node_config: &NodeConfigCached,
-    node: &Node,
 ) -> Result<(), Error> {
-    match raw_conn_handler_inner_try(stream, addr, node_config, node).await {
+    match raw_conn_handler_inner_try(stream, addr, node_config).await {
         Ok(_) => (),
         Err(mut ce) => {
             /*error!(
@@ -94,8 +87,7 @@ impl<E: Into<Error>> From<(E, OwnedWriteHalf)> for ConnErr {
 async fn raw_conn_handler_inner_try(
     stream: TcpStream,
     addr: SocketAddr,
-    node_config: &NodeConfig,
+    node_config: &NodeConfigCached,
-    node: &Node,
 ) -> Result<(), ConnErr> {
     info!("raw_conn_handler   SPAWNED   for {:?}", addr);
     let (netin, mut netout) = stream.into_split();
@@ -139,7 +131,7 @@ async fn raw_conn_handler_inner_try(
     }
     debug!("REQUEST  {:?}", evq);
     let range = &evq.range;
-    let channel_config = match read_local_config(&evq.channel, node).await {
+    let channel_config = match read_local_config(&evq.channel, &node_config.node).await {
         Ok(k) => k,
         Err(e) => return Err((e, netout))?,
     };
@@ -180,16 +172,21 @@ async fn raw_conn_handler_inner_try(
         buffer_size: 1024 * 4,
     };
     let buffer_size = 1024 * 4;
-    let mut s1 = EventBlobsComplete::new(range.clone(), query.channel_config.clone(), node.clone(), buffer_size)
+    let mut s1 = EventBlobsComplete::new(
-        .into_dim_1_f32_stream()
+        range.clone(),
-        .into_binned_x_bins_1();
+        query.channel_config.clone(),
+        node_config.node.clone(),
+        buffer_size,
+    )
+    .into_dim_1_f32_stream()
+    .into_binned_x_bins_1();
     let mut e = 0;
     while let Some(item) = s1.next().await {
         if let Ok(k) = &item {
             e += 1;
             trace!(
                 "emit items  sp {:2}  e {:3}  len {:3}  {:10?}  {:10?}",
-                node.split,
+                node_config.node.split,
                 e,
                 k.tss.len(),
                 k.tss.first().map(|k| k / SEC),

httpret/src/lib.rs

@@ -9,7 +9,7 @@ use http::{HeaderMap, Method, StatusCode};
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{server::Server, Body, Request, Response};
 use net::SocketAddr;
-use netpod::{Node, NodeConfig};
+use netpod::{Node, NodeConfigCached};
 use panic::{AssertUnwindSafe, UnwindSafe};
 use pin::Pin;
 use std::{future, net, panic, pin, task};
@@ -18,21 +18,19 @@ use tracing::field::Empty;
 #[allow(unused_imports)]
 use tracing::{debug, error, info, span, trace, warn, Level};
 
-pub async fn host(node_config: NodeConfig, node: Node) -> Result<(), Error> {
+pub async fn host(node_config: NodeConfigCached) -> Result<(), Error> {
     let node_config = node_config.clone();
-    let node = node.clone();
+    let rawjh = taskrun::spawn(raw_service(node_config.clone()));
-    let rawjh = taskrun::spawn(raw_service(node_config.clone(), node.clone()));
     use std::str::FromStr;
-    let addr = SocketAddr::from_str(&format!("{}:{}", node.listen, node.port))?;
+    let addr = SocketAddr::from_str(&format!("{}:{}", node_config.node.listen, node_config.node.port))?;
     let make_service = make_service_fn({
         move |conn| {
             info!("new raw {:?}", conn);
             let node_config = node_config.clone();
-            let node = node.clone();
             async move {
                 Ok::<_, Error>(service_fn({
                     move |req| {
-                        let f = data_api_proxy(req, node_config.clone(), node.clone());
+                        let f = data_api_proxy(req, node_config.clone());
                         Cont { f: Box::pin(f) }
                     }
                 }))
@@ -44,8 +42,8 @@ pub async fn host(node_config: NodeConfig, node: Node) -> Result<(), Error> {
     Ok(())
 }
 
-async fn data_api_proxy(req: Request<Body>, node_config: NodeConfig, node: Node) -> Result<Response<Body>, Error> {
+async fn data_api_proxy(req: Request<Body>, node_config: NodeConfigCached) -> Result<Response<Body>, Error> {
-    match data_api_proxy_try(req, &node_config, &node).await {
+    match data_api_proxy_try(req, &node_config).await {
         Ok(k) => Ok(k),
         Err(e) => {
             error!("{:?}", e);
@@ -84,28 +82,24 @@ where
 
 impl<F> UnwindSafe for Cont<F> {}
 
-async fn data_api_proxy_try(
+async fn data_api_proxy_try(req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
-    req: Request<Body>,
-    node_config: &NodeConfig,
-    node: &Node,
-) -> Result<Response<Body>, Error> {
     let uri = req.uri().clone();
     let path = uri.path();
     if path == "/api/1/parsed_raw" {
         if req.method() == Method::POST {
-            Ok(parsed_raw(req, node).await?)
+            Ok(parsed_raw(req, &node_config.node).await?)
         } else {
             Ok(response(StatusCode::METHOD_NOT_ALLOWED).body(Body::empty())?)
         }
     } else if path == "/api/1/binned" {
         if req.method() == Method::GET {
-            Ok(binned(req, &node_config, node).await?)
+            Ok(binned(req, node_config).await?)
         } else {
             Ok(response(StatusCode::METHOD_NOT_ALLOWED).body(Body::empty())?)
         }
     } else if path == "/api/1/prebinned" {
         if req.method() == Method::GET {
-            Ok(prebinned(req, &node_config, node).await?)
+            Ok(prebinned(req, &node_config).await?)
         } else {
             Ok(response(StatusCode::METHOD_NOT_ALLOWED).body(Body::empty())?)
         }
@@ -224,10 +218,10 @@ where
     }
 }
 
-async fn binned(req: Request<Body>, node_config: &NodeConfig, node: &Node) -> Result<Response<Body>, Error> {
+async fn binned(req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
     let (head, _body) = req.into_parts();
     let query = disk::cache::Query::from_request(&head)?;
-    let ret = match disk::cache::binned_bytes_for_http(node_config, node, &query).await {
+    let ret = match disk::cache::binned_bytes_for_http(node_config, &query).await {
         Ok(s) => response(StatusCode::OK).body(BodyStream::wrapped(s, format!("desc-BINNED")))?,
         Err(e) => {
             error!("{:?}", e);
@@ -237,14 +231,14 @@ async fn binned(req: Request<Body>, node_config: &NodeConfig, node: &Node) -> Re
     Ok(ret)
 }
 
-async fn prebinned(req: Request<Body>, node_config: &NodeConfig, node: &Node) -> Result<Response<Body>, Error> {
+async fn prebinned(req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
     let (head, _body) = req.into_parts();
     let q = PreBinnedQuery::from_request(&head)?;
     let desc = format!("pre-b-{}", q.patch.bin_t_len() / 1000000000);
     let span1 = span!(Level::INFO, "httpret::prebinned", desc = &desc.as_str());
     span1.in_scope(|| {
         trace!("prebinned");
-        let ret = match disk::cache::pre_binned_bytes_for_http(node_config, node, &q) {
+        let ret = match disk::cache::pre_binned_bytes_for_http(node_config, &q) {
             Ok(s) => response(StatusCode::OK).body(BodyStream::wrapped(
                 s,
                 format!(

netpod/src/lib.rs

@@ -107,12 +107,6 @@ pub struct Node {
     pub ksprefix: String,
 }
 
-impl Node {
-    pub fn _name(&self) -> String {
-        format!("{}-{}", self.host, self.port)
-    }
-}
-
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Database {
     pub name: String,
@@ -152,6 +146,29 @@ impl NodeConfig {
     }
 }
 
+#[derive(Clone, Debug)]
+pub struct NodeConfigCached {
+    pub node_config: NodeConfig,
+    pub node: Node,
+    pub ix: usize,
+}
+
+impl From<NodeConfig> for Result<NodeConfigCached, Error> {
+    fn from(k: NodeConfig) -> Self {
+        match k.get_node() {
+            Some(node) => {
+                let ret = NodeConfigCached {
+                    node: node.clone(),
+                    node_config: k,
+                    ix: 0,
+                };
+                Ok(ret)
+            }
+            None => Err(Error::with_msg(format!("can not find node {:?}", k))),
+        }
+    }
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Channel {
     pub backend: String,
@@ -623,3 +640,18 @@ pub mod log {
     #[allow(unused_imports)]
     pub use tracing::{debug, error, info, span, trace, warn, Level};
 }
+
+#[derive(Serialize, Deserialize)]
+pub struct EventDataReadStats {
+    pub parsed_bytes: u64,
+}
+
+impl EventDataReadStats {
+    pub fn new() -> Self {
+        Self { parsed_bytes: 0 }
+    }
+    pub fn trans(&mut self, k: &mut Self) {
+        self.parsed_bytes += k.parsed_bytes;
+        k.parsed_bytes = 0;
+    }
+}

retrieval/src/bin/retrieval.rs

@@ -1,5 +1,5 @@
 use err::Error;
-use netpod::NodeConfig;
+use netpod::{NodeConfig, NodeConfigCached};
 use tokio::io::AsyncReadExt;
 #[allow(unused_imports)]
 use tracing::{debug, error, info, trace, warn};
@@ -28,10 +28,9 @@ async fn go() -> Result<(), Error> {
             let mut buf = vec![];
             config_file.read_to_end(&mut buf).await?;
             let node_config: NodeConfig = serde_json::from_slice(&buf)?;
-            let node = node_config
+            let node_config: Result<NodeConfigCached, Error> = node_config.into();
-                .get_node()
+            let node_config = node_config?;
-                .ok_or(Error::with_msg(format!("nodeid config error {:?}", node_config)))?;
+            retrieval::run_node(node_config.clone()).await?;
-            retrieval::run_node(node_config.clone(), node.clone()).await?;
         }
         SubCmd::Client(client) => match client.client_type {
             ClientType::Binned(opts) => {
@@ -93,9 +92,10 @@ fn simple_fetch() {
             name: format!("{}:{}", cluster.nodes[0].host, cluster.nodes[0].port),
             cluster,
         };
-        let node = node_config.get_node().unwrap();
+        let node_config: Result<NodeConfigCached, Error> = node_config.into();
+        let node_config = node_config?;
         let query_string = serde_json::to_string(&query).unwrap();
-        let host = tokio::spawn(httpret::host(node_config.clone(), node.clone()));
+        let host = tokio::spawn(httpret::host(node_config.clone()));
         let req = hyper::Request::builder()
             .method(http::Method::POST)
             .uri("http://localhost:8360/api/1/parsed_raw")

retrieval/src/lib.rs

@@ -1,5 +1,5 @@
 use err::Error;
-use netpod::{Cluster, Node, NodeConfig};
+use netpod::{Cluster, NodeConfig, NodeConfigCached};
 use tokio::task::JoinHandle;
 #[allow(unused_imports)]
 use tracing::{debug, error, info, trace, warn};
@@ -16,13 +16,15 @@ pub fn spawn_test_hosts(cluster: Cluster) -> Vec<JoinHandle<Result<(), Error>>>
             cluster: cluster.clone(),
             name: format!("{}:{}", node.host, node.port),
         };
-        let h = tokio::spawn(httpret::host(node_config, node.clone()));
+        let node_config: Result<NodeConfigCached, Error> = node_config.into();
+        let node_config = node_config.unwrap();
+        let h = tokio::spawn(httpret::host(node_config));
         ret.push(h);
     }
     ret
 }
 
-pub async fn run_node(node_config: NodeConfig, node: Node) -> Result<(), Error> {
+pub async fn run_node(node_config: NodeConfigCached) -> Result<(), Error> {
-    httpret::host(node_config, node).await?;
+    httpret::host(node_config).await?;
     Ok(())
 }