Item clone test

commonio/Cargo.toml

@@ -17,7 +17,7 @@ serde = { version = "1", features = ["derive"] }
 serde_derive = "1"
 serde_json = "1"
 chrono = "0.4"
-async-channel = "1"
+async-channel = "1.8.0"
 parking_lot = "0.11"
 crc32fast = "1.2"
 err = { path = "../err" }

daqbuffer/Cargo.toml

@@ -13,7 +13,6 @@ tracing-subscriber = "0.2.17"
 futures-core = "0.3.14"
 futures-util = "0.3.14"
 bytes = "1.0.1"
-#async-channel = "1"
 #dashmap = "3"
 serde = "1.0"
 serde_derive = "1.0"

dbconn/Cargo.toml

@@ -18,10 +18,9 @@ byteorder = "1.4"
 futures-util = "0.3.25"
 bytes = "1.3"
 pin-project = "1"
-#async-channel = "1"
 #dashmap = "3"
 scylla = "0.7"
-async-channel = "1.6"
+async-channel = "1.8.0"
 chrono = "0.4"
 regex = "1.7.0"
 err = { path = "../err" }

disk/Cargo.toml

@@ -16,7 +16,7 @@ chrono = { version = "0.4.19", features = ["serde"] }
 tokio = { version = "1.21.2", features = ["rt-multi-thread", "io-util", "net", "time", "sync", "fs"] }
 tokio-stream = {version = "0.1.5", features = ["fs"]}
 hyper = { version = "0.14", features = ["http1", "http2", "client", "server", "tcp", "stream"] }
-async-channel = "1.6"
+async-channel = "1.8.0"
 crossbeam = "0.8"
 bytes = "1.0.1"
 crc32fast = "1.3.2"

err/Cargo.toml

@@ -11,7 +11,7 @@ serde_json = "1.0"
 serde_cbor = "0.11"
 erased-serde = "0.3"
 rmp-serde = "1.1.1"
-async-channel = "1.6"
+async-channel = "1.8.0"
 chrono = { version = "0.4", features = ["serde"] }
 url = "2.2"
 regex = "1.5"

httpclient/Cargo.toml

@@ -19,7 +19,7 @@ tracing = "0.1.37"
 hyper = { version = "0.14.23", features = ["http1", "http2", "client", "server", "tcp", "stream"] }
 hyper-tls = { version = "0.5.0" }
 bytes = "1.3.0"
-async-channel = "1.7.1"
+async-channel = "1.8.0"
 err = { path = "../err" }
 netpod = { path = "../netpod" }
 parse = { path = "../parse" }

httpret/Cargo.toml

@@ -18,7 +18,7 @@ bytes = "1.3.0"
 futures-util = "0.3.14"
 tracing = "0.1"
 tracing-futures = "0.2"
-async-channel = "1.6"
+async-channel = "1.8.0"
 itertools = "0.10.1"
 chrono = "0.4.23"
 scylla = "0.7"

items_0/src/timebin.rs

