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

WIP refactor data event pipeline

Browse Source
This commit is contained in:
Dominik Werder
2022-11-15 16:16:16 +01:00
parent fb78f1887e
commit eebf8665ce
24 changed files with 800 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
streams/src/collect.rs Normal file
View File

@@ -0,0 +1,114 @@
use err::Error;
use futures_util::{Stream, StreamExt};
use items::{RangeCompletableItem, Sitemty, StreamItem};
use items_2::streams::{Collectable, Collector};
use netpod::log::*;
use serde::Serialize;
use serde_json::Value as JsonValue;
use std::fmt;
use std::time::Duration;
// This is meant to work with trait object event containers (crate items_2)
// TODO rename, it is also used for binned:
pub async fn collect_plain_events_json<T, S>(
stream: S,
timeout: Duration,
events_max: u64,
do_log: bool,
) -> Result<JsonValue, Error>
where
S: Stream<Item = Sitemty<T>> + Unpin,
T: Collectable + fmt::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: Option<Box<dyn Collector>> = None;
let mut i1 = 0;
let mut stream = stream;
let mut total_duration = Duration::ZERO;
loop {
let item = if i1 == 0 {
stream.next().await
} else {
if false {
None
} else {
match tokio::time::timeout_at(deadline, stream.next()).await {
Ok(k) => k,
Err(_) => {
eprintln!("TODO [smc3j3rwha732ru8wcnfgi]");
err::todo();
//collector.set_timed_out();
None
}
}
}
};
match item {
Some(item) => {
match item {
Ok(item) => match item {
StreamItem::Log(item) => {
if do_log {
debug!("collect_plain_events_json log {:?}", item);
}
}
StreamItem::Stats(item) => {
use items::StatsItem;
use netpod::DiskStats;
match item {
// TODO factor and simplify the stats collection:
StatsItem::EventDataReadStats(_) => {}
StatsItem::RangeFilterStats(_) => {}
StatsItem::DiskStats(item) => match item {
DiskStats::OpenStats(k) => {
total_duration += k.duration;
}
DiskStats::SeekStats(k) => {
total_duration += k.duration;
}
DiskStats::ReadStats(k) => {
total_duration += k.duration;
}
DiskStats::ReadExactStats(k) => {
total_duration += k.duration;
}
},
}
}
StreamItem::DataItem(item) => match item {
RangeCompletableItem::RangeComplete => {
eprintln!("TODO [73jdfcgf947d]");
err::todo();
//collector.set_range_complete();
}
RangeCompletableItem::Data(item) => {
eprintln!("TODO [nx298nu98venusfc8]");
err::todo();
//collector.ingest(&item);
i1 += 1;
if i1 >= events_max {
break;
}
}
},
},
Err(e) => {
// TODO Need to use some flags to get good enough error message for remote user.
Err(e)?;
}
};
}
None => break,
}
}
let res = collector
.ok_or_else(|| Error::with_msg_no_trace(format!("no collector created")))?
.result()?;
let ret = serde_json::to_value(&res)?;
debug!("Total duration: {:?}", total_duration);
Ok(ret)
}

5
streams/src/eventchunker.rs
View File

@@ -5,10 +5,7 @@ use bytes::{Buf, BytesMut};
use err::Error;
use futures_util::{Stream, StreamExt};
use items::eventfull::EventFull;
use items::{
RangeCompletableItem, StatsItem,
StreamItem, WithLen,
};
use items::{RangeCompletableItem, StatsItem, StreamItem, WithLen};
use netpod::histo::HistoLog2;
use netpod::log::*;
use netpod::timeunits::SEC;

3
streams/src/frames/eventsfromframes.rs
View File

@@ -1,6 +1,5 @@
use super::inmem::InMemoryFrameAsyncReadStream;
use futures_core::Stream;
use futures_util::StreamExt;
use futures_util::{Stream, StreamExt};
use items::frame::decode_frame;
use items::{FrameTypeInnerStatic, Sitemty, StreamItem};
use netpod::log::*;

3
streams/src/frames/inmem.rs
View File

@@ -1,7 +1,6 @@
use bytes::{BufMut, BytesMut};
use err::Error;
use futures_core::Stream;
use futures_util::pin_mut;
use futures_util::{pin_mut, Stream};
use items::inmem::InMemoryFrame;
use items::StreamItem;
use items::{INMEM_FRAME_FOOT, INMEM_FRAME_HEAD, INMEM_FRAME_MAGIC};

3
streams/src/lib.rs
View File

@@ -1,7 +1,10 @@
pub mod collect;
pub mod dtflags;
pub mod eventchunker;
pub mod filechunkread;
pub mod frames;
pub mod merge;
pub mod needminbuffer;
pub mod plaineventsjson;
pub mod rangefilter;
pub mod tcprawclient;

