495 lines
19 KiB
Rust
495 lines
19 KiB
Rust
#[cfg(test)]
|
|
mod test;
|
|
|
|
use crate::log::*;
|
|
use core::ops::Range;
|
|
use futures_util::Stream;
|
|
use futures_util::StreamExt;
|
|
use items_0::container::ByteEstimate;
|
|
use items_0::merge::DrainIntoDstResult;
|
|
use items_0::merge::DrainIntoNewResult;
|
|
use items_0::merge::MergeableTy;
|
|
use items_0::on_sitemty_data;
|
|
use items_0::streamitem::sitem_data;
|
|
use items_0::streamitem::sitem_err2_from_string;
|
|
use items_0::streamitem::LogItem;
|
|
use items_0::streamitem::RangeCompletableItem;
|
|
use items_0::streamitem::SitemErrTy;
|
|
use items_0::streamitem::Sitemty;
|
|
use items_0::streamitem::StreamItem;
|
|
use items_0::Events;
|
|
use items_0::WithLen;
|
|
use netpod::TsMs;
|
|
use netpod::TsNano;
|
|
use std::collections::VecDeque;
|
|
use std::fmt;
|
|
use std::ops::ControlFlow;
|
|
use std::pin::Pin;
|
|
use std::task::Context;
|
|
use std::task::Poll;
|
|
|
|
const OUT_MAX_BYTES: u64 = 1024 * 200;
|
|
const DO_DETECT_NON_MONO: bool = true;
|
|
|
|
macro_rules! trace2 { ($($arg:tt)*) => ( if false { trace!($($arg)*); } ) }
|
|
|
|
macro_rules! trace3 { ($($arg:tt)*) => ( if false { trace!($($arg)*); } ) }
|
|
|
|
macro_rules! trace4 { ($($arg:tt)*) => ( if false { trace!($($arg)*); } ) }
|
|
|
|
macro_rules! trace_emit { ($($arg:tt)*) => ( if true { trace!($($arg)*); } ) }
|
|
|
|
#[derive(Debug, thiserror::Error)]
|
|
#[cstm(name = "MergerError")]
|
|
pub enum Error {
|
|
NoPendingButMissing,
|
|
Input(SitemErrTy),
|
|
ShouldFindTsMin,
|
|
ItemShouldHaveTsMax,
|
|
PartialPathDrainedAllItems,
|
|
}
|
|
|
|
type MergeInp<T> = Pin<Box<dyn Stream<Item = Sitemty<T>> + Send>>;
|
|
|
|
pub struct Merger<T> {
|
|
inps: Vec<Option<MergeInp<T>>>,
|
|
items: Vec<Option<T>>,
|
|
out: Option<T>,
|
|
do_clear_out: bool,
|
|
out_max_len: usize,
|
|
range_complete: Vec<bool>,
|
|
out_of_band_queue: VecDeque<Sitemty<T>>,
|
|
log_queue: VecDeque<LogItem>,
|
|
dim0ix_max: TsNano,
|
|
done_inp: bool,
|
|
done_range_complete: bool,
|
|
complete: bool,
|
|
poll_count: usize,
|
|
}
|
|
|
|
impl<T> fmt::Debug for Merger<T>
|
|
where
|
|
T: MergeableTy,
|
|
{
|
|
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
|
|
let inps: Vec<_> = self.inps.iter().map(|x| x.is_some()).collect();
|
|
fmt.debug_struct(std::any::type_name::<Self>())
|
|
.field("inps", &inps)
|
|
.field("items", &self.items)
|
|
.field("out_max_len", &self.out_max_len)
|
|
.field("range_complete", &self.range_complete)
|
|
.field("out_of_band_queue", &self.out_of_band_queue.len())
|
|
.field("done_data", &self.done_inp)
|
|
.field("done_range_complete", &self.done_range_complete)
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
impl<T> Merger<T>
|
|
where
|
|
T: MergeableTy,
|
|
{
|
|
pub fn new(inps: Vec<MergeInp<T>>, out_max_len: Option<u32>) -> Self {
|
|
let n = inps.len();
|
|
Self {
|
|
inps: inps.into_iter().map(|x| Some(x)).collect(),
|
|
items: (0..n).into_iter().map(|_| None).collect(),
|
|
out: None,
|
|
do_clear_out: false,
|
|
out_max_len: out_max_len.unwrap_or(1000) as usize,
|
|
range_complete: vec![false; n],
|
|
out_of_band_queue: VecDeque::new(),
|
|
log_queue: VecDeque::new(),
|
|
dim0ix_max: TsNano::from_ns(0),
|
|
done_inp: false,
|
|
done_range_complete: false,
|
|
complete: false,
|
|
poll_count: 0,
|
|
}
|
|
}
|
|
|
|
fn drain_into_dst_upto(src: &mut T, dst: &mut T, upto: TsNano) -> DrainIntoDstResult {
|
|
match src.find_lowest_index_gt(upto) {
|
|
Some(ilgt) => src.drain_into(dst, 0..ilgt),
|
|
None => {
|
|
// TODO should not be here.
|
|
src.drain_into(dst, 0..src.len())
|
|
}
|
|
}
|
|
}
|
|
|
|
fn drain_into_new_upto(src: &mut T, upto: TsNano) -> DrainIntoNewResult<T> {
|
|
match src.find_lowest_index_gt(upto) {
|
|
Some(ilgt) => src.drain_into_new(0..ilgt),
|
|
None => {
|
|
// TODO should not be here.
|
|
src.drain_into_new(0..src.len())
|
|
}
|
|
}
|
|
}
|
|
|
|
fn take_into_output_upto(&mut self, src: &mut T, upto: TsNano) -> DrainIntoDstResult {
|
|
// TODO optimize the case when some large batch should be added to some existing small batch already in out.
|
|
// TODO maybe use two output slots?
|
|
if let Some(out) = self.out.as_mut() {
|
|
Self::drain_into_dst_upto(src, out, upto)
|
|
} else {
|
|
trace2!("move into fresh");
|
|
match Self::drain_into_new_upto(src, upto) {
|
|
DrainIntoNewResult::Done(x) => {
|
|
self.out = Some(x);
|
|
DrainIntoDstResult::Done
|
|
}
|
|
DrainIntoNewResult::Partial(x) => {
|
|
self.out = Some(x);
|
|
DrainIntoDstResult::Partial
|
|
}
|
|
DrainIntoNewResult::NotCompatible => DrainIntoDstResult::NotCompatible,
|
|
}
|
|
}
|
|
}
|
|
|
|
fn take_into_output_all(&mut self, src: &mut T) -> DrainIntoDstResult {
|
|
// TODO optimize the case when some large batch should be added to some existing small batch already in out.
|
|
// TODO maybe use two output slots?
|
|
self.take_into_output_upto(src, TsNano::from_ns(u64::MAX))
|
|
}
|
|
|
|
fn process(mut self: Pin<&mut Self>, _cx: &mut Context) -> Result<ControlFlow<()>, Error> {
|
|
use ControlFlow::*;
|
|
trace4!("process");
|
|
let mut log_items = Vec::new();
|
|
let mut tslows = [None, None];
|
|
for (i1, itemopt) in self.items.iter_mut().enumerate() {
|
|
if let Some(item) = itemopt {
|
|
if let Some(t1) = item.ts_min() {
|
|
if let Some((_, a)) = tslows[0] {
|
|
if t1 < a {
|
|
tslows[1] = tslows[0];
|
|
tslows[0] = Some((i1, t1));
|
|
} else {
|
|
if let Some((_, b)) = tslows[1] {
|
|
if t1 < b {
|
|
tslows[1] = Some((i1, t1));
|
|
} else {
|
|
// nothing to do
|
|
}
|
|
} else {
|
|
tslows[1] = Some((i1, t1));
|
|
}
|
|
}
|
|
} else {
|
|
tslows[0] = Some((i1, t1));
|
|
}
|
|
} else {
|
|
// the item seems empty.
|
|
// TODO count for stats.
|
|
trace2!("empty item, something to do here?");
|
|
*itemopt = None;
|
|
return Ok(Continue(()));
|
|
}
|
|
}
|
|
}
|
|
if DO_DETECT_NON_MONO {
|
|
if let Some((i1, t1)) = tslows[0].as_ref() {
|
|
if *t1 <= self.dim0ix_max {
|
|
self.dim0ix_max = *t1;
|
|
let item = LogItem {
|
|
node_ix: *i1 as _,
|
|
level: Level::INFO,
|
|
msg: format!(
|
|
"dim0ix_max {} vs {} diff {}",
|
|
self.dim0ix_max,
|
|
t1,
|
|
self.dim0ix_max.ns() - t1.ns()
|
|
),
|
|
};
|
|
log_items.push(item);
|
|
}
|
|
}
|
|
}
|
|
trace4!("tslows {:?}", tslows);
|
|
if let Some((il0, _tl0)) = tslows[0] {
|
|
if let Some((_il1, tl1)) = tslows[1] {
|
|
// There is a second input, take only up to the second highest timestamp
|
|
let item = self.items[il0].as_mut().unwrap();
|
|
if let Some(th0) = item.ts_max() {
|
|
if th0 <= tl1 {
|
|
// Can take the whole item
|
|
// TODO gather stats about this case. Should be never for databuffer, and often for scylla.
|
|
let mut item = self.items[il0].take().unwrap();
|
|
trace3!("Take all from item {:?}", item);
|
|
match self.take_into_output_all(&mut item) {
|
|
DrainIntoDstResult::Done => Ok(Break(())),
|
|
DrainIntoDstResult::Partial => {
|
|
// TODO count for stats
|
|
trace3!("Put item back");
|
|
self.items[il0] = Some(item);
|
|
self.do_clear_out = true;
|
|
Ok(Break(()))
|
|
}
|
|
DrainIntoDstResult::NotCompatible => {
|
|
// TODO count for stats
|
|
trace3!("Put item back");
|
|
self.items[il0] = Some(item);
|
|
self.do_clear_out = true;
|
|
Ok(Break(()))
|
|
}
|
|
}
|
|
} else {
|
|
// Take only up to the lowest ts of the second-lowest input
|
|
let mut item = self.items[il0].take().unwrap();
|
|
trace3!("Take up to {} from item {:?}", tl1, item);
|
|
match self.take_into_output_upto(&mut item, tl1) {
|
|
DrainIntoDstResult::Done => {
|
|
if item.len() == 0 {
|
|
// TODO should never be here because we should have taken the whole item
|
|
Err(Error::PartialPathDrainedAllItems)
|
|
} else {
|
|
self.items[il0] = Some(item);
|
|
Ok(Break(()))
|
|
}
|
|
}
|
|
DrainIntoDstResult::Partial => {
|
|
// TODO count for stats
|
|
trace3!("Put item back because Partial");
|
|
self.items[il0] = Some(item);
|
|
self.do_clear_out = true;
|
|
Ok(Break(()))
|
|
}
|
|
DrainIntoDstResult::NotCompatible => {
|
|
// TODO count for stats
|
|
trace3!("Put item back because NotCompatible");
|
|
self.items[il0] = Some(item);
|
|
self.do_clear_out = true;
|
|
Ok(Break(()))
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
Err(Error::ItemShouldHaveTsMax)
|
|
}
|
|
} else {
|
|
// No other input, take the whole item
|
|
let mut item = self.items[il0].take().unwrap();
|
|
trace3!("Take all from item (no other input) {:?}", item);
|
|
match self.take_into_output_all(&mut item) {
|
|
DrainIntoDstResult::Done => Ok(Break(())),
|
|
DrainIntoDstResult::Partial => {
|
|
// TODO count for stats
|
|
trace3!("Put item back");
|
|
self.items[il0] = Some(item);
|
|
self.do_clear_out = true;
|
|
Ok(Break(()))
|
|
}
|
|
DrainIntoDstResult::NotCompatible => {
|
|
// TODO count for stats
|
|
trace3!("Put item back");
|
|
self.items[il0] = Some(item);
|
|
self.do_clear_out = true;
|
|
Ok(Break(()))
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
Err(Error::ShouldFindTsMin)
|
|
}
|
|
}
|
|
|
|
fn refill(mut self: Pin<&mut Self>, cx: &mut Context) -> Result<Poll<()>, Error> {
|
|
trace4!("refill");
|
|
use Poll::*;
|
|
let mut has_pending = false;
|
|
for i in 0..self.inps.len() {
|
|
if self.items[i].is_none() {
|
|
while let Some(inp) = self.inps[i].as_mut() {
|
|
match inp.poll_next_unpin(cx) {
|
|
Ready(Some(Ok(k))) => match k {
|
|
StreamItem::DataItem(k) => match k {
|
|
RangeCompletableItem::Data(k) => {
|
|
self.items[i] = Some(k);
|
|
trace4!("refilled {}", i);
|
|
}
|
|
RangeCompletableItem::RangeComplete => {
|
|
self.range_complete[i] = true;
|
|
trace!("range_complete {:?}", self.range_complete);
|
|
continue;
|
|
}
|
|
},
|
|
StreamItem::Log(item) => {
|
|
// TODO limit queue length
|
|
self.out_of_band_queue.push_back(Ok(StreamItem::Log(item)));
|
|
continue;
|
|
}
|
|
StreamItem::Stats(item) => {
|
|
// TODO limit queue length
|
|
self.out_of_band_queue
|
|
.push_back(Ok(StreamItem::Stats(item)));
|
|
continue;
|
|
}
|
|
},
|
|
Ready(Some(Err(e))) => {
|
|
self.inps[i] = None;
|
|
return Err(Error::Input(e));
|
|
}
|
|
Ready(None) => {
|
|
self.inps[i] = None;
|
|
}
|
|
Pending => {
|
|
has_pending = true;
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
if has_pending {
|
|
Ok(Pending)
|
|
} else {
|
|
Ok(Ready(()))
|
|
}
|
|
}
|
|
|
|
fn poll3(mut self: Pin<&mut Self>, cx: &mut Context) -> ControlFlow<Poll<Option<Error>>> {
|
|
use ControlFlow::*;
|
|
use Poll::*;
|
|
trace4!("poll3");
|
|
#[allow(unused)]
|
|
let ninps = self.inps.iter().filter(|a| a.is_some()).count();
|
|
let nitems = self.items.iter().filter(|a| a.is_some()).count();
|
|
let nitemsmissing = self
|
|
.inps
|
|
.iter()
|
|
.zip(self.items.iter())
|
|
.filter(|(a, b)| a.is_some() && b.is_none())
|
|
.count();
|
|
trace3!(
|
|
"ninps {} nitems {} nitemsmissing {}",
|
|
ninps,
|
|
nitems,
|
|
nitemsmissing
|
|
);
|
|
if nitemsmissing != 0 {
|
|
let e = Error::NoPendingButMissing;
|
|
return Break(Ready(Some(e)));
|
|
}
|
|
let last_emit = nitems == 0;
|
|
if nitems != 0 {
|
|
match Self::process(Pin::new(&mut self), cx) {
|
|
Ok(Break(())) => {}
|
|
Ok(Continue(())) => {}
|
|
Err(e) => return Break(Ready(Some(e))),
|
|
}
|
|
}
|
|
if let Some(o) = self.out.as_ref() {
|
|
if o.len() >= self.out_max_len
|
|
|| o.byte_estimate() >= OUT_MAX_BYTES
|
|
|| self.do_clear_out
|
|
|| last_emit
|
|
{
|
|
if o.len() > 2 * self.out_max_len {
|
|
debug!(
|
|
"MERGER OVERLENGTH ITEM {} vs {}",
|
|
o.len(),
|
|
self.out_max_len
|
|
);
|
|
}
|
|
if o.byte_estimate() > 2 * OUT_MAX_BYTES {
|
|
debug!(
|
|
"MERGER OVERWEIGHT ITEM {} vs {}",
|
|
o.byte_estimate(),
|
|
OUT_MAX_BYTES
|
|
);
|
|
}
|
|
trace3!("decide to output");
|
|
self.do_clear_out = false;
|
|
let item = sitem_data(self.out.take().unwrap());
|
|
self.out_of_band_queue.push_back(item);
|
|
Continue(())
|
|
} else {
|
|
trace4!("not enough output yet");
|
|
Continue(())
|
|
}
|
|
} else {
|
|
trace!("no output candidate");
|
|
if last_emit {
|
|
Break(Ready(None))
|
|
} else {
|
|
Continue(())
|
|
}
|
|
}
|
|
}
|
|
|
|
fn poll2(mut self: Pin<&mut Self>, cx: &mut Context) -> ControlFlow<Poll<Option<Error>>> {
|
|
use ControlFlow::*;
|
|
use Poll::*;
|
|
match Self::refill(Pin::new(&mut self), cx) {
|
|
Ok(Ready(())) => Self::poll3(self, cx),
|
|
Ok(Pending) => Break(Pending),
|
|
Err(e) => Break(Ready(Some(e))),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T> Stream for Merger<T>
|
|
where
|
|
T: MergeableTy,
|
|
{
|
|
type Item = Sitemty<T>;
|
|
|
|
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
|
|
use Poll::*;
|
|
self.poll_count += 1;
|
|
let span1 = span!(Level::INFO, "Merger", pc = self.poll_count);
|
|
let _spg = span1.enter();
|
|
loop {
|
|
trace3!("poll");
|
|
break if self.poll_count == usize::MAX {
|
|
self.done_range_complete = true;
|
|
continue;
|
|
} else if self.complete {
|
|
panic!("poll after complete");
|
|
} else if self.done_range_complete {
|
|
self.complete = true;
|
|
Ready(None)
|
|
} else if let Some(item) = self.log_queue.pop_front() {
|
|
Ready(Some(Ok(StreamItem::Log(item))))
|
|
} else if let Some(item) = self.out_of_band_queue.pop_front() {
|
|
trace_emit!("emit item");
|
|
let item = on_sitemty_data!(item, |k: T| {
|
|
trace_emit!("emit item len {}", k.len());
|
|
sitem_data(k)
|
|
});
|
|
Ready(Some(item))
|
|
} else if self.done_inp == false {
|
|
match Self::poll2(self.as_mut(), cx) {
|
|
ControlFlow::Continue(()) => continue,
|
|
ControlFlow::Break(k) => match k {
|
|
Ready(Some(e)) => {
|
|
self.done_inp = true;
|
|
Ready(Some(Err(sitem_err2_from_string(e))))
|
|
}
|
|
Ready(None) => {
|
|
self.done_inp = true;
|
|
if let Some(out) = self.out.take() {
|
|
trace!("done_data emit buffered len {}", out.len());
|
|
self.out_of_band_queue.push_back(sitem_data(out));
|
|
}
|
|
continue;
|
|
}
|
|
Pending => Pending,
|
|
},
|
|
}
|
|
} else {
|
|
self.done_range_complete = true;
|
|
if self.range_complete.iter().all(|x| *x) {
|
|
trace!("emit RangeComplete");
|
|
let item = Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete));
|
|
self.out_of_band_queue.push_back(item);
|
|
}
|
|
continue;
|
|
};
|
|
}
|
|
}
|
|
}
|