@@ -57,7 +57,9 @@ pub trait TimeBinnableTy: fmt::Debug + WithLen + Send + Sized {
 
 /// Data in time-binned form.
 pub trait TimeBinned: Any + TypeName + TimeBinnable + Collectable + erased_serde::Serialize {
+    fn clone_box_time_binned(&self) -> Box<dyn TimeBinned>;
     fn as_time_binnable_dyn(&self) -> &dyn TimeBinnable;
+    fn as_time_binnable_mut(&mut self) -> &mut dyn TimeBinnable;
     fn as_collectable_mut(&mut self) -> &mut dyn Collectable;
     fn edges_slice(&self) -> (&[u64], &[u64]);
     fn counts(&self) -> &[u64];
@@ -67,6 +69,12 @@ pub trait TimeBinned: Any + TypeName + TimeBinnable + Collectable + erased_serde
     fn validate(&self) -> Result<(), String>;
 }
 
+impl Clone for Box<dyn TimeBinned> {
+    fn clone(&self) -> Self {
+        self.clone_box_time_binned()
+    }
+}
+
 impl RangeOverlapInfo for Box<dyn TimeBinned> {
     fn ends_before(&self, range: &SeriesRange) -> bool {
         todo!()
@@ -219,7 +227,7 @@ impl TimeBinnerTy for TimeBinnerDynStruct {
     type Output = Box<dyn TimeBinned>;
 
     fn ingest(&mut self, item: &mut Self::Input) {
-        trace!("{}  INGEST", Self::type_name());
+        trace!("{}  INGEST  {:?}", Self::type_name(), item);
         if self.binner.is_none() {
             self.binner = Some(Box::new(TimeBinnableTy::time_binner_new(
                 item,
@@ -312,6 +320,129 @@ impl TimeBinner for TimeBinnerDynStruct {
     }
 }
 
+#[derive(Debug)]
+pub struct TimeBinnerDynStruct2 {
+    binrange: BinnedRangeEnum,
+    do_time_weight: bool,
+    binner: Option<Box<dyn TimeBinner>>,
+}
+
+impl TimeBinnerDynStruct2 {
+    pub fn type_name() -> &'static str {
+        std::any::type_name::<Self>()
+    }
+
+    pub fn new(binrange: BinnedRangeEnum, do_time_weight: bool, binner: Box<dyn TimeBinner>) -> Self {
+        Self {
+            binrange,
+            do_time_weight,
+            binner: Some(binner),
+        }
+    }
+}
+
+impl TimeBinnerTy for TimeBinnerDynStruct2 {
+    type Input = Box<dyn TimeBinned>;
+    type Output = Box<dyn TimeBinned>;
+
+    fn ingest(&mut self, item: &mut Self::Input) {
+        trace!("{}  INGEST  {:?}", Self::type_name(), item);
+        if self.binner.is_none() {
+            self.binner = Some(Box::new(TimeBinnableTy::time_binner_new(
+                item,
+                self.binrange.clone(),
+                self.do_time_weight,
+            )));
+        }
+        self.binner
+            .as_mut()
+            .unwrap()
+            .as_mut()
+            .ingest(item.as_time_binnable_mut())
+    }
+
+    fn set_range_complete(&mut self) {
+        if let Some(k) = self.binner.as_mut() {
+            k.set_range_complete()
+        }
+    }
+
+    fn bins_ready_count(&self) -> usize {
+        if let Some(k) = self.binner.as_ref() {
+            k.bins_ready_count()
+        } else {
+            0
+        }
+    }
+
+    fn bins_ready(&mut self) -> Option<Self::Output> {
+        if let Some(k) = self.binner.as_mut() {
+            k.bins_ready()
+        } else {
+            None
+        }
+    }
+
+    fn push_in_progress(&mut self, push_empty: bool) {
+        if let Some(k) = self.binner.as_mut() {
+            k.push_in_progress(push_empty)
+        }
+    }
+
+    fn cycle(&mut self) {
+        if let Some(k) = self.binner.as_mut() {
+            k.cycle()
+        }
+    }
+
+    fn empty(&self) -> Option<Self::Output> {
+        if let Some(k) = self.binner.as_ref() {
+            Some(k.empty())
+        } else {
+            warn!("TimeBinnerDynStruct::empty called with binner None");
+            None
+        }
+    }
+
+    fn append_empty_until_end(&mut self) {
+        todo!()
+    }
+}
+
+impl TimeBinner for TimeBinnerDynStruct2 {
+    fn ingest(&mut self, item: &mut dyn TimeBinnable) {
+        todo!()
+    }
+
+    fn bins_ready_count(&self) -> usize {
+        todo!()
+    }
+
+    fn bins_ready(&mut self) -> Option<Box<dyn TimeBinned>> {
+        todo!()
+    }
+
+    fn push_in_progress(&mut self, push_empty: bool) {
+        todo!()
+    }
+
+    fn cycle(&mut self) {
+        todo!()
+    }
+
+    fn set_range_complete(&mut self) {
+        todo!()
+    }
+
+    fn empty(&self) -> Box<dyn TimeBinned> {
+        todo!()
+    }
+
+    fn append_empty_until_end(&mut self) {
+        todo!()
+    }
+}
+
 impl TimeBinnableTy for Box<dyn TimeBinnable> {
     type TimeBinner = TimeBinnerDynStruct;
 
@@ -321,6 +452,17 @@ impl TimeBinnableTy for Box<dyn TimeBinnable> {
     }
 }
 
+impl TimeBinnableTy for Box<dyn TimeBinned> {
+    type TimeBinner = TimeBinnerDynStruct2;
+
+    fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Self::TimeBinner {
+        let binner = self
+            .as_time_binnable_dyn()
+            .time_binner_new(binrange.clone(), do_time_weight);
+        TimeBinnerDynStruct2::new(binrange, do_time_weight, binner)
+    }
+}
+
 pub trait TimeBinnerIngest: fmt::Debug + TypeName + Send {
     fn ingest_inrange(&mut self, item: &mut dyn TimeBinnable) -> Result<(), Error>;
 }

items_2/src/binsdim0.rs

@@ -857,8 +857,16 @@ impl<NTY: ScalarOps> TimeBinner for BinsDim0TimeBinner<NTY> {
 }
 
 impl<NTY: ScalarOps> TimeBinned for BinsDim0<NTY> {
+    fn clone_box_time_binned(&self) -> Box<dyn TimeBinned> {
+        Box::new(self.clone())
+    }
+
     fn as_time_binnable_dyn(&self) -> &dyn TimeBinnable {
-        self as &dyn TimeBinnable
+        self
+    }
+
+    fn as_time_binnable_mut(&mut self) -> &mut dyn TimeBinnable {
+        self
     }
 
     fn edges_slice(&self) -> (&[u64], &[u64]) {

items_2/src/binsxbindim0.rs

@@ -753,8 +753,16 @@ impl<NTY: ScalarOps> TimeBinner for BinsXbinDim0TimeBinner<NTY> {
 }
 
 impl<NTY: ScalarOps> TimeBinned for BinsXbinDim0<NTY> {
+    fn clone_box_time_binned(&self) -> Box<dyn TimeBinned> {
+        Box::new(self.clone())
+    }
+
     fn as_time_binnable_dyn(&self) -> &dyn TimeBinnable {
-        self as &dyn TimeBinnable
+        self
+    }
+
+    fn as_time_binnable_mut(&mut self) -> &mut dyn TimeBinnable {
+        self
     }
 
     fn edges_slice(&self) -> (&[u64], &[u64]) {

items_2/src/channelevents.rs

@@ -830,7 +830,7 @@ impl TimeBinnerTy for ChannelEventsTimeBinner {
     type Output = Box<dyn TimeBinned>;
 
     fn ingest(&mut self, item: &mut Self::Input) {
-        trace!("{}  INGEST", Self::type_name());
+        trace!("{}  INGEST  {:?}", Self::type_name(), item);
         match item {
             ChannelEvents::Events(item) => {
                 if self.binner.is_none() {
@@ -948,6 +948,7 @@ impl TimeBinnableTy for ChannelEvents {
     type TimeBinner = ChannelEventsTimeBinner;
 
     fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Self::TimeBinner {
+        trace!("TimeBinnableTy for ChannelEvents  make ChannelEventsTimeBinner");
         // TODO probably wrong?
         let (binner, status) = match self {
             ChannelEvents::Events(_events) => (None, ConnStatus::Connect),

netpod/Cargo.toml

@@ -11,7 +11,7 @@ path = "src/netpod.rs"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 humantime-serde = "1.1.1"
-async-channel = "1.6"
+async-channel = "1.8.0"
 bytes = "1.3"
 chrono = { version = "0.4.19", features = ["serde"] }
 futures-util = "0.3.14"

nodenet/Cargo.toml

@@ -11,7 +11,7 @@ serde_cbor = "0.11.1"
 chrono = { version = "0.4.19", features = ["serde"] }
 tokio = { version = "1.21.2", features = ["io-util", "net", "time", "sync"] }
 #tokio-stream = {version = "0.1.5", features = ["fs"]}
-async-channel = "1.6"
+async-channel = "1.8.0"
 bytes = "1.0.1"
 crc32fast = "1.2.1"
 arrayref = "0.3.6"

scyllaconn/Cargo.toml

@@ -20,7 +20,7 @@ num-traits = "0.2.15"
 chrono = { version = "0.4.19", features = ["serde"] }
 crc32fast = "1.3.2"
 futures-util = "0.3.24"
-async-channel = "1.7.1"
+async-channel = "1.8.0"
 scylla = "0.7"
 tokio-postgres = { version = "0.7.7", features = ["with-chrono-0_4", "with-serde_json-1"] }
 err = { path = "../err" }

streams/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 [dependencies]
 tokio = { version = "1.21.2", features = ["io-util", "net", "time", "sync", "fs"] }
 futures-util = "0.3.15"
+pin-project = "1.0.12"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_cbor = "0.11.1"
@@ -15,6 +16,7 @@ bytes = "1.3"
 arrayref = "0.3.6"
 crc32fast = "1.3.2"
 byteorder = "1.4.3"
+async-channel = "1.8.0"
 chrono = { version = "0.4.19", features = ["serde"] }
 err = { path = "../err" }
 netpod = { path = "../netpod" }

streams/src/itemclone.rs

@@ -0,0 +1,88 @@
+use async_channel::Send;
+use async_channel::Sender;
+use err::Error;
+use futures_util::pin_mut;
+use futures_util::Future;
+use futures_util::Stream;
+use futures_util::StreamExt;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+
+#[pin_project::pin_project]
+pub struct Itemclone<T, INP>
+where
+    T: 'static,
+{
+    #[pin]
+    sender: Sender<T>,
+    inp: INP,
+    send_fut: Option<Send<'static, T>>,
+}
+
+impl<T, INP> Itemclone<T, INP> {
+    pub fn new(inp: INP, sender: Sender<T>) -> Self
+    where
+        INP: Stream<Item = T> + Unpin,
+        T: Clone + Unpin,
+    {
+        Self {
+            sender,
+            inp,
+            send_fut: None,
+        }
+    }
+}
+
+impl<T, INP> Itemclone<T, INP>
+where
+    INP: Stream<Item = T> + Unpin,
+    T: Clone + Unpin,
+{
+    fn poll_fresh(&mut self, cx: &mut Context) -> Poll<Option<Result<T, Error>>> {
+        use Poll::*;
+        match self.inp.poll_next_unpin(cx) {
+            Ready(Some(item)) => {
+                let sender = unsafe { &mut *((&mut self.sender) as *mut Sender<T>) };
+                self.send_fut = Some(sender.send(item.clone()));
+                Ready(Some(Ok(item)))
+            }
+            Ready(None) => {
+                self.sender.close();
+                Ready(None)
+            }
+            Pending => Pending,
+        }
+    }
+
+    fn send_copy(fut: &mut Send<T>, cx: &mut Context) -> Poll<Result<(), Error>> {
+        use Poll::*;
+        pin_mut!(fut);
+        match fut.poll(cx) {
+            Ready(Ok(())) => Ready(Ok(())),
+            Ready(Err(e)) => Ready(Err(e.into())),
+            Pending => Pending,
+        }
+    }
+}
+
+impl<T, INP> Stream for Itemclone<T, INP>
+where
+    INP: Stream<Item = T> + Unpin,
+    T: Clone + Unpin,
+{
+    type Item = Result<T, Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        use Poll::*;
+        let proj = self.as_mut().project();
+        match self.send_fut.as_mut() {
+            Some(fut) => match Self::send_copy(fut, cx) {
+                Ready(Ok(())) => self.poll_fresh(cx),
+                Ready(Err(e)) => Ready(Some(Err(e))),
+                Pending => Pending,
+            },
+            None => self.poll_fresh(cx),
+        }
+    }
+}

streams/src/lib.rs

@@ -4,6 +4,7 @@ pub mod dtflags;
 pub mod filechunkread;
 pub mod frames;
 pub mod generators;
+pub mod itemclone;
 pub mod needminbuffer;
 pub mod plaineventsjson;
 pub mod rangefilter2;

streams/src/test/timebin.rs

@@ -1,6 +1,9 @@
 use crate::collect::collect;
 use crate::generators::GenerateI32V00;
+use crate::generators::GenerateI32V01;
+use crate::itemclone::Itemclone;
 use crate::test::runfut;
+use crate::timebin::TimeBinnedStream;
 use crate::transform::build_event_transform;
 use err::Error;
 use futures_util::stream;
@@ -8,8 +11,11 @@ use futures_util::StreamExt;
 use items_0::on_sitemty_data;
 use items_0::streamitem::sitem_data;
 use items_0::streamitem::RangeCompletableItem;
+use items_0::streamitem::Sitemty;
 use items_0::streamitem::StreamItem;
 use items_0::timebin::TimeBinnable;
+use items_0::timebin::TimeBinned;
+use items_0::AppendAllFrom;
 use items_0::Empty;
 use items_2::binsdim0::BinsDim0;
 use items_2::channelevents::ChannelEvents;
@@ -70,7 +76,7 @@ fn time_bin_00() -> Result<(), Error> {
             d.push_back(bins);
             d
         };
-        let mut binned_stream = crate::timebin::TimeBinnedStream::new(stream0, binned_range, true);
+        let mut binned_stream = TimeBinnedStream::new(stream0, binned_range, true);
         while let Some(item) = binned_stream.next().await {
             eprintln!("{item:?}");
             match item {
@@ -135,7 +141,7 @@ fn time_bin_01() -> Result<(), Error> {
             }
         });
         let stream0 = Box::pin(stream0);
-        let mut binned_stream = crate::timebin::TimeBinnedStream::new(stream0, binned_range, true);
+        let mut binned_stream = TimeBinnedStream::new(stream0, binned_range, true);
         while let Some(item) = binned_stream.next().await {
             if true {
                 eprintln!("{item:?}");
@@ -199,7 +205,7 @@ fn time_bin_02() -> Result<(), Error> {
             x
         });
         let stream = Box::pin(stream);
-        let mut binned_stream = crate::timebin::TimeBinnedStream::new(stream, binned_range.clone(), do_time_weight);
+        let mut binned_stream = TimeBinnedStream::new(stream, binned_range.clone(), do_time_weight);
         // From there on it should no longer be neccessary to distinguish whether its still events or time bins.
         // Then, optionally collect for output type like json, or stream as batches.
         // TODO the timebinner should already provide batches to make this efficient.
@@ -280,7 +286,7 @@ fn time_bin_03() -> Result<(), Error> {
             sitem_data(v01),
             sitem_data(v03),
         ]));
-        let mut binned_stream = crate::timebin::TimeBinnedStream::new(stream0, binned_range, true);
+        let mut binned_stream = TimeBinnedStream::new(stream0, binned_range, true);
         while let Some(item) = binned_stream.next().await {
             eprintln!("{item:?}");
             match item {
@@ -312,3 +318,120 @@ fn transform_chain_correctness_00() -> Result<(), Error> {
     build_event_transform(&tq)?;
     Ok(())
 }
+
+#[test]
+fn timebin_multi_stage_00() -> Result<(), Error> {
+    // TODO chain two timebin stages with different binning grid.
+    let fut = async {
+        let do_time_weight = true;
+        let one_before_range = do_time_weight;
+        let range = nano_range_from_str("1970-01-01T00:00:10Z", "1970-01-01T00:01:03Z")?;
+        let range = SeriesRange::TimeRange(range);
+        let stream_evs = GenerateI32V01::new(0, 1, range.clone(), one_before_range);
+        let mut exp1 = {
+            let mut bins = BinsDim0::<i32>::empty();
+            for i in 0..53 {
+                bins.push(
+                    SEC * (10 + i),
+                    SEC * (11 + i),
+                    2,
+                    20 + 2 * i as i32,
+                    21 + 2 * i as i32,
+                    20.5 + 2. * i as f32,
+                );
+            }
+            bins
+        };
+        // NOTE:
+        // can store all bins in cache for which there is some non-empty bin following, or if the container has range-final.
+        let (q1tx, q1rx) = async_channel::bounded(128);
+        let (q2tx, q2rx) = async_channel::bounded(128);
+        let stream_evs = Box::pin(stream_evs);
+        let binned_stream = {
+            let binned_range = BinnedRangeEnum::covering_range(range.clone(), 48)?;
+            dbg!(&binned_range);
+            TimeBinnedStream::new(stream_evs, binned_range, do_time_weight).map(|x| {
+                //eprintln!("STAGE  1 -- {:?}", x);
+                x
+            })
+        };
+        let binned_stream = Itemclone::new(binned_stream, q1tx).map(|x| match x {
+            Ok(x) => x,
+            Err(e) => Err(e),
+        });
+        let binned_stream = {
+            let binned_range = BinnedRangeEnum::covering_range(range.clone(), 22)?;
+            dbg!(&binned_range);
+            TimeBinnedStream::new(Box::pin(binned_stream), binned_range, do_time_weight).map(|x| {
+                eprintln!("STAGE --  2 {:?}", x);
+                x
+            })
+        };
+        let binned_stream = Itemclone::new(binned_stream, q2tx).map(|x| match x {
+            Ok(x) => x,
+            Err(e) => Err(e),
+        });
+        let mut have_range_final = false;
+        let mut binned_stream = binned_stream;
+        while let Some(item) = binned_stream.next().await {
+            //eprintln!("{item:?}");
+            match item {
+                Ok(item) => match item {
+                    StreamItem::DataItem(item) => match item {
+                        RangeCompletableItem::Data(item) => {
+                            if let Some(item) = item.as_any_ref().downcast_ref::<BinsDim0<i32>>() {
+                                if false {
+                                    eprintln!("-----------------------");
+                                    eprintln!("item {:?}", item);
+                                    eprintln!("-----------------------");
+                                }
+                            } else {
+                                return Err(Error::with_msg_no_trace(format!("bad, got item with unexpected type")));
+                            }
+                        }
+                        RangeCompletableItem::RangeComplete => {
+                            have_range_final = true;
+                        }
+                    },
+                    StreamItem::Log(_) => {}
+                    StreamItem::Stats(_) => {}
+                },
+                Err(e) => Err(e).unwrap(),
+            }
+        }
+        assert!(have_range_final);
+        {
+            eprintln!("---------------------------------------------------------------------");
+            let mut coll = BinsDim0::empty();
+            let stream = q1rx;
+            while let Ok(item) = stream.recv().await {
+                //eprintln!("RECV [q1rx] {:?}", item);
+                on_sitemty_data!(item, |mut item: Box<dyn TimeBinned>| {
+                    if let Some(k) = item.as_any_mut().downcast_mut::<BinsDim0<i32>>() {
+                        coll.append_all_from(k);
+                    }
+                    sitem_data(item)
+                });
+            }
+            eprintln!("collected 1: {:?}", coll);
+            assert_eq!(coll, exp1);
+        }
+        {
+            eprintln!("---------------------------------------------------------------------");
+            let mut coll = BinsDim0::empty();
+            let stream = q2rx;
+            while let Ok(item) = stream.recv().await {
+                //eprintln!("RECV [q2rx] {:?}", item);
+                on_sitemty_data!(item, |mut item: Box<dyn TimeBinned>| {
+                    if let Some(k) = item.as_any_mut().downcast_mut::<BinsDim0<i32>>() {
+                        coll.append_all_from(k);
+                    }
+                    sitem_data(item)
+                });
+            }
+            eprintln!("collected 1: {:?}", coll);
+        }
+        Ok(())
+    };
+    runfut(fut)
+}