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

Refactor more

Browse Source
This commit is contained in:
Dominik Werder
2021-04-26 16:39:23 +02:00
parent 9e60831d88
commit bf376af81e
16 changed files with 827 additions and 755 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
disk/src/binnedstream.rs Normal file
View File

@@ -0,0 +1,54 @@
use crate::agg::MinMaxAvgScalarBinBatch;
use crate::cache::pbvfs::PreBinnedValueFetchedStream;
use err::Error;
use futures_core::Stream;
use futures_util::StreamExt;
#[allow(unused_imports)]
use netpod::log::*;
use netpod::RetStreamExt;
use netpod::{AggKind, Channel, NodeConfig, PreBinnedPatchIterator};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
pub struct BinnedStream {
inp: Pin<Box<dyn Stream<Item = Result<MinMaxAvgScalarBinBatch, Error>> + Send>>,
}
impl BinnedStream {
pub fn new(
patch_it: PreBinnedPatchIterator,
channel: Channel,
agg_kind: AggKind,
node_config: Arc<NodeConfig>,
) -> Self {
warn!("BinnedStream will open a PreBinnedValueStream");
let inp = futures_util::stream::iter(patch_it)
.map(move |coord| {
PreBinnedValueFetchedStream::new(coord, channel.clone(), agg_kind.clone(), node_config.clone())
})
.flatten()
.only_first_error()
.map(|k| {
match k {
Ok(ref k) => {
trace!("BinnedStream got good item {:?}", k);
}
Err(_) => {
error!("\n\n----------------------------------------------------- BinnedStream got error")
}
}
k
});
Self { inp: Box::pin(inp) }
}
}
impl Stream for BinnedStream {
// TODO make this generic over all possible things
type Item = Result<MinMaxAvgScalarBinBatch, Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
self.inp.poll_next_unpin(cx)
}
}

157
disk/src/cache.rs
View File

