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

Tests pass

Browse Source
This commit is contained in:
Dominik Werder
2021-10-14 22:34:06 +02:00
parent 3b917d6bd3
commit 7b79070e8a
16 changed files with 756 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
disk/src/aggtest.rs
View File

@@ -17,6 +17,7 @@ pub fn make_test_node(id: u32) -> Node {
backend: "testbackend".into(),
splits: None,
archiver_appliance: None,
channel_archiver: None,
}
}

7
disk/src/dataopen.rs
View File

@@ -30,9 +30,12 @@ async fn position_file(
expand_left: bool,
expand_right: bool,
) -> Result<Positioned, Error> {
info!(
trace!(
"position_file called {} {} {:?} {:?}",
expand_left, expand_right, range, path
expand_left,
expand_right,
range,
path
);
assert_eq!(expand_left && expand_right, false);
match OpenOptions::new().read(true).open(&path).await {

4
disk/src/eventchunker.rs
View File

@@ -89,6 +89,7 @@ impl EventChunker {
expand: bool,
do_decompress: bool,
) -> Self {
trace!("EventChunker::from_start");
let mut inp = NeedMinBuffer::new(inp);
inp.set_need_min(6);
Self {
@@ -609,9 +610,12 @@ impl Stream for EventChunker {
Ready(None)
} else if self.final_stats_sent {
self.sent_beyond_range = true;
trace!("sent_beyond_range");
if self.seen_beyond_range {
trace!("sent_beyond_range RangeComplete");
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
} else {
trace!("sent_beyond_range non-complete");
continue 'outer;
}
} else if self.data_emit_complete {

5
disk/src/gen.rs
View File

@@ -12,7 +12,9 @@ use tracing::{debug, error, info, trace, warn};
#[test]
pub fn gen_test_data_test() {
taskrun::run(gen_test_data()).unwrap();
if false {
taskrun::run(gen_test_data()).unwrap();
}
}
pub async fn gen_test_data() -> Result<(), Error> {
@@ -126,6 +128,7 @@ pub async fn gen_test_data() -> Result<(), Error> {
backend: "testbackend".into(),
splits: None,
archiver_appliance: None,
channel_archiver: None,
};
ensemble.nodes.push(node);
}

89
disk/src/lib.rs
View File

@@ -65,6 +65,8 @@ impl Stream for FileReader {
type Item = Result<Bytes, Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
err::todo();
// TODO remove if no longer used?
let blen = self.buffer_size as usize;
let mut buf2 = BytesMut::with_capacity(blen);
buf2.resize(buf2.capacity(), 0);
@@ -312,58 +314,55 @@ impl Stream for NeedMinBuffer {
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 z = match self.inp.poll_next_unpin(cx) {
Ready(Some(Ok(fcr))) => {
self.buf_len_histo.ingest(fcr.buf.len() as u32);
//info!("NeedMinBuffer got buf len {}", fcr.buf.len());
match self.left.take() {
Some(mut lfcr) => {
// TODO measure:
lfcr.buf.unsplit(fcr.buf);
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);
Ready(Some(Ok(fcr)))
} else {
//info!("with left not enough len {} need_min {}", buf.len(), self.need_min);
self.left.replace(fcr);
again = true;
Pending
break if self.completed {
panic!("NeedMinBuffer poll_next on completed");
} else if self.errored {
self.completed = true;
return Ready(None);
} else {
match self.inp.poll_next_unpin(cx) {
Ready(Some(Ok(fcr))) => {
self.buf_len_histo.ingest(fcr.buf.len() as u32);
//info!("NeedMinBuffer got buf len {}", fcr.buf.len());
match self.left.take() {
Some(mut lfcr) => {
// TODO measure:
lfcr.buf.unsplit(fcr.buf);
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);
Ready(Some(Ok(fcr)))
} else {
//info!("with left not enough len {} need_min {}", buf.len(), self.need_min);
self.left.replace(fcr);
continue;
}
}
}
None => {
if fcr.buf.len() as u32 >= self.need_min {
//info!("simply ready len {} need_min {}", buf.len(), self.need_min);
Ready(Some(Ok(fcr)))
} else {
//info!("no previous leftover, need more len {} need_min {}", buf.len(), self.need_min);
self.left.replace(fcr);
again = true;
Pending
None => {
if fcr.buf.len() as u32 >= self.need_min {
//info!("simply ready len {} need_min {}", buf.len(), self.need_min);
Ready(Some(Ok(fcr)))
} else {
//info!("no previous leftover, need more len {} need_min {}", buf.len(), self.need_min);
self.left.replace(fcr);
continue;
}
}
}
}
Ready(Some(Err(e))) => {
self.errored = true;
Ready(Some(Err(e.into())))
}
Ready(None) => {
info!("NeedMinBuffer histo: {:?}", self.buf_len_histo);
Ready(None)
}
Pending => Pending,
}
Ready(Some(Err(e))) => Ready(Some(Err(e.into()))),
Ready(None) => {
info!("NeedMinBuffer histo: {:?}", self.buf_len_histo);
Ready(None)
}
Pending => Pending,
};
if !again {
break z;
}
}
}
}

1
disk/src/merge.rs
View File

@@ -56,6 +56,7 @@ where
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 {

172
disk/src/rangefilter.rs
View File

@@ -9,10 +9,13 @@ use std::task::{Context, Poll};
pub struct RangeFilter<S, ITY> {
inp: S,
range: NanoRange,
range_str: String,
expand: bool,
prerange: ITY,
have_pre: bool,
have_range_complete: bool,
emitted_post: bool,
data_done: bool,
done: bool,
complete: bool,
}
@@ -22,19 +25,109 @@ where
ITY: Appendable,
{
pub fn new(inp: S, range: NanoRange, expand: bool) -> Self {
trace!("RangeFilter::new range: {:?} expand: {:?}", range, expand);
Self {
inp,
range_str: format!("{:?}", range),
range,
expand,
prerange: ITY::empty(),
have_pre: false,
have_range_complete: false,
emitted_post: false,
data_done: false,
done: false,
complete: false,
}
}
}
impl<S, ITY> RangeFilter<S, ITY>
where
S: Stream<Item = Sitemty<ITY>> + Unpin,
ITY: WithTimestamps + PushableIndex + Appendable + Clearable + Unpin,
{
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<<Self as Stream>::Item>> {
use Poll::*;
loop {
break if self.complete {
panic!("poll_next on complete");
} else if self.done {
self.complete = true;
Ready(None)
} else if self.data_done {
self.done = true;
if self.have_range_complete {
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
} else {
continue;
}
} else {
match self.inp.poll_next_unpin(cx) {
Ready(Some(item)) => match item {
Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))) => {
let mut ret = ITY::empty();
for i1 in 0..item.len() {
let ts = item.ts(i1);
if ts < self.range.beg {
if self.expand {
self.prerange.clear();
self.prerange.push_index(&item, i1);
self.have_pre = true;
}
} else if ts >= self.range.end {
self.have_range_complete = true;
if self.expand {
if self.have_pre {
ret.push_index(&self.prerange, 0);
self.prerange.clear();
self.have_pre = false;
}
if !self.emitted_post {
self.emitted_post = true;
ret.push_index(&item, i1);
//self.data_done = true;
}
} else {
//self.data_done = true;
}
} else {
if self.expand {
if self.have_pre {
ret.push_index(&self.prerange, 0);
self.prerange.clear();
self.have_pre = false;
}
}
ret.push_index(&item, i1);
};
}
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(ret)))))
}
Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete)) => {
self.have_range_complete = true;
continue;
}
k => Ready(Some(k)),
},
Ready(None) => {
self.data_done = true;
if self.have_pre {
let mut ret = ITY::empty();
ret.push_index(&self.prerange, 0);
self.have_pre = false;
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(ret)))))
} else {
continue;
}
}
Pending => Pending,
}
};
}
}
}
impl<S, ITY> Stream for RangeFilter<S, ITY>
where
S: Stream<Item = Sitemty<ITY>> + Unpin,
@@ -42,80 +135,9 @@ where
{
type Item = Sitemty<ITY>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
if self.complete {
panic!("poll_next on complete");
} else if self.done {
self.complete = true;
Ready(None)
} else {
match self.inp.poll_next_unpin(cx) {
Ready(Some(item)) => match item {
Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))) => {
let mut ret = ITY::empty();
for i1 in 0..item.len() {
let ts = item.ts(i1);
if ts < self.range.beg {
if self.expand {
self.prerange.clear();
self.prerange.push_index(&item, i1);
self.have_pre = true;
} else {
};
} else if ts >= self.range.end {
if self.expand {
if self.have_pre {
ret.push_index(&self.prerange, 0);
self.prerange.clear();
self.have_pre = false;
};
if !self.emitted_post {
self.emitted_post = true;
ret.push_index(&item, i1);
self.done = true;
} else {
panic!();
};
} else {
self.done = true;
};
} else {
if self.expand {
if self.have_pre {
ret.push_index(&self.prerange, 0);
self.prerange.clear();
self.have_pre = false;
}
ret.push_index(&item, i1);
} else {
ret.push_index(&item, i1);
};
};
}
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(ret)))))
}
Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete)) => {
warn!("\n\nRangeFilter got RangeComplete\n");
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
}
k => Ready(Some(k)),
},
Ready(None) => {
if self.have_pre {
let mut ret = ITY::empty();
ret.push_index(&self.prerange, 0);
self.have_pre = false;
self.done = true;
Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(ret)))))
} else {
self.done = true;
self.complete = true;
Ready(None)
}
}
Pending => Pending,
}
}
fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
let span1 = span!(Level::INFO, "RangeFilter", range = tracing::field::Empty);
span1.record("range", &self.range_str.as_str());
span1.in_scope(|| Self::poll_next(self, cx))
}
}