50
streams/src/merge.rs Normal file
View File

@@ -0,0 +1,50 @@
// Sets up the raw tcp connections: disk::merge::mergedfromremotes::MergedFromRemotes
// and then sets up a disk::merge::MergedStream
pub mod mergedstream;
use crate::frames::eventsfromframes::EventsFromFrames;
use crate::frames::inmem::InMemoryFrameAsyncReadStream;
use err::Error;
use futures_util::Stream;
use items::frame::make_frame;
use items::frame::make_term_frame;
use items::sitem_data;
use items::EventQueryJsonStringFrame;
use items::Sitemty;
use items_2::ChannelEvents;
use netpod::log::*;
use netpod::Cluster;
use std::pin::Pin;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
pub type ChannelEventsBoxedStream = Pin<Box<dyn Stream<Item = Sitemty<ChannelEvents>> + Send>>;
pub async fn open_tcp_streams(
query: &dyn erased_serde::Serialize,
cluster: &Cluster,
) -> Result<Vec<ChannelEventsBoxedStream>, Error> {
// TODO when unit tests established, change to async connect:
let mut streams = Vec::new();
for node in &cluster.nodes {
debug!("x_processed_stream_from_node to: {}:{}", node.host, node.port_raw);
let net = TcpStream::connect(format!("{}:{}", node.host, node.port_raw)).await?;
let qjs = serde_json::to_string(&query)?;
let (netin, mut netout) = net.into_split();
let item = EventQueryJsonStringFrame(qjs);
let item = sitem_data(item);
let buf = make_frame(&item)?;
netout.write_all(&buf).await?;
let buf = make_term_frame()?;
netout.write_all(&buf).await?;
netout.flush().await?;
netout.forget();
// TODO for images, we need larger buffer capacity
let frames = InMemoryFrameAsyncReadStream::new(netin, 1024 * 128);
type ITEM = ChannelEvents;
let stream = EventsFromFrames::<_, ITEM>::new(frames);
streams.push(Box::pin(stream) as _);
}
Ok(streams)
}

308
streams/src/merge/mergedstream.rs Normal file
View File