@@ -1,6 +1,9 @@
use crate::agg::{IntoBinnedT, MinMaxAvgScalarBinBatch, MinMaxAvgScalarEventBatch};
use crate::binnedstream::BinnedStream;
use crate::cache::pbvfs::{PreBinnedHttpFrame, PreBinnedValueFetchedStream};
use crate::frame::makeframe::make_frame;
use crate::merge::MergedMinMaxAvgScalarStream;
use crate::raw::{EventsQuery, FrameType, InMemoryFrameAsyncReadStream};
use crate::raw::EventsQuery;
use bytes::Bytes;
use chrono::{DateTime, Utc};
use err::Error;
@@ -9,7 +12,7 @@ use futures_util::{pin_mut, FutureExt, StreamExt, TryStreamExt};
use hyper::Response;
use netpod::{
AggKind, BinSpecDimT, Channel, Cluster, NanoRange, NodeConfig, PreBinnedPatchCoord, PreBinnedPatchIterator,
PreBinnedPatchRange, RetStreamExt, ToNanos,
PreBinnedPatchRange, ToNanos,
};
use serde::{Deserialize, Serialize};
use std::future::{ready, Future};
@@ -21,6 +24,8 @@ use tokio::io::{AsyncRead, ReadBuf};
#[allow(unused_imports)]
use tracing::{debug, error, info, trace, warn};
pub mod pbvfs;
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Query {
range: NanoRange,
@@ -144,7 +149,7 @@ impl Stream for BinnedBytesForHttpStream {
return Ready(None);
}
match self.inp.poll_next_unpin(cx) {
Ready(Some(item)) => match super::raw::make_frame::<BinnedBytesForHttpStreamFrame>(&item) {
Ready(Some(item)) => match make_frame::<BinnedBytesForHttpStreamFrame>(&item) {
Ok(buf) => Ready(Some(Ok(buf.freeze()))),
Err(e) => {
self.errored = true;
@@ -208,7 +213,6 @@ pub struct PreBinnedValueByteStream {
impl PreBinnedValueByteStream {
pub fn new(patch: PreBinnedPatchCoord, channel: Channel, agg_kind: AggKind, node_config: Arc<NodeConfig>) -> Self {
warn!("PreBinnedValueByteStream");
Self {
inp: PreBinnedValueStream::new(patch, channel, agg_kind, node_config),
errored: false,
@@ -230,7 +234,7 @@ impl Stream for PreBinnedValueByteStream {
return Ready(None);
}
match self.inp.poll_next_unpin(cx) {
Ready(Some(item)) => match super::raw::make_frame::<PreBinnedHttpFrame>(&item) {
Ready(Some(item)) => match make_frame::<PreBinnedHttpFrame>(&item) {
Ok(buf) => Ready(Some(Ok(buf.freeze()))),
Err(e) => {
self.errored = true;
@@ -416,98 +420,6 @@ impl Stream for PreBinnedValueStream {
}
}
pub struct PreBinnedValueFetchedStream {
uri: http::Uri,
resfut: Option<hyper::client::ResponseFuture>,
res: Option<InMemoryFrameAsyncReadStream<HttpBodyAsAsyncRead>>,
}
impl PreBinnedValueFetchedStream {
pub fn new(
patch_coord: PreBinnedPatchCoord,
channel: Channel,
agg_kind: AggKind,
node_config: Arc<NodeConfig>,
) -> Self {
let nodeix = node_ix_for_patch(&patch_coord, &channel, &node_config.cluster);
let node = &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!(
"http://{}:{}/api/1/prebinned?{}&channel_backend={}&channel_name={}&agg_kind={:?}",
node.host,
node.port,
patch_coord.to_url_params_strings(),
channel.backend,
channel.name,
agg_kind,
)
.parse()
.unwrap();
Self {
uri,
resfut: None,
res: None,
}
}
}
// TODO use a newtype here to use a different FRAME_TYPE_ID compared to
// impl FrameType for BinnedBytesForHttpStreamFrame
pub type PreBinnedHttpFrame = Result<MinMaxAvgScalarBinBatch, Error>;
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;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
'outer: loop {
break if let Some(res) = self.res.as_mut() {
pin_mut!(res);
match res.poll_next(cx) {
Ready(Some(Ok(frame))) => {
assert!(frame.tyid() == <PreBinnedHttpFrame as FrameType>::FRAME_TYPE_ID);
match bincode::deserialize::<PreBinnedHttpFrame>(frame.buf()) {
Ok(item) => Ready(Some(item)),
Err(e) => Ready(Some(Err(e.into()))),
}
}
Ready(Some(Err(e))) => Ready(Some(Err(e.into()))),
Ready(None) => Ready(None),
Pending => Pending,
}
} else if let Some(resfut) = self.resfut.as_mut() {
match resfut.poll_unpin(cx) {
Ready(res) => match res {
Ok(res) => {
info!("PreBinnedValueFetchedStream GOT result from SUB REQUEST: {:?}", res);
let s1 = HttpBodyAsAsyncRead::new(res);
let s2 = InMemoryFrameAsyncReadStream::new(s1);
self.res = Some(s2);
continue 'outer;
}
Err(e) => {
error!("PreBinnedValueStream error in stream {:?}", e);
Ready(Some(Err(e.into())))
}
},
Pending => Pending,
}
} else {
let req = hyper::Request::builder()
.method(http::Method::GET)
.uri(&self.uri)
.body(hyper::Body::empty())?;
let client = hyper::Client::new();
info!("PreBinnedValueFetchedStream START REQUEST FOR {:?}", req);
self.resfut = Some(client.request(req));
continue 'outer;
};
}
}
}
pub struct HttpBodyAsAsyncRead {
inp: Response<hyper::Body>,
left: Bytes,
@@ -659,7 +571,7 @@ impl Stream for MergedFromRemotes {
self.merged = Some(Box::pin(s1));
} else {
info!(
"MergedFromRemotes conn / estab {} {}",
"MergedFromRemotes raw / estab {} {}",
c1,
self.tcp_establish_futs.len()
);
@@ -671,54 +583,6 @@ impl Stream for MergedFromRemotes {
}
}
pub struct BinnedStream {
inp: Pin<Box<dyn Stream<Item = Result<MinMaxAvgScalarBinBatch, Error>> + Send>>,
}
impl BinnedStream {
pub fn new(
patch_it: PreBinnedPatchIterator,
channel: Channel,
agg_kind: AggKind,
node_config: Arc<NodeConfig>,
) -> Self {
warn!("BinnedStream will open a PreBinnedValueStream");
let inp = futures_util::stream::iter(patch_it)
.map(move |coord| {
PreBinnedValueFetchedStream::new(coord, channel.clone(), agg_kind.clone(), node_config.clone())
})
.flatten()
.only_first_error()
.map(|k| {
match k {
Ok(ref k) => {
info!("BinnedStream got good item {:?}", k);
}
Err(_) => {
error!("\n\n----------------------------------------------------- BinnedStream got error")
}
}
k
});
Self { inp: Box::pin(inp) }
}
}
impl Stream for BinnedStream {
// TODO make this generic over all possible things
type Item = Result<MinMaxAvgScalarBinBatch, Error>;
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(Ok(k))) => Ready(Some(Ok(k))),
Ready(Some(Err(e))) => Ready(Some(Err(e))),
Ready(None) => Ready(None),
Pending => Pending,
}
}
}
pub struct SomeReturnThing {}
impl From<SomeReturnThing> for Bytes {
@@ -739,6 +603,5 @@ pub fn node_ix_for_patch(patch_coord: &PreBinnedPatchCoord, channel: &Channel, c
hash.finalize(&mut out);
let a = [out[0], out[1], out[2], out[3]];
let ix = u32::from_le_bytes(a) % cluster.nodes.len() as u32;
info!("node_ix_for_patch {}", ix);
ix
}

105
disk/src/cache/pbvfs.rs vendored Normal file
View File

@@ -0,0 +1,105 @@
use crate::agg::MinMaxAvgScalarBinBatch;
use crate::cache::{node_ix_for_patch, HttpBodyAsAsyncRead};
use crate::frame::inmem::InMemoryFrameAsyncReadStream;
use crate::frame::makeframe::FrameType;
use err::Error;
use futures_core::Stream;
use futures_util::{pin_mut, FutureExt};
#[allow(unused_imports)]
use netpod::log::*;
use netpod::{AggKind, Channel, NodeConfig, PreBinnedPatchCoord};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
pub struct PreBinnedValueFetchedStream {
uri: http::Uri,
resfut: Option<hyper::client::ResponseFuture>,
res: Option<InMemoryFrameAsyncReadStream<HttpBodyAsAsyncRead>>,
}
impl PreBinnedValueFetchedStream {
pub fn new(
patch_coord: PreBinnedPatchCoord,
channel: Channel,
agg_kind: AggKind,
node_config: Arc<NodeConfig>,
) -> Self {
let nodeix = node_ix_for_patch(&patch_coord, &channel, &node_config.cluster);
let node = &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!(
"http://{}:{}/api/1/prebinned?{}&channel_backend={}&channel_name={}&agg_kind={:?}",
node.host,
node.port,
patch_coord.to_url_params_strings(),
channel.backend,
channel.name,
agg_kind,
)
.parse()
.unwrap();
Self {
uri,
resfut: None,
res: None,
}
}
}
// TODO use a newtype here to use a different FRAME_TYPE_ID compared to
// impl FrameType for BinnedBytesForHttpStreamFrame
pub type PreBinnedHttpFrame = Result<MinMaxAvgScalarBinBatch, Error>;
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;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
'outer: loop {
break if let Some(res) = self.res.as_mut() {
pin_mut!(res);
match res.poll_next(cx) {
Ready(Some(Ok(frame))) => {
assert!(frame.tyid() == <PreBinnedHttpFrame as FrameType>::FRAME_TYPE_ID);
match bincode::deserialize::<PreBinnedHttpFrame>(frame.buf()) {
Ok(item) => Ready(Some(item)),
Err(e) => Ready(Some(Err(e.into()))),
}
}
Ready(Some(Err(e))) => Ready(Some(Err(e.into()))),
Ready(None) => Ready(None),
Pending => Pending,
}
} else if let Some(resfut) = self.resfut.as_mut() {
match resfut.poll_unpin(cx) {
Ready(res) => match res {
Ok(res) => {
info!("PreBinnedValueFetchedStream GOT result from SUB REQUEST: {:?}", res);
let s1 = HttpBodyAsAsyncRead::new(res);
let s2 = InMemoryFrameAsyncReadStream::new(s1);
self.res = Some(s2);
continue 'outer;
}
Err(e) => {
error!("PreBinnedValueStream error in stream {:?}", e);
Ready(Some(Err(e.into())))
}
},
Pending => Pending,
}
} else {
let req = hyper::Request::builder()
.method(http::Method::GET)
.uri(&self.uri)
.body(hyper::Body::empty())?;
let client = hyper::Client::new();
info!("PreBinnedValueFetchedStream START REQUEST FOR {:?}", req);
self.resfut = Some(client.request(req));
continue 'outer;
};
}
}
}

24
disk/src/eventchunker.rs
View File

@@ -4,11 +4,11 @@ use bytes::{Buf, BytesMut};
use err::Error;
use futures_core::Stream;
use futures_util::pin_mut;
#[allow(unused_imports)]
use netpod::log::*;
use netpod::{ChannelConfig, ScalarType, Shape};
use std::pin::Pin;
use std::task::{Context, Poll};
#[allow(unused_imports)]
use tracing::{debug, error, info, span, trace, warn, Level};
pub struct EventChunker {
inp: NeedMinBuffer,
@@ -55,7 +55,7 @@ impl EventChunker {
let mut ret = EventFull::empty();
use byteorder::{ReadBytesExt, BE};
loop {
info!("parse_buf LOOP buf len {} need_min {}", buf.len(), self.need_min);
trace!("parse_buf LOOP buf len {} need_min {}", buf.len(), self.need_min);
if (buf.len() as u32) < self.need_min {
break;
}
@@ -69,7 +69,7 @@ impl EventChunker {
assert!(len > 0 && len < 128, "unexpected data file header");
let totlen = len as usize + 2;
if buf.len() < totlen {
info!("parse_buf not enough A totlen {}", totlen);
debug!("parse_buf not enough A totlen {}", totlen);
self.need_min = totlen as u32;
break;
} else {
@@ -77,7 +77,7 @@ impl EventChunker {
let len2 = sl.read_i32::<BE>().unwrap();
assert!(len == len2, "len mismatch");
let s1 = String::from_utf8(buf.as_ref()[6..(len as usize + 6 - 8)].to_vec()).unwrap();
info!("channel name {} len {} len2 {}", s1, len, len2);
info!("channel name {}", s1);
self.state = DataFileState::Event;
self.need_min = 4;
buf.advance(totlen);
@@ -88,7 +88,7 @@ impl EventChunker {
let len = sl.read_i32::<BE>().unwrap();
assert!(len >= 20 && len < 1024 * 1024 * 10);
let len = len as u32;
info!(
trace!(
"+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+- event len {}",
len,
);
@@ -104,7 +104,7 @@ impl EventChunker {
sl.read_i32::<BE>().unwrap();
sl.read_i64::<BE>().unwrap();
let ts = sl.read_i64::<BE>().unwrap();
info!("parse_buf not enough C len {} have {} ts {}", len, buf.len(), ts);
trace!("parse_buf not enough C len {} have {} ts {}", len, buf.len(), ts);
}
self.need_min = len as u32;
break;
@@ -186,7 +186,7 @@ impl EventChunker {
) {
Ok(c1) => {
assert!(c1 as u32 == k1);
debug!("decompress result c1 {} k1 {}", c1, k1);
trace!("decompress result c1 {} k1 {}", c1, k1);
ret.add_event(ts, pulse, Some(decomp), ScalarType::from_dtype_index(type_index));
}
Err(e) => {
@@ -198,14 +198,14 @@ impl EventChunker {
"TODO uncompressed event parsing not yet implemented"
)))?;
}
info!("advance and reset need_min");
trace!("advance and reset need_min");
buf.advance(len as usize);
self.need_min = 4;
}
}
}
}
info!("AFTER PARSE LOOP len {}", ret.tss.len());
trace!("AFTER PARSE LOOP len {}", ret.tss.len());
Ok(ParseResult { events: ret })
}
}
@@ -254,10 +254,10 @@ impl Stream for EventChunker {
}
let g = &mut self.inp;
pin_mut!(g);
info!("EventChunker call input poll_next");
trace!("EventChunker call input poll_next");
match g.poll_next(cx) {
Ready(Some(Ok(mut buf))) => {
info!("EventChunker got buffer len {}", buf.len());
trace!("EventChunker got buffer len {}", buf.len());
let r = self.parse_buf(&mut buf);
match r {
Ok(res) => {

2
disk/src/frame.rs Normal file
View File

@@ -0,0 +1,2 @@
pub mod inmem;
pub mod makeframe;

287
disk/src/frame/inmem.rs Normal file
View File

@@ -0,0 +1,287 @@
use crate::frame::makeframe::{INMEM_FRAME_HEAD, INMEM_FRAME_MAGIC};
use bytes::{BufMut, Bytes, BytesMut};
use err::Error;
use futures_core::Stream;
use futures_util::pin_mut;
use netpod::log::*;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, ReadBuf};
/**
Interprets a byte stream as length-delimited frames.
Emits each frame as a single item. Therefore, each item must fit easily into memory.
*/
pub struct InMemoryFrameAsyncReadStream<T>
where
T: AsyncRead + Unpin,
{
inp: T,
buf: BytesMut,
bufcap: usize,
wp: usize,
tryparse: bool,
errored: bool,
completed: bool,
inp_bytes_consumed: u64,
}
impl<T> InMemoryFrameAsyncReadStream<T>
where
T: AsyncRead + Unpin,
{
pub fn new(inp: T) -> Self {
// TODO make capacity adjustable.
let bufcap = 512;
let mut t = Self {
inp,
buf: BytesMut::new(),
bufcap: bufcap,
wp: 0,
tryparse: false,
errored: false,
completed: false,
inp_bytes_consumed: 0,
};
t.buf = t.empty_buf();
t
}
fn empty_buf(&self) -> BytesMut {
let mut buf = BytesMut::with_capacity(self.bufcap);
buf.resize(buf.capacity(), 0);
buf
}
fn poll_upstream(&mut self, cx: &mut Context) -> Poll<Result<usize, Error>> {
if self.wp > 0 {
// TODO copy only if we gain capacity in the current buffer.
// Also copy if the bufcap got increased: how to find out with BytesMut? Question about how capacity is defined exactly...
// Avoid copies after e.g. after a previous Pending.
let mut bnew = self.empty_buf();
assert!(self.buf.len() >= self.wp);
assert!(bnew.capacity() >= self.wp);
trace!(
"InMemoryFrameAsyncReadStream re-use {} bytes from previous i/o",
self.wp,
);
bnew[0..].as_mut().put_slice(&self.buf[..self.wp]);
self.buf = bnew;
}
trace!(
".............. PREPARE READ FROM wp {} self.buf.len() {}",
self.wp,
self.buf.len(),
);
let gg = self.buf.len() - self.wp;
let mut buf2 = ReadBuf::new(&mut self.buf[self.wp..]);
assert!(gg > 0);
assert!(buf2.remaining() == gg);
assert!(buf2.capacity() == gg);
assert!(buf2.filled().len() == 0);
let j = &mut self.inp;
pin_mut!(j);
use Poll::*;
match AsyncRead::poll_read(j, cx, &mut buf2) {
Ready(Ok(_)) => {
let n1 = buf2.filled().len();
trace!("InMemoryFrameAsyncReadStream READ {} FROM UPSTREAM", n1);
Ready(Ok(n1))
}
Ready(Err(e)) => Ready(Err(e.into())),
Pending => Pending,
}
}
fn tryparse(
&mut self,
buf: BytesMut,
wp: usize,
) -> (Option<Option<Result<InMemoryFrame, Error>>>, BytesMut, usize) {
trace!(
"+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ tryparse with buf.len() {} wp {}",
buf.len(),
wp
);
const HEAD: usize = INMEM_FRAME_HEAD;
let mut buf = buf;
let nb = wp;
if nb >= HEAD {
let magic = u32::from_le_bytes(*arrayref::array_ref![buf, 0, 4]);
let encid = u32::from_le_bytes(*arrayref::array_ref![buf, 4, 4]);
let tyid = u32::from_le_bytes(*arrayref::array_ref![buf, 8, 4]);
let len = u32::from_le_bytes(*arrayref::array_ref![buf, 12, 4]);
if magic != INMEM_FRAME_MAGIC {
error!("InMemoryFrameAsyncReadStream tryparse incorrect magic: {}", magic);
return (
Some(Some(Err(Error::with_msg(format!(
"InMemoryFrameAsyncReadStream tryparse incorrect magic: {}",
magic
))))),
buf,
wp,
);
}
trace!("tryparse len {}", len);
if len == 0 {
if nb != HEAD {
return (
Some(Some(Err(Error::with_msg(format!(
"InMemoryFrameAsyncReadStream tryparse unexpected amount left {}",
nb
))))),
buf,
wp,
);
}
(Some(None), buf, wp)
} else {
if len > 1024 * 32 {
warn!("InMemoryFrameAsyncReadStream big len received {}", len);
}
if len > 1024 * 1024 * 2 {
error!("InMemoryFrameAsyncReadStream too long len {}", len);
return (
Some(Some(Err(Error::with_msg(format!(
"InMemoryFrameAsyncReadStream tryparse huge buffer len {} self.inp_bytes_consumed {}",
len, self.inp_bytes_consumed
))))),
buf,
wp,
);
}
if len == 0 && len > 1024 * 512 {
return (
Some(Some(Err(Error::with_msg(format!(
"InMemoryFrameAsyncReadStream tryparse len {} self.inp_bytes_consumed {}",
len, self.inp_bytes_consumed
))))),
buf,
wp,
);
}
let nl = len as usize + HEAD;
if self.bufcap < nl {
// TODO count cases in production
let n = 2 * nl;
warn!("Adjust bufcap old {} new {}", self.bufcap, n);
self.bufcap = n;
}
if nb >= nl {
use bytes::Buf;
let mut buf3 = buf.split_to(nl);
buf3.advance(HEAD);
self.inp_bytes_consumed += nl as u64;
let ret = InMemoryFrame {
len,
tyid,
encid,
buf: buf3.freeze(),
};
(Some(Some(Ok(ret))), buf, wp - nl)
} else {
(None, buf, wp)
}
}
} else {
(None, buf, wp)
}
}
}
pub struct InMemoryFrame {
encid: u32,
tyid: u32,
len: u32,
buf: Bytes,
}
impl InMemoryFrame {
pub fn encid(&self) -> u32 {
self.encid
}
pub fn tyid(&self) -> u32 {
self.tyid
}
pub fn len(&self) -> u32 {
self.len
}
pub fn buf(&self) -> &Bytes {
&self.buf
}
}
impl<T> Stream for InMemoryFrameAsyncReadStream<T>
where
T: AsyncRead + Unpin,
{
type Item = Result<InMemoryFrame, Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
trace!("InMemoryFrameAsyncReadStream poll_next");
use Poll::*;
assert!(!self.completed);
if self.errored {
self.completed = true;
return Ready(None);
}
'outer: loop {
if self.tryparse {
let r = {
let buf = std::mem::replace(&mut self.buf, BytesMut::new());
let wp = self.wp;
let (r, buf, wp) = self.tryparse(buf, wp);
self.buf = buf;
self.wp = wp;
r
};
break match r {
None => {
self.tryparse = false;
continue 'outer;
}
Some(None) => {
self.tryparse = false;
self.completed = true;
Ready(None)
}
Some(Some(Ok(k))) => Ready(Some(Ok(k))),
Some(Some(Err(e))) => {
self.tryparse = false;
self.errored = true;
Ready(Some(Err(e)))
}
};
} else {
let r = self.poll_upstream(cx);
break match r {
Ready(Ok(n1)) => {
trace!("poll_upstream GIVES Ready {}", n1);
self.wp += n1;
if n1 == 0 {
let n2 = self.buf.len();
if n2 != 0 {
warn!(
"InMemoryFrameAsyncReadStream n2 != 0 n2 {} consumed {} ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~",
n2, self.inp_bytes_consumed
);
}
self.completed = true;
Ready(None)
} else {
self.tryparse = true;
continue 'outer;
}
}
Ready(Err(e)) => {
trace!("poll_upstream GIVES Error");
self.errored = true;
Ready(Some(Err(e.into())))
}
Pending => Pending,
};
}
}
}
}

57
disk/src/frame/makeframe.rs Normal file
View File

@@ -0,0 +1,57 @@
use crate::cache::BinnedBytesForHttpStreamFrame;
use crate::raw::conn::RawConnOut;
use crate::raw::EventQueryJsonStringFrame;
use bytes::{BufMut, BytesMut};
use err::Error;
use serde::Serialize;
pub const INMEM_FRAME_HEAD: usize = 16;
pub const INMEM_FRAME_MAGIC: u32 = 0xc6c3b73d;
pub trait FrameType {
const FRAME_TYPE_ID: u32;
}
impl FrameType for BinnedBytesForHttpStreamFrame {
const FRAME_TYPE_ID: u32 = 0x02;
}
impl FrameType for EventQueryJsonStringFrame {
const FRAME_TYPE_ID: u32 = 0x03;
}
impl FrameType for RawConnOut {
const FRAME_TYPE_ID: u32 = 0x04;
}
pub fn make_frame<FT>(item: &FT) -> Result<BytesMut, Error>
where
FT: FrameType + Serialize,
{
match bincode::serialize(item) {
Ok(enc) => {
if enc.len() > u32::MAX as usize {
return Err(Error::with_msg(format!("too long payload {}", enc.len())));
}
let encid = 0x12121212;
let mut buf = BytesMut::with_capacity(enc.len() + INMEM_FRAME_HEAD);
buf.put_u32_le(INMEM_FRAME_MAGIC);
buf.put_u32_le(encid);
buf.put_u32_le(FT::FRAME_TYPE_ID);
buf.put_u32_le(enc.len() as u32);
buf.put(enc.as_ref());
Ok(buf)
}
Err(e) => Err(e)?,
}
}
pub fn make_term_frame() -> BytesMut {
let encid = 0x12121313;
let mut buf = BytesMut::with_capacity(INMEM_FRAME_HEAD);
buf.put_u32_le(INMEM_FRAME_MAGIC);
buf.put_u32_le(encid);
buf.put_u32_le(0x01);
buf.put_u32_le(0);
buf
}

2
disk/src/lib.rs
View File

@@ -18,10 +18,12 @@ use tracing::{debug, error, info, span, trace, warn, Level};
pub mod agg;
#[cfg(test)]
pub mod aggtest;
pub mod binnedstream;
pub mod cache;
pub mod channelconfig;
pub mod eventblobs;
pub mod eventchunker;
pub mod frame;
pub mod gen;
pub mod merge;
pub mod raw;

6
disk/src/merge.rs
View File

@@ -147,6 +147,7 @@ where
ixs: Vec<usize>,
emitted_complete: bool,
batch: MinMaxAvgScalarEventBatch,
ts_last_emit: u64,
}
impl<S> MergedMinMaxAvgScalarStream<S>
@@ -165,6 +166,7 @@ where
ixs: vec![0; n],
emitted_complete: false,
batch: MinMaxAvgScalarEventBatch::empty(),
ts_last_emit: 0,
}
}
}
@@ -241,7 +243,9 @@ where
break Ready(None);
}
} else {
info!("decided on next lowest ts {} ix {}", lowest_ts, lowest_ix);
trace!("decided on next lowest ts {} ix {}", lowest_ts, lowest_ix);
assert!(lowest_ts >= self.ts_last_emit);
self.ts_last_emit = lowest_ts;
self.batch.tss.push(lowest_ts);
let rix = self.ixs[lowest_ix];
let z = match &self.current[lowest_ix] {

598
disk/src/raw.rs
View File

@@ -5,26 +5,24 @@ Delivers event data (not yet time-binned) from local storage and provides client
to request such data from nodes.
*/
use crate::agg::{IntoBinnedXBins1, IntoDim1F32Stream, MinMaxAvgScalarEventBatch};
use crate::cache::BinnedBytesForHttpStreamFrame;
use bytes::{BufMut, Bytes, BytesMut};
use crate::agg::MinMaxAvgScalarEventBatch;
use crate::frame::inmem::InMemoryFrameAsyncReadStream;
use crate::frame::makeframe::{make_frame, make_term_frame};
use crate::raw::bffr::MinMaxAvgScalarEventBatchStreamFromFrames;
use err::Error;
use futures_core::Stream;
use futures_util::{pin_mut, StreamExt};
use netpod::timeunits::DAY;
use netpod::{AggKind, Channel, NanoRange, Node, NodeConfig, ScalarType, Shape};
use netpod::{AggKind, Channel, NanoRange, Node};
use serde::{Deserialize, Serialize};
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, AsyncWriteExt, ReadBuf};
use tokio::net::tcp::OwnedWriteHalf;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use tracing::Instrument;
#[allow(unused_imports)]
use tracing::{debug, error, info, span, trace, warn, Level};
pub mod bffr;
pub mod conn;
/**
Query parameters to request (optionally) X-processed, but not T-processed events.
*/
@@ -57,584 +55,6 @@ pub async fn x_processed_stream_from_node(
Ok(s3)
}
pub struct MinMaxAvgScalarEventBatchStreamFromFrames<T>
where
T: AsyncRead + Unpin,
{
inp: InMemoryFrameAsyncReadStream<T>,
}
impl<T> MinMaxAvgScalarEventBatchStreamFromFrames<T>
where
T: AsyncRead + Unpin,
{
pub fn new(inp: InMemoryFrameAsyncReadStream<T>) -> Self {
Self { inp }
}
}
impl<T> Stream for MinMaxAvgScalarEventBatchStreamFromFrames<T>
where
T: AsyncRead + Unpin,
{
type Item = Result<MinMaxAvgScalarEventBatch, Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
loop {
let j = &mut self.inp;
pin_mut!(j);
break match j.poll_next(cx) {
Ready(Some(Ok(frame))) => {
type ExpectedType = RawConnOut;
info!(
"MinMaxAvgScalarEventBatchStreamFromFrames got full frame buf {}",
frame.buf().len()
);
assert!(frame.tyid() == <ExpectedType as FrameType>::FRAME_TYPE_ID);
match bincode::deserialize::<ExpectedType>(frame.buf()) {
Ok(item) => match item {
Ok(item) => Ready(Some(Ok(item))),
Err(e) => Ready(Some(Err(e))),
},
Err(e) => {
trace!(
"MinMaxAvgScalarEventBatchStreamFromFrames ~~~~~~~~ ERROR on frame payload {}",
frame.buf().len(),
);
Ready(Some(Err(e.into())))
}
}
}
Ready(Some(Err(e))) => Ready(Some(Err(e))),
Ready(None) => Ready(None),
Pending => Pending,
};
}
}
}
pub const INMEM_FRAME_HEAD: usize = 16;
pub const INMEM_FRAME_MAGIC: u32 = 0xc6c3b73d;
/**
Interprets a byte stream as length-delimited frames.
Emits each frame as a single item. Therefore, each item must fit easily into memory.
*/
pub struct InMemoryFrameAsyncReadStream<T>
where
T: AsyncRead + Unpin,
{
inp: T,
buf: BytesMut,
bufcap: usize,
wp: usize,
tryparse: bool,
errored: bool,
completed: bool,
inp_bytes_consumed: u64,
}
impl<T> InMemoryFrameAsyncReadStream<T>
where
T: AsyncRead + Unpin,
{
pub fn new(inp: T) -> Self {
// TODO make capacity adjustable.
let bufcap = 512;
let mut t = Self {
inp,
buf: BytesMut::new(),
bufcap: bufcap,
wp: 0,
tryparse: false,
errored: false,
completed: false,
inp_bytes_consumed: 0,
};
t.buf = t.empty_buf();
t
}
fn empty_buf(&self) -> BytesMut {
let mut buf = BytesMut::with_capacity(self.bufcap);
buf.resize(buf.capacity(), 0);
buf
}
fn poll_upstream(&mut self, cx: &mut Context) -> Poll<Result<usize, Error>> {
if self.wp > 0 {
// TODO copy only if we gain capacity in the current buffer.
// Also copy if the bufcap got increased: how to find out with BytesMut? Question about how capacity is defined exactly...
// Avoid copies after e.g. after a previous Pending.
let mut bnew = self.empty_buf();
assert!(self.buf.len() >= self.wp);
assert!(bnew.capacity() >= self.wp);
info!(
"InMemoryFrameAsyncReadStream re-use {} bytes from previous i/o",
self.wp,
);
bnew[0..].as_mut().put(&self.buf[..self.wp]);
self.buf = bnew;
}
info!(
".............. PREPARE READ FROM wp {} self.buf.len() {}",
self.wp,
self.buf.len(),
);
let gg = self.buf.len() - self.wp;
let mut buf2 = ReadBuf::new(&mut self.buf[self.wp..]);
assert!(gg > 0);
assert!(buf2.remaining() == gg);
assert!(buf2.capacity() == gg);
assert!(buf2.filled().len() == 0);
let j = &mut self.inp;
pin_mut!(j);
use Poll::*;
match AsyncRead::poll_read(j, cx, &mut buf2) {
Ready(Ok(_)) => {
let n1 = buf2.filled().len();
info!("InMemoryFrameAsyncReadStream READ {} FROM UPSTREAM", n1);
Ready(Ok(n1))
}
Ready(Err(e)) => Ready(Err(e.into())),
Pending => Pending,
}
}
fn tryparse(
&mut self,
buf: BytesMut,
wp: usize,
) -> (Option<Option<Result<InMemoryFrame, Error>>>, BytesMut, usize) {
info!(
"+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ tryparse with buf.len() {} wp {}",
buf.len(),
wp
);
const HEAD: usize = INMEM_FRAME_HEAD;
let mut buf = buf;
let nb = wp;
if nb >= HEAD {
let magic = u32::from_le_bytes(*arrayref::array_ref![buf, 0, 4]);
let encid = u32::from_le_bytes(*arrayref::array_ref![buf, 4, 4]);
let tyid = u32::from_le_bytes(*arrayref::array_ref![buf, 8, 4]);
let len = u32::from_le_bytes(*arrayref::array_ref![buf, 12, 4]);
if magic != INMEM_FRAME_MAGIC {
error!("InMemoryFrameAsyncReadStream tryparse incorrect magic: {}", magic);
return (
Some(Some(Err(Error::with_msg(format!(
"InMemoryFrameAsyncReadStream tryparse incorrect magic: {}",
magic
))))),
buf,
wp,
);
}
info!("\n\ntryparse len {}\n\n", len);
if len == 0 {
if nb != HEAD {
return (
Some(Some(Err(Error::with_msg(format!(
"InMemoryFrameAsyncReadStream tryparse unexpected amount left {}",
nb
))))),
buf,
wp,
);
}
(Some(None), buf, wp)
} else {
if len > 1024 * 32 {
warn!("InMemoryFrameAsyncReadStream big len received {}", len);
}
if len > 1024 * 1024 * 2 {
error!("InMemoryFrameAsyncReadStream too long len {}", len);
return (
Some(Some(Err(Error::with_msg(format!(
"InMemoryFrameAsyncReadStream tryparse huge buffer len {} self.inp_bytes_consumed {}",
len, self.inp_bytes_consumed
))))),
buf,
wp,
);
}
if len == 0 && len > 1024 * 512 {
return (
Some(Some(Err(Error::with_msg(format!(
"InMemoryFrameAsyncReadStream tryparse len {} self.inp_bytes_consumed {}",
len, self.inp_bytes_consumed
))))),
buf,
wp,
);
}
let nl = len as usize + HEAD;
if self.bufcap < nl {
// TODO count cases in production
let n = 2 * nl;
warn!(
"eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee ADJUST bufcap old {} new {}",
self.bufcap, n
);
self.bufcap = n;
}
if nb >= nl {
use bytes::Buf;
let mut buf3 = buf.split_to(nl);
buf3.advance(HEAD);
self.inp_bytes_consumed += nl as u64;
let ret = InMemoryFrame {
len,
tyid,
encid,
buf: buf3.freeze(),
};
(Some(Some(Ok(ret))), buf, wp - nl)
} else {
(None, buf, wp)
}
}
} else {
(None, buf, wp)
}
}
}
pub struct InMemoryFrame {
encid: u32,
tyid: u32,
len: u32,
buf: Bytes,
}
impl InMemoryFrame {
pub fn encid(&self) -> u32 {
self.encid
}
pub fn tyid(&self) -> u32 {
self.tyid
}
pub fn len(&self) -> u32 {
self.len
}
pub fn buf(&self) -> &Bytes {
&self.buf
}
}
impl<T> Stream for InMemoryFrameAsyncReadStream<T>
where
T: AsyncRead + Unpin,
{
type Item = Result<InMemoryFrame, Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
info!("InMemoryFrameAsyncReadStream poll_next");
use Poll::*;
assert!(!self.completed);
if self.errored {
self.completed = true;
return Ready(None);
}
'outer: loop {
if self.tryparse {
let r = {
let buf = std::mem::replace(&mut self.buf, BytesMut::new());
let wp = self.wp;
let (r, buf, wp) = self.tryparse(buf, wp);
self.buf = buf;
self.wp = wp;
r
};
break match r {
None => {
self.tryparse = false;
continue 'outer;
}
Some(None) => {
self.tryparse = false;
self.completed = true;
Ready(None)
}
Some(Some(Ok(k))) => Ready(Some(Ok(k))),
Some(Some(Err(e))) => {
self.tryparse = false;
self.errored = true;
Ready(Some(Err(e)))
}
};
} else {
let r = self.poll_upstream(cx);
break match r {
Ready(Ok(n1)) => {
info!("poll_upstream GIVES Ready {}", n1);
self.wp += n1;
if n1 == 0 {
let n2 = self.buf.len();
if n2 != 0 {
warn!(
"InMemoryFrameAsyncReadStream n2 != 0 n2 {} consumed {} ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~",
n2, self.inp_bytes_consumed
);
}
self.completed = true;
Ready(None)
} else {
self.tryparse = true;
continue 'outer;
}
}
Ready(Err(e)) => {
info!("poll_upstream GIVES Error");
self.errored = true;
Ready(Some(Err(e.into())))
}
Pending => {
info!("poll_upstream GIVES Pending");
Pending
}
};
}
}
}
}
pub trait FrameType {
const FRAME_TYPE_ID: u32;
}
impl FrameType for BinnedBytesForHttpStreamFrame {
const FRAME_TYPE_ID: u32 = 0x02;
}
impl FrameType for EventQueryJsonStringFrame {
const FRAME_TYPE_ID: u32 = 0x03;
}
impl FrameType for RawConnOut {
const FRAME_TYPE_ID: u32 = 0x04;
}
pub fn make_frame<FT>(item: &FT) -> Result<BytesMut, Error>
where
FT: FrameType + Serialize,
{
match bincode::serialize(item) {
Ok(enc) => {
if enc.len() > u32::MAX as usize {
return Err(Error::with_msg(format!("too long payload {}", enc.len())));
}
let encid = 0x12121212;
let mut buf = BytesMut::with_capacity(enc.len() + INMEM_FRAME_HEAD);
buf.put_u32_le(INMEM_FRAME_MAGIC);
buf.put_u32_le(encid);
buf.put_u32_le(FT::FRAME_TYPE_ID);
buf.put_u32_le(enc.len() as u32);
buf.put(enc.as_ref());
Ok(buf)
}
Err(e) => Err(e)?,
}
}
pub fn make_term_frame() -> BytesMut {
let encid = 0x12121313;
let mut buf = BytesMut::with_capacity(INMEM_FRAME_HEAD);
buf.put_u32_le(INMEM_FRAME_MAGIC);
buf.put_u32_le(encid);
buf.put_u32_le(0x01);
buf.put_u32_le(0);
buf
}
pub async fn raw_service(node_config: Arc<NodeConfig>) -> Result<(), Error> {
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)) => {
taskrun::spawn(raw_conn_handler(stream, addr, node_config.clone()));
}
Err(e) => Err(e)?,
}
}
}
async fn raw_conn_handler(stream: TcpStream, addr: SocketAddr, node_config: Arc<NodeConfig>) -> 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)
.instrument(span1)
.await;
match r {
Ok(k) => Ok(k),
Err(e) => {
error!("raw_conn_handler sees error: {:?}", e);
Err(e)
}
}
}
type RawConnOut = Result<MinMaxAvgScalarEventBatch, Error>;
async fn raw_conn_handler_inner(
stream: TcpStream,
addr: SocketAddr,
node_config: Arc<NodeConfig>,
) -> Result<(), Error> {
match raw_conn_handler_inner_try(stream, addr, node_config).await {
Ok(_) => (),
Err(mut ce) => {
/*error!(
"raw_conn_handler_inner CAUGHT ERROR AND TRY TO SEND OVER TCP {:?}",
ce.err
);*/
let buf = make_frame::<RawConnOut>(&Err(ce.err))?;
match ce.netout.write(&buf).await {
Ok(_) => (),
Err(e) => return Err(e)?,
}
}
}
Ok(())
}
struct ConnErr {
err: Error,
netout: OwnedWriteHalf,
}
impl<E: Into<Error>> From<(E, OwnedWriteHalf)> for ConnErr {
fn from((err, netout): (E, OwnedWriteHalf)) -> Self {
Self {
err: err.into(),
netout,
}
}
}
async fn raw_conn_handler_inner_try(
stream: TcpStream,
addr: SocketAddr,
node_config: Arc<NodeConfig>,
) -> Result<(), ConnErr> {
info!("raw_conn_handler SPAWNED for {:?}", addr);
let (netin, mut netout) = stream.into_split();
let mut h = InMemoryFrameAsyncReadStream::new(netin);
let mut frames = vec![];
while let Some(k) = h
.next()
.instrument(span!(Level::INFO, "raw_conn_handler INPUT STREAM READ"))
.await
{
match k {
Ok(k) => {
info!(". . . . . . . . . . . . . . . . . . . . . . . . . . raw_conn_handler FRAME RECV");
frames.push(k);
}
Err(e) => {
return Err((e, netout))?;
}
}
}
if frames.len() != 1 {
error!("expect a command frame");
return Err((Error::with_msg("expect a command frame"), netout))?;
}
let qitem = match bincode::deserialize::<EventQueryJsonStringFrame>(frames[0].buf()) {
Ok(k) => k,
Err(e) => return Err((e, netout))?,
};
trace!("json: {}", qitem.0);
let res: Result<EventsQuery, _> = serde_json::from_str(&qitem.0);
let evq = match res {
Ok(k) => k,
Err(e) => {
error!("can not parse json {:?}", e);
return Err((Error::with_msg("can not parse request json"), netout))?;
}
};
error!(
"TODO decide on response content based on the parsed json query\n{:?}",
evq
);
let query = netpod::AggQuerySingleChannel {
channel_config: netpod::ChannelConfig {
channel: netpod::Channel {
backend: "test1".into(),
name: "wave1".into(),
},
keyspace: 3,
time_bin_size: DAY,
shape: Shape::Wave(17),
scalar_type: ScalarType::F64,
big_endian: true,
array: true,
compression: true,
},
// TODO use a NanoRange and search for matching files
timebin: 0,
tb_file_count: 1,
// TODO use the requested buffer size
buffer_size: 1024 * 4,
};
let mut s1 =
super::eventblobs::EventBlobsComplete::new(&query, query.channel_config.clone(), node_config.node.clone())
.into_dim_1_f32_stream()
.take(10)
.into_binned_x_bins_1();
while let Some(item) = s1.next().await {
if let Ok(k) = &item {
info!("???????????????? emit item ts0: {:?}", k.tss.first());
}
match make_frame::<RawConnOut>(&item) {
Ok(buf) => match netout.write(&buf).await {
Ok(_) => {}
Err(e) => return Err((e, netout))?,
},
Err(e) => {
return Err((e, netout))?;
}
}
}
if false {
// Manual test batch.
let mut batch = MinMaxAvgScalarEventBatch::empty();
batch.tss.push(42);
batch.tss.push(43);
batch.mins.push(7.1);
batch.mins.push(7.2);
batch.maxs.push(8.3);
batch.maxs.push(8.4);
batch.avgs.push(9.5);
batch.avgs.push(9.6);
let mut s1 = futures_util::stream::iter(vec![batch]).map(Result::Ok);
while let Some(item) = s1.next().await {
match make_frame::<RawConnOut>(&item) {
Ok(buf) => match netout.write(&buf).await {
Ok(_) => {}
Err(e) => return Err((e, netout))?,
},
Err(e) => {
return Err((e, netout))?;
}
}
}
}
let buf = make_term_frame();
match netout.write(&buf).await {
Ok(_) => (),
Err(e) => return Err((e, netout))?,
}
match netout.flush().await {
Ok(_) => (),
Err(e) => return Err((e, netout))?,
}
Ok(())
}
pub fn crchex<T>(t: T) -> String
where
T: AsRef<[u8]>,

