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

WIP on binning for archeng

Browse Source
This commit is contained in:
Dominik Werder
2021-11-10 17:25:40 +01:00
parent 2f608a8a4e
commit 21e16cfa6d
18 changed files with 504 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
archapp/src/archeng.rs
View File

@@ -10,7 +10,7 @@ pub mod indexfiles;
pub mod indextree;
pub mod pipe;
use self::indexfiles::list_index_files;
use self::indexfiles::{database_connect, list_index_files};
use self::indextree::channel_list;
use crate::timed::Timed;
use crate::wrap_task;
@@ -22,7 +22,9 @@ use items::{StreamItem, WithLen};
use netpod::log::*;
use netpod::timeunits::SEC;
use netpod::{ChannelArchiver, ChannelConfigQuery, ChannelConfigResponse};
use netpod::{ScalarType, Shape};
use serde::Serialize;
use serde_json::Value as JsVal;
use std::convert::TryInto;
const EPICS_EPOCH_OFFSET: u64 = 631152000 * SEC;
@@ -161,6 +163,47 @@ pub fn list_all_channels(node: &ChannelArchiver) -> Receiver<Result<ListChannelI
rx
}
pub async fn channel_config_from_db(
q: &ChannelConfigQuery,
conf: &ChannelArchiver,
) -> Result<ChannelConfigResponse, Error> {
let dbc = database_connect(&conf.database).await?;
let sql = "select config from channels where name = $1";
let rows = dbc.query(sql, &[&q.channel.name()]).await?;
if let Some(row) = rows.first() {
let cfg: JsVal = row.try_get(0)?;
let val = cfg
.get("shape")
.ok_or_else(|| Error::with_msg_no_trace("shape not found on config"))?;
let shape = Shape::from_db_jsval(val)?;
let val = cfg
.get("scalarType")
.ok_or_else(|| Error::with_msg_no_trace("no scalarType in db"))?;
let s = if let JsVal::String(s) = val {
s
} else {
return Err(Error::with_msg_no_trace(format!(
"channel_config_from_db bad scalar type {:?}",
cfg
)));
};
let scalar_type = ScalarType::from_archeng_db_str(s)?;
let ret = ChannelConfigResponse {
channel: q.channel.clone(),
scalar_type,
// TODO.. only binary endpoint would care.
byte_order: None,
shape,
};
Ok(ret)
} else {
Err(Error::with_msg_no_trace(format!(
"can not find config for {}",
q.channel.name()
)))
}
}
pub async fn channel_config(q: &ChannelConfigQuery, conf: &ChannelArchiver) -> Result<ChannelConfigResponse, Error> {
let _timed = Timed::new("channel_config");
let mut type_info = None;

14
archapp/src/archeng/blockrefstream.rs
View File

@@ -89,14 +89,20 @@ impl BlockrefStream {
}
SelectIndexFile => {
let dbc = database_connect(&self.conf.database).await?;
let sql = "select path from indexfiles i, channels c, channel_index_map m where c.name = $1 and m.channel = c.rowid and i.rowid = m.index";
let sql = "select i.path from indexfiles i, channels c, channel_index_map m where c.name = $1 and m.channel = c.rowid and i.rowid = m.index";
let rows = dbc.query(sql, &[&self.channel.name()]).await?;
for row in rows {
self.paths.push_back(row.try_get(0)?);
let p: String = row.try_get(0)?;
if self.paths.is_empty() && (p.contains("_ST/") || p.contains("_SH/")) {
self.paths.push_back(p);
}
}
if self.paths.len() == 0 {
self.steps = Done;
Ok(Some((BlockrefItem::JsVal(JsVal::String(format!("NOMOREPATHS"))), self)))
Ok(Some((
BlockrefItem::JsVal(JsVal::String(format!("NOPATHSFROMDB"))),
self,
)))
} else {
self.steps = SetupNextPath;
Ok(Some((BlockrefItem::JsVal(JsVal::String(format!("DBQUERY"))), self)))
@@ -122,7 +128,7 @@ impl BlockrefStream {
};
Ok(Some((BlockrefItem::JsVal(JsVal::String(format!("NEXTPATH"))), self)))
} else {
self.steps = SelectIndexFile;
self.steps = Done;
Ok(Some((
BlockrefItem::JsVal(JsVal::String(format!("PATHQUEUEEMPTY"))),
self,

87
archapp/src/archeng/blockstream.rs
View File

@@ -9,7 +9,7 @@ use futures_util::stream::FuturesOrdered;
use futures_util::StreamExt;
use items::eventsitem::EventsItem;
use items::{WithLen, WithTimestamps};
use netpod::{log::*, NanoRange};
use netpod::{log::*, NanoRange, Nanos};
use serde::Serialize;
use serde_json::Value as JsVal;
use std::collections::{BTreeMap, VecDeque};
@@ -58,9 +58,9 @@ struct Reader {
impl Reader {}
struct FutAItem {
#[allow(unused)]
fname: String,
path: PathBuf,
dpos: DataheaderPos,
dfnotfound: bool,
reader: Option<Reader>,
bytes_read: u64,
@@ -101,6 +101,8 @@ pub struct BlockStream<S> {
last_dfname: String,
last_dfhpos: DataheaderPos,
ts_max: u64,
data_done: bool,
raco: bool,
done: bool,
complete: bool,
acc: StatsAcc,
@@ -115,6 +117,7 @@ impl<S> BlockStream<S> {
where
S: Stream<Item = Result<BlockrefItem, Error>> + Unpin,
{
debug!("new BlockStream");
Self {
inp,
inp_done: false,
@@ -126,6 +129,8 @@ impl<S> BlockStream<S> {
last_dfname: String::new(),
last_dfhpos: DataheaderPos(u64::MAX),
ts_max: 0,
data_done: false,
raco: false,
done: false,
complete: false,
acc: StatsAcc::new(),
@@ -159,6 +164,14 @@ where
} else if self.done {
self.complete = true;
Ready(None)
} else if self.data_done {
self.done = true;
if self.raco {
// currently handled downstream
continue;
} else {
continue;
}
} else {
let item1 = if self.inp_done {
Int::Done
@@ -201,7 +214,7 @@ where
Some(reader)
} else {
let stats = StatsChannel::dummy();
debug!("open new reader file {:?}", dpath);
trace!("open new reader file {:?}", dpath);
match open_read(dpath.clone(), &stats).await {
Ok(file) => {
//
@@ -217,7 +230,8 @@ where
if let Some(mut reader) = reader {
let rp1 = reader.rb.bytes_read();
let dfheader =
read_datafile_header2(&mut reader.rb, pos).await?;
read_datafile_header2(&mut reader.rb, pos.clone())
.await?;
// TODO handle expand
let expand = false;
let data =
@@ -234,6 +248,7 @@ where
let ret = FutAItem {
fname,
path: dpath,
dpos: pos,
dfnotfound: false,
reader: Some(reader),
bytes_read,
@@ -245,6 +260,7 @@ where
let ret = FutAItem {
fname,
path: dpath,
dpos: pos,
dfnotfound: true,
reader: None,
bytes_read: 0,
@@ -285,23 +301,62 @@ where
} else {
match self.block_reads.poll_next_unpin(cx) {
Ready(Some(Ok(item))) => {
let mut item = item;
item.events = if let Some(ev) = item.events {
if ev.len() > 0 {
if ev.ts(ev.len() - 1) > self.range.end {
debug!(". . . . ===== DATA DONE ----------------------");
self.raco = true;
self.data_done = true;
}
}
if ev.len() == 1 {
debug!("From {} {:?} {}", item.fname, item.path, item.dpos.0);
debug!("See 1 event {:?}", Nanos::from_ns(ev.ts(0)));
} else if ev.len() > 1 {
debug!("From {} {:?} {}", item.fname, item.path, item.dpos.0);
debug!(
"See {} events {:?} to {:?}",
ev.len(),
Nanos::from_ns(ev.ts(0)),
Nanos::from_ns(ev.ts(ev.len() - 1))
);
}
let mut contains_unordered = false;
for i in 0..ev.len() {
let ts = ev.ts(i);
debug!("\nSEE EVENT {:?}", Nanos::from_ns(ts));
if ts < self.ts_max {
contains_unordered = true;
if true {
let msg = format!(
"unordered event in item at {} ts {:?} ts_max {:?}",
i,
Nanos::from_ns(ts),
Nanos::from_ns(self.ts_max)
);
error!("{}", msg);
self.done = true;
return Ready(Some(Err(Error::with_msg_no_trace(msg))));
}
}
self.ts_max = ts;
}
if contains_unordered {
Some(ev)
} else {
Some(ev)
}
} else {
None
};
let item = item;
if item.dfnotfound {
self.dfnotfound.insert(item.path, true);
self.dfnotfound.insert(item.path.clone(), true);
}
if let Some(reader) = item.reader {
self.readers.push_back(reader);
}
if let Some(ev) = &item.events {
for i in 0..ev.len() {
let ts = ev.ts(i);
if ts < self.ts_max {
let msg = format!("unordered event: {} {}", ts, self.ts_max);
error!("{}", msg);
self.done = true;
return Ready(Some(Err(Error::with_msg_no_trace(msg))));
}
}
}
self.acc.add(item.events_read, item.bytes_read);
if false {
let item = JsVal::String(format!(

89
archapp/src/archeng/pipe.rs
View File

@@ -1,17 +1,92 @@
use crate::archeng::blockrefstream::blockref_stream;
use crate::archeng::blockstream::BlockStream;
use crate::archeng::datablockstream::DatablockStream;
use crate::events::{FrameMaker, FrameMakerTrait};
use err::Error;
use futures_core::Stream;
use futures_util::StreamExt;
use items::Framable;
use netpod::ChannelConfigQuery;
use netpod::{query::RawEventsQuery, ChannelArchiver};
use futures_util::{Stream, StreamExt};
use items::binnedevents::XBinnedEvents;
use items::eventsitem::EventsItem;
use items::plainevents::PlainEvents;
use items::{Framable, LogItem, RangeCompletableItem, StreamItem};
use netpod::query::RawEventsQuery;
use netpod::{log::*, AggKind, Shape};
use netpod::{ChannelArchiver, ChannelConfigQuery};
use std::pin::Pin;
use streams::rangefilter::RangeFilter;
pub async fn make_event_pipe(
evq: &RawEventsQuery,
conf: &ChannelArchiver,
conf: ChannelArchiver,
) -> Result<Pin<Box<dyn Stream<Item = Box<dyn Framable>> + Send>>, Error> {
debug!("make_event_pipe {:?}", evq);
let channel_config = {
let q = ChannelConfigQuery {
channel: evq.channel.clone(),
range: evq.range.clone(),
};
crate::archeng::channel_config_from_db(&q, &conf).await?
};
debug!("Channel config: {:?}", channel_config);
use crate::archeng::blockstream::BlockItem;
let refs = blockref_stream(evq.channel.clone(), evq.range.clone(), conf.clone());
let blocks = BlockStream::new(Box::pin(refs), evq.range.clone(), 1);
let blocks = blocks.map(|k| match k {
Ok(item) => match item {
BlockItem::EventsItem(item) => Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))),
BlockItem::JsVal(jsval) => Ok(StreamItem::Log(LogItem::quick(Level::DEBUG, format!("{:?}", jsval)))),
},
Err(e) => Err(e),
});
let filtered = RangeFilter::new(blocks, evq.range.clone(), evq.agg_kind.need_expand());
let xtrans = match channel_config.shape {
Shape::Scalar => match evq.agg_kind {
AggKind::Plain => Box::pin(filtered) as Pin<Box<dyn Stream<Item = _> + Send>>,
AggKind::TimeWeightedScalar | AggKind::DimXBins1 => {
let tr = filtered.map(|j| match j {
Ok(j) => match j {
StreamItem::DataItem(j) => match j {
RangeCompletableItem::RangeComplete => {
Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))
}
RangeCompletableItem::Data(j) => match j {
EventsItem::Plain(j) => match j {
PlainEvents::Scalar(j) => {
let item = XBinnedEvents::Scalar(j);
let item = EventsItem::XBinnedEvents(item);
Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))
}
PlainEvents::Wave(_) => panic!(),
},
EventsItem::XBinnedEvents(_) => panic!(),
},
},
StreamItem::Log(j) => Ok(StreamItem::Log(j)),
StreamItem::Stats(j) => Ok(StreamItem::Stats(j)),
},
Err(e) => Err(e),
});
Box::pin(tr) as _
}
AggKind::DimXBinsN(_) => err::todoval(),
AggKind::EventBlobs => err::todoval(),
},
_ => {
error!("TODO shape {:?}", channel_config.shape);
panic!()
}
};
let mut frame_maker = Box::new(FrameMaker::with_item_type(
channel_config.scalar_type.clone(),
channel_config.shape.clone(),
evq.agg_kind.clone(),
)) as Box<dyn FrameMakerTrait>;
let ret = xtrans.map(move |j| frame_maker.make_frame(j));
Ok(Box::pin(ret))
}
pub async fn make_event_pipe1(
evq: &RawEventsQuery,
conf: ChannelArchiver,
) -> Result<Pin<Box<dyn Stream<Item = Box<dyn Framable>> + Send>>, Error> {
let range = evq.range.clone();
let channel = evq.channel.clone();
@@ -25,7 +100,7 @@ pub async fn make_event_pipe(
channel: channel.clone(),
range: range.clone(),
};
crate::archeng::channel_config(&q, conf).await?
crate::archeng::channel_config_from_db(&q, &conf).await?
};
let data = DatablockStream::for_channel_range(

30
archapp/src/events.rs
View File

@@ -138,7 +138,10 @@ macro_rules! arm2 {
_ => panic!(),
},
_ => err::todoval(),
_ => {
error!("unexpected arm2 case");
err::todoval()
}
},
},
StreamItem::Log(k) => Ok(StreamItem::Log(k)),
@@ -223,6 +226,7 @@ macro_rules! arm1 {
},
Shape::Image(..) => {
// There should be no images on archiver.
warn!("TODO for {:?}", $shape);
err::todoval()
}
}
@@ -240,7 +244,10 @@ impl FrameMakerTrait for FrameMaker {
ScalarType::I32 => arm1!(item, i32, Int, shape, agg_kind),
ScalarType::F32 => arm1!(item, f32, Float, shape, agg_kind),
ScalarType::F64 => arm1!(item, f64, Double, shape, agg_kind),
_ => err::todoval(),
_ => {
warn!("TODO for scalar_type {:?}", scalar_type);
err::todoval()
}
}
}
}
@@ -273,14 +280,13 @@ pub async fn make_single_event_pipe(
base_path: PathBuf,
) -> Result<Pin<Box<dyn Stream<Item = Sitemty<EventsItem>> + Send>>, Error> {
// TODO must apply the proper x-binning depending on the requested AggKind.
info!("make_event_pipe {:?}", evq);
debug!("make_single_event_pipe {:?}", evq);
let evq = evq.clone();
let DirAndPrefix { dir, prefix } = directory_for_channel_files(&evq.channel, base_path)?;
//let dtbeg = Utc.timestamp((evq.range.beg / 1000000000) as i64, (evq.range.beg % 1000000000) as u32);
let (tx, rx) = async_channel::bounded(16);
let block1 = async move {
trace!("++++++++++++++++++++++++++++");
info!("start read of {:?}", dir);
debug!("start read of {:?}", dir);
// TODO first collect all matching filenames, then sort, then open files.
// TODO if dir does not exist, should notify client but not log as error.
@@ -299,13 +305,13 @@ pub async fn make_single_event_pipe(
if s.starts_with(&prefix) && s.ends_with(".pb") {
match parse_data_filename(&s) {
Ok(df) => {
info!("parse went ok: {} {}", df.year, df.month);
debug!("parse went ok: {} {}", df.year, df.month);
let ts0 = Utc.ymd(df.year as i32, df.month, 1).and_hms(0, 0, 0);
let ts1 = ts0.timestamp() as u64 * SEC + ts0.timestamp_subsec_nanos() as u64;
info!("file {} {}", ts1, ts1 + DAY * 27);
info!("range {} {}", evq.range.beg, evq.range.end);
debug!("file {} {}", ts1, ts1 + DAY * 27);
debug!("range {} {}", evq.range.beg, evq.range.end);
if evq.range.beg < ts1 + DAY * 27 && evq.range.end > ts1 {
info!("•••••••••••••••••••••••••• file matches requested range");
debug!("•••••••••••••••••••••••••• file matches requested range");
let f1 = File::open(de.path()).await?;
info!("opened {:?}", de.path());
@@ -322,7 +328,7 @@ pub async fn make_single_event_pipe(
f1.seek(SeekFrom::Start(0)).await?;
let mut pbr = PbFileReader::new(f1).await;
pbr.read_header().await?;
info!("✓ read header {:?}", pbr.payload_type());
debug!("✓ read header {:?}", pbr.payload_type());
pbr.file().seek(SeekFrom::Start(pos1)).await?;
pbr.reset_io(pos1);
@@ -347,7 +353,7 @@ pub async fn make_single_event_pipe(
}
}
Ok(None) => {
info!("reached end of file");
debug!("reached end of file");
break;
}
Err(e) => {
@@ -363,7 +369,7 @@ pub async fn make_single_event_pipe(
}
}
} else {
info!("prefix {} s {}", prefix, s);
debug!("prefix {} s {}", prefix, s);
}
}
Ok::<_, Error>(())