@@ -0,0 +1,308 @@
use err::Error;
use futures_util::{Stream, StreamExt};
use items::ByteEstimate;
use items::{Appendable, LogItem, PushableIndex, RangeCompletableItem, Sitemty, StatsItem, StreamItem, WithTimestamps};
use netpod::histo::HistoLog2;
use netpod::log::*;
use netpod::ByteSize;
use std::collections::VecDeque;
use std::pin::Pin;
use std::task::{Context, Poll};
// TODO compare with items_2::merger::*
const LOG_EMIT_ITEM: bool = false;
enum MergedCurVal<T> {
None,
Finish,
Val(T),
}
pub struct MergedStream<S, ITY> {
inps: Vec<S>,
current: Vec<MergedCurVal<ITY>>,
ixs: Vec<usize>,
errored: bool,
completed: bool,
batch: Option<ITY>,
ts_last_emit: u64,
range_complete_observed: Vec<bool>,
range_complete_observed_all: bool,
range_complete_observed_all_emitted: bool,
data_emit_complete: bool,
batch_size: ByteSize,
batch_len_emit_histo: HistoLog2,
logitems: VecDeque<LogItem>,
stats_items: VecDeque<StatsItem>,
}
impl<S, ITY> Drop for MergedStream<S, ITY> {
fn drop(&mut self) {
// TODO collect somewhere
debug!(
"MergedStream Drop Stats:\nbatch_len_emit_histo: {:?}",
self.batch_len_emit_histo
);
}
}
impl<S, ITY> MergedStream<S, ITY>
where
S: Stream<Item = Sitemty<ITY>> + Unpin,
ITY: Appendable + Unpin,
{
pub fn new(inps: Vec<S>) -> Self {
trace!("MergedStream::new");
let n = inps.len();
let current = (0..n).into_iter().map(|_| MergedCurVal::None).collect();
Self {
inps,
current: current,
ixs: vec![0; n],
errored: false,
completed: false,
batch: None,
ts_last_emit: 0,
range_complete_observed: vec![false; n],
range_complete_observed_all: false,
range_complete_observed_all_emitted: false,
data_emit_complete: false,
batch_size: ByteSize::kb(128),
batch_len_emit_histo: HistoLog2::new(0),
logitems: VecDeque::new(),
stats_items: VecDeque::new(),
}
}
fn replenish(self: &mut Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Error>> {
use Poll::*;
let mut pending = 0;
for i1 in 0..self.inps.len() {
match self.current[i1] {
MergedCurVal::None => {
'l1: loop {
break match self.inps[i1].poll_next_unpin(cx) {
Ready(Some(Ok(k))) => match k {
StreamItem::Log(item) => {
self.logitems.push_back(item);
continue 'l1;
}
StreamItem::Stats(item) => {
self.stats_items.push_back(item);
continue 'l1;
}
StreamItem::DataItem(item) => match item {
RangeCompletableItem::RangeComplete => {
self.range_complete_observed[i1] = true;
let d = self.range_complete_observed.iter().filter(|&&k| k).count();
if d == self.range_complete_observed.len() {
self.range_complete_observed_all = true;
debug!("MergedStream range_complete d {} COMPLETE", d);
} else {
trace!("MergedStream range_complete d {}", d);
}
continue 'l1;
}
RangeCompletableItem::Data(item) => {
self.ixs[i1] = 0;
self.current[i1] = MergedCurVal::Val(item);
}
},
},
Ready(Some(Err(e))) => {
// TODO emit this error, consider this stream as done, anything more to do here?
//self.current[i1] = CurVal::Err(e);
self.errored = true;
return Ready(Err(e));
}
Ready(None) => {
self.current[i1] = MergedCurVal::Finish;
}
Pending => {
pending += 1;
}
};
}
}
_ => (),
}
}
if pending > 0 {
Pending
} else {
Ready(Ok(()))
}
}
}
impl<S, ITY> Stream for MergedStream<S, ITY>
where
S: Stream<Item = Sitemty<ITY>> + Unpin,
ITY: PushableIndex + Appendable + ByteEstimate + WithTimestamps + Unpin,
{
type Item = Sitemty<ITY>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
'outer: loop {
break if self.completed {
panic!("poll_next on completed");
} else if self.errored {
self.completed = true;
Ready(None)
} else if let Some(item) = self.logitems.pop_front() {
Ready(Some(Ok(StreamItem::Log(item))))
} else if let Some(item) = self.stats_items.pop_front() {
Ready(Some(Ok(StreamItem::Stats(item))))
} else if self.range_complete_observed_all_emitted {
self.completed = true;
Ready(None)
} else if self.data_emit_complete {
if self.range_complete_observed_all {
if self.range_complete_observed_all_emitted {
self.completed = true;
Ready(None)
} else {
self.range_complete_observed_all_emitted = true;
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
}
} else {
self.completed = true;
Ready(None)
}
} else {
match self.replenish(cx) {
Ready(Ok(_)) => {
let mut lowest_ix = usize::MAX;
let mut lowest_ts = u64::MAX;
for i1 in 0..self.inps.len() {
if let MergedCurVal::Val(val) = &self.current[i1] {
let u = self.ixs[i1];
if u >= val.len() {
self.ixs[i1] = 0;
self.current[i1] = MergedCurVal::None;
continue 'outer;
} else {
let ts = val.ts(u);
if ts < lowest_ts {
lowest_ix = i1;
lowest_ts = ts;
}
}
}
}
if lowest_ix == usize::MAX {
if let Some(batch) = self.batch.take() {
if batch.len() != 0 {
self.batch_len_emit_histo.ingest(batch.len() as u32);
self.data_emit_complete = true;
if LOG_EMIT_ITEM {
let mut aa = vec![];
for ii in 0..batch.len() {
aa.push(batch.ts(ii));
}
debug!("MergedBlobsStream A emits {} events tss {:?}", batch.len(), aa);
};
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(batch)))))
} else {
self.data_emit_complete = true;
continue 'outer;
}
} else {
self.data_emit_complete = true;
continue 'outer;
}
} else {
// TODO unordered cases
if lowest_ts < self.ts_last_emit {
self.errored = true;
let msg = format!(
"unordered event at lowest_ts {} ts_last_emit {}",
lowest_ts, self.ts_last_emit
);
return Ready(Some(Err(Error::with_public_msg(msg))));
} else {
self.ts_last_emit = self.ts_last_emit.max(lowest_ts);
}
{
let batch = self.batch.take();
let rix = self.ixs[lowest_ix];
match &self.current[lowest_ix] {
MergedCurVal::Val(val) => {
let mut ldst = batch.unwrap_or_else(|| val.empty_like_self());
if false {
info!(
"Push event rix {} lowest_ix {} lowest_ts {}",
rix, lowest_ix, lowest_ts
);
}
ldst.push_index(val, rix);
self.batch = Some(ldst);
}
MergedCurVal::None => panic!(),
MergedCurVal::Finish => panic!(),
}
}
self.ixs[lowest_ix] += 1;
let curlen = match &self.current[lowest_ix] {
MergedCurVal::Val(val) => val.len(),
MergedCurVal::None => panic!(),
MergedCurVal::Finish => panic!(),
};
if self.ixs[lowest_ix] >= curlen {
self.ixs[lowest_ix] = 0;
self.current[lowest_ix] = MergedCurVal::None;
}
let emit_packet_now = if let Some(batch) = &self.batch {
if batch.byte_estimate() >= self.batch_size.bytes() as u64 {
true
} else {
false
}
} else {
false
};
if emit_packet_now {
if let Some(batch) = self.batch.take() {
trace!("emit item because over threshold len {}", batch.len());
self.batch_len_emit_histo.ingest(batch.len() as u32);
if LOG_EMIT_ITEM {
let mut aa = vec![];
for ii in 0..batch.len() {
aa.push(batch.ts(ii));
}
debug!("MergedBlobsStream B emits {} events tss {:?}", batch.len(), aa);
};
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(batch)))))
} else {
continue 'outer;
}
} else {
continue 'outer;
}
}
}
Ready(Err(e)) => {
self.errored = true;
Ready(Some(Err(e)))
}
Pending => Pending,
}
};
}
}
}
#[cfg(test)]
mod test {
use items_2::{ChannelEvents, Empty};
#[test]
fn merge_channel_events() {
let mut evs = items_2::eventsdim0::EventsDim0::empty();
evs.push(1, 100, 17u8);
evs.push(3, 300, 16);
let _cevs = ChannelEvents::Events(Box::new(evs));
}
}