69
disk/src/raw/bffr.rs Normal file
View File

@@ -0,0 +1,69 @@
use crate::agg::MinMaxAvgScalarEventBatch;
use crate::frame::inmem::InMemoryFrameAsyncReadStream;
use crate::frame::makeframe::FrameType;
use crate::raw::conn::RawConnOut;
use err::Error;
use futures_core::Stream;
use futures_util::pin_mut;
#[allow(unused_imports)]
use netpod::log::*;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::AsyncRead;
pub struct MinMaxAvgScalarEventBatchStreamFromFrames<T>
where
T: AsyncRead + Unpin,
{
inp: InMemoryFrameAsyncReadStream<T>,
}
impl<T> MinMaxAvgScalarEventBatchStreamFromFrames<T>
where
T: AsyncRead + Unpin,
{
pub fn new(inp: InMemoryFrameAsyncReadStream<T>) -> Self {
Self { inp }
}
}
impl<T> Stream for MinMaxAvgScalarEventBatchStreamFromFrames<T>
where
T: AsyncRead + Unpin,
{
type Item = Result<MinMaxAvgScalarEventBatch, Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
loop {
let j = &mut self.inp;
pin_mut!(j);
break match j.poll_next(cx) {
Ready(Some(Ok(frame))) => {
type ExpectedType = RawConnOut;
trace!(
"MinMaxAvgScalarEventBatchStreamFromFrames got full frame buf {}",
frame.buf().len()
);
assert!(frame.tyid() == <ExpectedType as FrameType>::FRAME_TYPE_ID);
match bincode::deserialize::<ExpectedType>(frame.buf()) {
Ok(item) => match item {
Ok(item) => Ready(Some(Ok(item))),
Err(e) => Ready(Some(Err(e))),
},
Err(e) => {
error!(
"MinMaxAvgScalarEventBatchStreamFromFrames ~~~~~~~~ ERROR on frame payload {}",
frame.buf().len(),
);
Ready(Some(Err(e.into())))
}
}
}
Ready(Some(Err(e))) => Ready(Some(Err(e))),
Ready(None) => Ready(None),
Pending => Pending,
};
}
}
}

202
disk/src/raw/conn.rs Normal file
View File

@@ -0,0 +1,202 @@
use crate::agg::{IntoBinnedXBins1, IntoDim1F32Stream, MinMaxAvgScalarEventBatch};
use crate::eventblobs::EventBlobsComplete;
use crate::frame::inmem::InMemoryFrameAsyncReadStream;
use crate::frame::makeframe::{make_frame, make_term_frame};
use crate::raw::{EventQueryJsonStringFrame, EventsQuery};
use err::Error;
use futures_util::StreamExt;
#[allow(unused_imports)]
use netpod::log::*;
use netpod::timeunits::DAY;
use netpod::{NodeConfig, ScalarType, Shape};
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::io::AsyncWriteExt;
use tokio::net::tcp::OwnedWriteHalf;
use tokio::net::TcpStream;
use tracing::Instrument;
pub async fn raw_service(node_config: Arc<NodeConfig>) -> Result<(), Error> {
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)) => {
taskrun::spawn(raw_conn_handler(stream, addr, node_config.clone()));
}
Err(e) => Err(e)?,
}
}
}
async fn raw_conn_handler(stream: TcpStream, addr: SocketAddr, node_config: Arc<NodeConfig>) -> 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)
.instrument(span1)
.await;
match r {
Ok(k) => Ok(k),
Err(e) => {
error!("raw_conn_handler sees error: {:?}", e);
Err(e)
}
}
}
pub type RawConnOut = Result<MinMaxAvgScalarEventBatch, Error>;
async fn raw_conn_handler_inner(
stream: TcpStream,
addr: SocketAddr,
node_config: Arc<NodeConfig>,
) -> Result<(), Error> {
match raw_conn_handler_inner_try(stream, addr, node_config).await {
Ok(_) => (),
Err(mut ce) => {
/*error!(
"raw_conn_handler_inner CAUGHT ERROR AND TRY TO SEND OVER TCP {:?}",
ce.err
);*/
let buf = make_frame::<RawConnOut>(&Err(ce.err))?;
match ce.netout.write(&buf).await {
Ok(_) => (),
Err(e) => return Err(e)?,
}
}
}
Ok(())
}
struct ConnErr {
err: Error,
netout: OwnedWriteHalf,
}
impl<E: Into<Error>> From<(E, OwnedWriteHalf)> for ConnErr {
fn from((err, netout): (E, OwnedWriteHalf)) -> Self {
Self {
err: err.into(),
netout,
}
}
}
async fn raw_conn_handler_inner_try(
stream: TcpStream,
addr: SocketAddr,
node_config: Arc<NodeConfig>,
) -> Result<(), ConnErr> {
info!("raw_conn_handler SPAWNED for {:?}", addr);
let (netin, mut netout) = stream.into_split();
let mut h = InMemoryFrameAsyncReadStream::new(netin);
let mut frames = vec![];
while let Some(k) = h
.next()
.instrument(span!(Level::INFO, "raw_conn_handler INPUT STREAM READ"))
.await
{
match k {
Ok(k) => {
info!(". . . . . . . . . . . . . . . . . . . . . . . . . . raw_conn_handler FRAME RECV");
frames.push(k);
}
Err(e) => {
return Err((e, netout))?;
}
}
}
if frames.len() != 1 {
error!("expect a command frame");
return Err((Error::with_msg("expect a command frame"), netout))?;
}
let qitem = match bincode::deserialize::<EventQueryJsonStringFrame>(frames[0].buf()) {
Ok(k) => k,
Err(e) => return Err((e, netout))?,
};
trace!("json: {}", qitem.0);
let res: Result<EventsQuery, _> = serde_json::from_str(&qitem.0);
let evq = match res {
Ok(k) => k,
Err(e) => {
error!("can not parse json {:?}", e);
return Err((Error::with_msg("can not parse request json"), netout))?;
}
};
error!(
"TODO decide on response content based on the parsed json query\n{:?}",
evq
);
let query = netpod::AggQuerySingleChannel {
channel_config: netpod::ChannelConfig {
channel: netpod::Channel {
backend: "test1".into(),
name: "wave1".into(),
},
keyspace: 3,
time_bin_size: DAY,
shape: Shape::Wave(17),
scalar_type: ScalarType::F64,
big_endian: true,
array: true,
compression: true,
},
// TODO use a NanoRange and search for matching files
timebin: 0,
tb_file_count: 1,
// TODO use the requested buffer size
buffer_size: 1024 * 4,
};
let mut s1 = EventBlobsComplete::new(&query, query.channel_config.clone(), node_config.node.clone())
.into_dim_1_f32_stream()
.take(10)
.into_binned_x_bins_1();
while let Some(item) = s1.next().await {
if let Ok(k) = &item {
trace!("???????????????? emit item ts0: {:?}", k.tss.first());
}
match make_frame::<RawConnOut>(&item) {
Ok(buf) => match netout.write(&buf).await {
Ok(_) => {}
Err(e) => return Err((e, netout))?,
},
Err(e) => {
return Err((e, netout))?;
}
}
}
if false {
// Manual test batch.
let mut batch = MinMaxAvgScalarEventBatch::empty();
batch.tss.push(42);
batch.tss.push(43);
batch.mins.push(7.1);
batch.mins.push(7.2);
batch.maxs.push(8.3);
batch.maxs.push(8.4);
batch.avgs.push(9.5);
batch.avgs.push(9.6);
let mut s1 = futures_util::stream::iter(vec![batch]).map(Result::Ok);
while let Some(item) = s1.next().await {
match make_frame::<RawConnOut>(&item) {
Ok(buf) => match netout.write(&buf).await {
Ok(_) => {}
Err(e) => return Err((e, netout))?,
},
Err(e) => {
return Err((e, netout))?;
}
}
}
}
let buf = make_term_frame();
match netout.write(&buf).await {
Ok(_) => (),
Err(e) => return Err((e, netout))?,
}
match netout.flush().await {
Ok(_) => (),
Err(e) => return Err((e, netout))?,
}
Ok(())
}