73
streams/src/plaineventsjson.rs Normal file
View File

@@ -0,0 +1,73 @@
use crate::merge::open_tcp_streams;
use bytes::Bytes;
use err::Error;
use futures_util::{future, stream, FutureExt, Stream, StreamExt};
use items::streams::collect_plain_events_json;
use items::{sitem_data, RangeCompletableItem, Sitemty, StreamItem};
use items_2::ChannelEvents;
use netpod::log::*;
use netpod::Cluster;
use serde::Serialize;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
pub struct BytesStream(Pin<Box<dyn Stream<Item = Sitemty<Bytes>> + Send>>);
impl Stream for BytesStream {
type Item = Sitemty<Bytes>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
StreamExt::poll_next_unpin(&mut self, cx)
}
}
pub async fn plain_events_json<SER>(query: SER, cluster: &Cluster) -> Result<BytesStream, Error>
where
SER: Serialize,
{
let inps = open_tcp_streams(&query, cluster).await?;
let mut merged = items_2::merger::ChannelEventsMerger::new(inps);
let timeout = Duration::from_millis(2000);
let events_max = 100;
let do_log = false;
let mut coll = None;
while let Some(item) = merged.next().await {
let item = item?;
match item {
StreamItem::DataItem(item) => match item {
RangeCompletableItem::RangeComplete => todo!(),
RangeCompletableItem::Data(item) => match item {
ChannelEvents::Events(mut item) => {
if coll.is_none() {
coll = Some(item.new_collector());
}
let coll = coll
.as_mut()
.ok_or_else(|| Error::with_msg_no_trace(format!("no collector")))?;
coll.ingest(&mut item);
}
ChannelEvents::Status(_) => todo!(),
},
},
StreamItem::Log(item) => {
info!("log {item:?}");
}
StreamItem::Stats(item) => {
info!("stats {item:?}");
}
}
}
// TODO compare with
// streams::collect::collect_plain_events_json
// and remove duplicate functionality.
let mut coll = coll.ok_or_else(|| Error::with_msg_no_trace(format!("no collector created")))?;
let res = coll.result()?;
// TODO factor the conversion of the result out to a higher level.
// The output of this function should again be collectable, maybe even binnable and otherwise processable.
let js = serde_json::to_vec(&res)?;
let item = sitem_data(Bytes::from(js));
let stream = stream::once(future::ready(item));
let stream = BytesStream(Box::pin(stream));
Ok(stream)
}

3
streams/src/rangefilter.rs
View File

@@ -1,6 +1,5 @@
use err::Error;
use futures_core::Stream;
use futures_util::StreamExt;
use futures_util::{Stream, StreamExt};
use items::StatsItem;
use items::{Appendable, Clearable, PushableIndex, RangeCompletableItem, Sitemty, StreamItem, WithTimestamps};
use netpod::{log::*, RangeFilterStats};

2
streams/src/tcprawclient.rs
View File

@@ -8,7 +8,7 @@ to request such data from nodes.
use crate::frames::eventsfromframes::EventsFromFrames;
use crate::frames::inmem::InMemoryFrameAsyncReadStream;
use err::Error;
use futures_core::Stream;
use futures_util::Stream;
use items::eventfull::EventFull;
use items::frame::{make_frame, make_term_frame};
use items::{EventQueryJsonStringFrame, EventsNodeProcessor, RangeCompletableItem, Sitemty, StreamItem};