5
httpret/src/lib.rs
View File

@@ -1,5 +1,6 @@
use bytes::Bytes;
use disk::cache::PreBinnedQuery;
use disk::raw::conn::raw_service;
use err::Error;
use future::Future;
use futures_core::Stream;
@@ -19,12 +20,12 @@ use tracing::field::Empty;
use tracing::{debug, error, info, span, trace, warn, Level};
pub async fn host(node_config: Arc<NodeConfig>) -> Result<(), Error> {
let rawjh = taskrun::spawn(disk::raw::raw_service(node_config.clone()));
let rawjh = taskrun::spawn(raw_service(node_config.clone()));
use std::str::FromStr;
let addr = SocketAddr::from_str(&format!("{}:{}", node_config.node.listen, node_config.node.port))?;
let make_service = make_service_fn({
move |conn| {
info!("new conn {:?}", conn);
info!("new raw {:?}", conn);
let node_config = node_config.clone();
async move {
Ok::<_, Error>(service_fn({

5
netpod/src/lib.rs
View File

@@ -536,3 +536,8 @@ where
}
}
}
pub mod log {
#[allow(unused_imports)]
pub use tracing::{debug, error, info, span, trace, warn, Level};
}

7
retrieval/src/test.rs
View File

@@ -1,6 +1,7 @@
use crate::spawn_test_hosts;
use bytes::BytesMut;
use chrono::Utc;
use disk::frame::inmem::InMemoryFrameAsyncReadStream;
use err::Error;
use futures_util::TryStreamExt;
use hyper::Body;
@@ -64,7 +65,7 @@ async fn get_cached_0_inner() -> Result<(), Error> {
info!("client response {:?}", res);
//let (res_head, mut res_body) = res.into_parts();
let s1 = disk::cache::HttpBodyAsAsyncRead::new(res);
let s2 = disk::raw::InMemoryFrameAsyncReadStream::new(s1);
let s2 = InMemoryFrameAsyncReadStream::new(s1);
/*use hyper::body::HttpBody;
loop {
match res_body.data().await {
@@ -90,11 +91,11 @@ async fn get_cached_0_inner() -> Result<(), Error> {
let g = match item {
Ok(frame) => {
type ExpectedType = disk::cache::BinnedBytesForHttpStreamFrame;
info!("TEST GOT FRAME len {}", frame.buf().len());
//info!("TEST GOT FRAME len {}", frame.buf().len());
match bincode::deserialize::<ExpectedType>(frame.buf()) {
Ok(item) => match item {
Ok(item) => {
info!("TEST GOT ITEM");
info!("TEST GOT ITEM {:?}", item);
bin_count += 1;
Some(Ok(item))
}

2
taskrun/src/lib.rs
View File

@@ -56,7 +56,7 @@ pub fn tracing_init() {
.with_thread_names(true)
//.with_max_level(tracing::Level::INFO)
.with_env_filter(tracing_subscriber::EnvFilter::new(
"info,retrieval=trace,retrieval::test=trace,disk=info,tokio_postgres=info",
"info,retrieval=trace,retrieval::test=trace,tokio_postgres=info",
))
.init();
}