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

Run on cryo

Browse Source
This commit is contained in:
Dominik Werder
2022-12-20 16:36:42 +01:00
parent 646ec38b3c
commit 3ded7c6136
6 changed files with 185 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
dbconn/src/search.rs
View File

@@ -100,7 +100,7 @@ pub async fn search_channel_scylla(
" series, facility, channel, scalar_type, shape_dims",
" from series_by_channel",
" where channel ~* $1",
" limit 100",
" limit 400000",
));
let pgclient = crate::create_connection(pgconf).await?;
let rows = pgclient.query(sql.as_str(), &[&query.name_regex]).await.err_conv()?;

3
httpret/src/api4/binned.rs
View File

@@ -15,7 +15,7 @@ use tracing::Instrument;
use url::Url;
async fn binned_json(url: Url, req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
info!("httpret plain_events_json req: {:?}", req);
debug!("httpret plain_events_json req: {:?}", req);
let (_head, _body) = req.into_parts();
let query = BinnedQuery::from_url(&url).map_err(|e| {
let msg = format!("can not parse query: {}", e.msg());
@@ -46,7 +46,6 @@ async fn binned_json(url: Url, req: Request<Body>, node_config: &NodeConfigCache
}
async fn binned(req: Request<Body>, node_config: &NodeConfigCached) -> Result<Response<Body>, Error> {
info!("req: {:?}", req);
let accept = req
.headers()
.get(http::header::ACCEPT)

2
nodenet/src/conn.rs
View File

@@ -252,7 +252,7 @@ async fn events_conn_handler_inner_try(
let item = item.make_frame();
match item {
Ok(buf) => {
info!("write {} bytes", buf.len());
trace!("write {} bytes", buf.len());
buf_len_histo.ingest(buf.len() as u32);
match netout.write_all(&buf).await {
Ok(_) => {}

342
scyllaconn/src/events.rs
View File

@@ -1,23 +1,44 @@
use crate::errconv::ErrConv;
use err::Error;
use futures_util::{Future, FutureExt, Stream, StreamExt};
use items_0::{Empty, Events, WithLen};
use items_2::channelevents::{ChannelEvents, ConnStatus, ConnStatusEvent};
use futures_util::Future;
use futures_util::FutureExt;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::Empty;
use items_0::Events;
use items_0::WithLen;
use items_2::channelevents::ChannelEvents;
use items_2::channelevents::ConnStatus;
use items_2::channelevents::ConnStatusEvent;
use items_2::eventsdim0::EventsDim0;
use netpod::log::*;
use netpod::query::ChannelStateEventsQuery;
use netpod::timeunits::*;
use netpod::{NanoRange, ScalarType, Shape};
use netpod::NanoRange;
use netpod::ScalarType;
use netpod::Shape;
use scylla::Session as ScySession;
use std::collections::VecDeque;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::task::Context;
use std::task::Poll;
async fn find_ts_msp(series: u64, range: NanoRange, scy: Arc<ScySession>) -> Result<VecDeque<u64>, Error> {
info!("find_ts_msp series {} {:?}", series, range);
let mut ret = VecDeque::new();
async fn find_ts_msp(
series: u64,
range: NanoRange,
scy: Arc<ScySession>,
) -> Result<(VecDeque<u64>, VecDeque<u64>), Error> {
trace!("find_ts_msp series {} {:?}", series, range);
let mut ret1 = VecDeque::new();
let mut ret2 = VecDeque::new();
// TODO use prepared statements
let cql = "select ts_msp from ts_msp where series = ? and ts_msp <= ? order by ts_msp desc limit 2";
let res = scy.query(cql, (series as i64, range.beg as i64)).await.err_conv()?;
for row in res.rows_typed_or_empty::<(i64,)>() {
let row = row.err_conv()?;
ret1.push_front(row.0 as u64);
}
let cql = "select ts_msp from ts_msp where series = ? and ts_msp > ? and ts_msp < ?";
let res = scy
.query(cql, (series as i64, range.beg as i64, range.end as i64))
@@ -25,22 +46,16 @@ async fn find_ts_msp(series: u64, range: NanoRange, scy: Arc<ScySession>) -> Res
.err_conv()?;
for row in res.rows_typed_or_empty::<(i64,)>() {
let row = row.err_conv()?;
ret.push_back(row.0 as u64);
ret2.push_back(row.0 as u64);
}
let cql = "select ts_msp from ts_msp where series = ? and ts_msp >= ? limit 1";
let res = scy.query(cql, (series as i64, range.end as i64)).await.err_conv()?;
for row in res.rows_typed_or_empty::<(i64,)>() {
let row = row.err_conv()?;
ret.push_back(row.0 as u64);
ret2.push_back(row.0 as u64);
}
let cql = "select ts_msp from ts_msp where series = ? and ts_msp <= ? order by ts_msp desc limit 2";
let res = scy.query(cql, (series as i64, range.beg as i64)).await.err_conv()?;
for row in res.rows_typed_or_empty::<(i64,)>() {
let row = row.err_conv()?;
ret.push_front(row.0 as u64);
}
trace!("found in total {} rows", ret.len());
Ok(ret)
trace!("find_ts_msp n1 {} n2 {}", ret1.len(), ret2.len());
Ok((ret1, ret2))
}
macro_rules! read_next_scalar_values {
@@ -50,7 +65,6 @@ macro_rules! read_next_scalar_values {
ts_msp: u64,
range: NanoRange,
fwd: bool,
do_one_before: bool,
scy: Arc<ScySession>,
) -> Result<EventsDim0<$st>, Error> {
type ST = $st;
@@ -64,7 +78,7 @@ macro_rules! read_next_scalar_values {
if range.end > i64::MAX as u64 {
return Err(Error::with_msg_no_trace(format!("range.end overflows i64")));
}
let res = if fwd {
let ret = if fwd {
let ts_lsp_min = if ts_msp < range.beg { range.beg - ts_msp } else { 0 };
let ts_lsp_max = if ts_msp < range.end { range.end - ts_msp } else { 0 };
trace!(
@@ -74,7 +88,7 @@ macro_rules! read_next_scalar_values {
ts_lsp_max,
range.beg,
range.end,
stringify!($fname)
stringify!($table_name)
);
// TODO use prepared!
let cql = concat!(
@@ -82,18 +96,37 @@ macro_rules! read_next_scalar_values {
$table_name,
" where series = ? and ts_msp = ? and ts_lsp >= ? and ts_lsp < ?"
);
scy.query(cql, (series, ts_msp as i64, ts_lsp_min as i64, ts_lsp_max as i64))
let res = scy
.query(cql, (series, ts_msp as i64, ts_lsp_min as i64, ts_lsp_max as i64))
.await
.err_conv()?
.err_conv()?;
let mut last_before = None;
let mut ret = EventsDim0::<ST>::empty();
for row in res.rows_typed_or_empty::<(i64, i64, SCYTY)>() {
let row = row.err_conv()?;
let ts = ts_msp + row.0 as u64;
let pulse = row.1 as u64;
let value = row.2 as ST;
if ts >= range.end {
} else if ts >= range.beg {
ret.push(ts, pulse, value);
} else {
if last_before.is_none() {
warn!("encounter event before range in forward read {ts}");
}
last_before = Some((ts, pulse, value));
}
}
ret
} else {
let ts_lsp_max = if ts_msp < range.beg { range.beg - ts_msp } else { 0 };
info!(
trace!(
"BCK ts_msp {} ts_lsp_max {} beg {} end {} {}",
ts_msp,
ts_lsp_max,
range.beg,
range.end,
stringify!($fname)
stringify!($table_name)
);
// TODO use prepared!
let cql = concat!(
@@ -101,30 +134,29 @@ macro_rules! read_next_scalar_values {
$table_name,
" where series = ? and ts_msp = ? and ts_lsp < ? order by ts_lsp desc limit 1"
);
scy.query(cql, (series, ts_msp as i64, ts_lsp_max as i64))
let res = scy
.query(cql, (series, ts_msp as i64, ts_lsp_max as i64))
.await
.err_conv()?
.err_conv()?;
let mut seen_before = false;
let mut ret = EventsDim0::<ST>::empty();
for row in res.rows_typed_or_empty::<(i64, i64, SCYTY)>() {
let row = row.err_conv()?;
let ts = ts_msp + row.0 as u64;
let pulse = row.1 as u64;
let value = row.2 as ST;
if ts >= range.end {
} else if ts < range.beg {
ret.push(ts, pulse, value);
} else {
if !seen_before {
warn!("encounter event before range in forward read {ts}");
}
seen_before = true;
}
}
ret
};
let mut last_before = None;
let mut ret = EventsDim0::<ST>::empty();
for row in res.rows_typed_or_empty::<(i64, i64, SCYTY)>() {
let row = row.err_conv()?;
let ts = ts_msp + row.0 as u64;
let pulse = row.1 as u64;
let value = row.2 as ST;
if ts >= range.end {
} else if ts >= range.beg {
ret.push(ts, pulse, value);
} else {
last_before = Some((ts, pulse, value));
}
}
if do_one_before {
if let Some((ts, pulse, value)) = last_before {
info!("PREPENDING THE LAST BEFORE {ts} {value:?}");
ret.push_front(ts, pulse, value);
}
}
trace!("found in total {} events ts_msp {}", ret.tss.len(), ts_msp);
Ok(ret)
}
@@ -138,7 +170,6 @@ macro_rules! read_next_array_values {
ts_msp: u64,
_range: NanoRange,
_fwd: bool,
_do_one_before: bool,
scy: Arc<ScySession>,
) -> Result<EventsDim0<$st>, Error> {
// TODO change return type: so far EventsDim1 does not exist.
@@ -149,7 +180,7 @@ macro_rules! read_next_array_values {
}
type ST = $st;
type _SCYTY = $scyty;
info!("{} series {} ts_msp {}", stringify!($fname), series, ts_msp);
trace!("{} series {} ts_msp {}", stringify!($fname), series, ts_msp);
let cql = concat!(
"select ts_lsp, pulse, value from ",
$table_name,
@@ -172,9 +203,14 @@ macro_rules! read_next_array_values {
};
}
read_next_scalar_values!(read_next_values_scalar_u8, u8, i8, "events_scalar_u8");
read_next_scalar_values!(read_next_values_scalar_u16, u16, i16, "events_scalar_u16");
read_next_scalar_values!(read_next_values_scalar_u32, u32, i32, "events_scalar_u32");
read_next_scalar_values!(read_next_values_scalar_u64, u64, i64, "events_scalar_u64");
read_next_scalar_values!(read_next_values_scalar_i8, i8, i8, "events_scalar_i8");
read_next_scalar_values!(read_next_values_scalar_i16, i16, i16, "events_scalar_i16");
read_next_scalar_values!(read_next_values_scalar_i32, i32, i32, "events_scalar_i32");
read_next_scalar_values!(read_next_values_scalar_i64, i64, i64, "events_scalar_i64");
read_next_scalar_values!(read_next_values_scalar_f32, f32, f32, "events_scalar_f32");
read_next_scalar_values!(read_next_values_scalar_f64, f64, f64, "events_scalar_f64");
@@ -182,14 +218,7 @@ read_next_array_values!(read_next_values_array_u16, u16, i16, "events_wave_u16")
macro_rules! read_values {
($fname:ident, $self:expr, $ts_msp:expr) => {{
let fut = $fname(
$self.series,
$ts_msp,
$self.range.clone(),
$self.fwd,
$self.do_one_before_range,
$self.scy.clone(),
);
let fut = $fname($self.series, $ts_msp, $self.range.clone(), $self.fwd, $self.scy.clone());
let fut = fut.map(|x| match x {
Ok(k) => {
let self_name = std::any::type_name::<Self>();
@@ -211,7 +240,6 @@ struct ReadValues {
range: NanoRange,
ts_msps: VecDeque<u64>,
fwd: bool,
do_one_before_range: bool,
fut: Pin<Box<dyn Future<Output = Result<Box<dyn Events>, Error>> + Send>>,
scy: Arc<ScySession>,
}
@@ -224,7 +252,6 @@ impl ReadValues {
range: NanoRange,
ts_msps: VecDeque<u64>,
fwd: bool,
do_one_before_range: bool,
scy: Arc<ScySession>,
) -> Self {
let mut ret = Self {
@@ -234,7 +261,6 @@ impl ReadValues {
range,
ts_msps,
fwd,
do_one_before_range,
fut: Box::pin(futures_util::future::ready(Err(Error::with_msg_no_trace(
"future not initialized",
)))),
@@ -246,20 +272,28 @@ impl ReadValues {
fn next(&mut self) -> bool {
if let Some(ts_msp) = self.ts_msps.pop_front() {
self.fut = self.make_fut(ts_msp, self.ts_msps.len() > 0);
self.fut = self.make_fut(ts_msp);
true
} else {
false
}
}
fn make_fut(
&mut self,
ts_msp: u64,
_has_more_msp: bool,
) -> Pin<Box<dyn Future<Output = Result<Box<dyn Events>, Error>> + Send>> {
fn make_fut(&mut self, ts_msp: u64) -> Pin<Box<dyn Future<Output = Result<Box<dyn Events>, Error>> + Send>> {
let fut = match &self.shape {
Shape::Scalar => match &self.scalar_type {
ScalarType::U8 => {
read_values!(read_next_values_scalar_u8, self, ts_msp)
}
ScalarType::U16 => {
read_values!(read_next_values_scalar_u16, self, ts_msp)
}
ScalarType::U32 => {
read_values!(read_next_values_scalar_u32, self, ts_msp)
}
ScalarType::U64 => {
read_values!(read_next_values_scalar_u64, self, ts_msp)
}
ScalarType::I8 => {
read_values!(read_next_values_scalar_i8, self, ts_msp)
}
@@ -269,6 +303,9 @@ impl ReadValues {
ScalarType::I32 => {
read_values!(read_next_values_scalar_i32, self, ts_msp)
}
ScalarType::I64 => {
read_values!(read_next_values_scalar_i64, self, ts_msp)
}
ScalarType::F32 => {
read_values!(read_next_values_scalar_f32, self, ts_msp)
}
@@ -300,7 +337,7 @@ impl ReadValues {
enum FrState {
New,
FindMsp(Pin<Box<dyn Future<Output = Result<VecDeque<u64>, Error>> + Send>>),
FindMsp(Pin<Box<dyn Future<Output = Result<(VecDeque<u64>, VecDeque<u64>), Error>> + Send>>),
ReadBack1(ReadValues),
ReadBack2(ReadValues),
ReadValues(ReadValues),
@@ -313,10 +350,14 @@ pub struct EventsStreamScylla {
scalar_type: ScalarType,
shape: Shape,
range: NanoRange,
#[allow(unused)]
do_one_before_range: bool,
ts_msp_b1: Option<u64>,
ts_msp_b2: Option<u64>,
ts_msps: VecDeque<u64>,
scy: Arc<ScySession>,
do_test_stream_error: bool,
outqueue: VecDeque<Box<dyn Events>>,
}
impl EventsStreamScylla {
@@ -329,8 +370,6 @@ impl EventsStreamScylla {
scy: Arc<ScySession>,
do_test_stream_error: bool,
) -> Self {
let self_name = std::any::type_name::<Self>();
info!("{self_name} do_one_before_range {do_one_before_range}");
Self {
state: FrState::New,
series,
@@ -338,37 +377,41 @@ impl EventsStreamScylla {
shape,
range,
do_one_before_range,
ts_msp_b1: None,
ts_msp_b2: None,
ts_msps: VecDeque::new(),
scy,
do_test_stream_error,
outqueue: VecDeque::new(),
}
}
fn ts_msps_found(&mut self, ts_msps: VecDeque<u64>) {
info!("ts_msps_found ts_msps {ts_msps:?}");
self.ts_msps = ts_msps;
// Find the largest MSP which can potentially contain some event before the range.
let befores: Vec<_> = self
.ts_msps
.iter()
.map(|x| *x)
.filter(|x| *x < self.range.beg)
.collect();
if self.do_one_before_range && befores.len() >= 1 {
info!("Try ReadBack1");
fn ts_msps_found(&mut self, msps1: VecDeque<u64>, msps2: VecDeque<u64>) {
trace!("ts_msps_found msps1 {msps1:?} msps2 {msps2:?}");
let mut msps1 = msps1;
self.ts_msp_b1 = msps1.pop_back();
self.ts_msp_b2 = msps1.pop_back();
self.ts_msps = msps2;
if let Some(x) = self.ts_msp_b1.clone() {
self.ts_msps.push_front(x);
}
trace!("ts_msp_b1 {:?}", self.ts_msp_b1);
trace!("ts_msp_b2 {:?}", self.ts_msp_b2);
trace!("ts_msps {:?}", self.ts_msps);
if let Some(msp) = self.ts_msp_b1.clone() {
trace!("Try ReadBack1");
let st = ReadValues::new(
self.series as i64,
self.scalar_type.clone(),
self.shape.clone(),
self.range.clone(),
[befores[befores.len() - 1]].into(),
[msp].into(),
false,
self.do_one_before_range,
self.scy.clone(),
);
self.state = FrState::ReadBack1(st);
} else if self.ts_msps.len() >= 1 {
info!("Go straight for forward read");
trace!("Go straight for forward read");
let st = ReadValues::new(
self.series as i64,
self.scalar_type.clone(),
@@ -376,7 +419,6 @@ impl EventsStreamScylla {
self.range.clone(),
self.ts_msps.clone(),
true,
self.do_one_before_range,
self.scy.clone(),
);
self.state = FrState::ReadValues(st);
@@ -385,51 +427,10 @@ impl EventsStreamScylla {
}
}
fn back_1_done(&mut self, item: Box<dyn Events>) -> Option<Box<dyn Events>> {
info!("back_1_done len {}", item.len());
if item.len() == 0 {
info!("ReadBack1 returned empty");
// Find the 2nd largest MSP which can potentially contain some event before the range.
let befores: Vec<_> = self
.ts_msps
.iter()
.map(|x| *x)
.filter(|x| *x < self.range.beg)
.collect();
if befores.len() >= 2 {
info!("Try ReadBack2");
let st = ReadValues::new(
self.series as i64,
self.scalar_type.clone(),
self.shape.clone(),
self.range.clone(),
[befores[befores.len() - 2]].into(),
false,
self.do_one_before_range,
self.scy.clone(),
);
self.state = FrState::ReadBack2(st);
None
} else if self.ts_msps.len() >= 1 {
info!("No 2nd back MSP, go for forward read");
let st = ReadValues::new(
self.series as i64,
self.scalar_type.clone(),
self.shape.clone(),
self.range.clone(),
self.ts_msps.clone(),
true,
self.do_one_before_range,
self.scy.clone(),
);
self.state = FrState::ReadValues(st);
None
} else {
self.state = FrState::Done;
None
}
} else {
info!("FOUND ONE BEFORE");
fn back_1_done(&mut self, item: Box<dyn Events>) {
trace!("back_1_done item len {}", item.len());
if item.len() > 0 {
self.outqueue.push_back(item);
if self.ts_msps.len() > 0 {
let st = ReadValues::new(
self.series as i64,
@@ -438,22 +439,47 @@ impl EventsStreamScylla {
self.range.clone(),
self.ts_msps.clone(),
true,
self.do_one_before_range,
self.scy.clone(),
);
self.state = FrState::ReadValues(st);
Some(item)
} else {
self.state = FrState::Done;
Some(item)
}
} else {
if let Some(msp) = self.ts_msp_b2.clone() {
trace!("Try ReadBack2");
let st = ReadValues::new(
self.series as i64,
self.scalar_type.clone(),
self.shape.clone(),
self.range.clone(),
[msp].into(),
false,
self.scy.clone(),
);
self.state = FrState::ReadBack2(st);
} else if self.ts_msps.len() >= 1 {
trace!("No 2nd back MSP, go for forward read");
let st = ReadValues::new(
self.series as i64,
self.scalar_type.clone(),
self.shape.clone(),
self.range.clone(),
self.ts_msps.clone(),
true,
self.scy.clone(),
);
self.state = FrState::ReadValues(st);
} else {
self.state = FrState::Done;
}
}
}
fn back_2_done(&mut self, item: Box<dyn Events>) -> Option<Box<dyn Events>> {
info!("back_2_done len {}", item.len());
if item.len() == 0 {
info!("ReadBack2 returned empty");
fn back_2_done(&mut self, item: Box<dyn Events>) {
trace!("back_1_done item len {}", item.len());
if item.len() > 0 {
self.outqueue.push_back(item);
}
if self.ts_msps.len() >= 1 {
let st = ReadValues::new(
@@ -463,18 +489,12 @@ impl EventsStreamScylla {
self.range.clone(),
self.ts_msps.clone(),
true,
self.do_one_before_range,
self.scy.clone(),
);
self.state = FrState::ReadValues(st);
} else {
self.state = FrState::Done;
}
if item.len() > 0 {
Some(item)
} else {
None
}
}
}
@@ -489,6 +509,11 @@ impl Stream for EventsStreamScylla {
return Ready(Some(Err(e)));
}
loop {
if let Some(item) = self.outqueue.pop_front() {
item.verify();
item.output_info();
break Ready(Some(Ok(ChannelEvents::Events(item))));
}
break match self.state {
FrState::New => {
let fut = find_ts_msp(self.series, self.range.clone(), self.scy.clone());
@@ -497,8 +522,8 @@ impl Stream for EventsStreamScylla {
continue;
}
FrState::FindMsp(ref mut fut) => match fut.poll_unpin(cx) {
Ready(Ok(ts_msps)) => {
self.ts_msps_found(ts_msps);
Ready(Ok((msps1, msps2))) => {
self.ts_msps_found(msps1, msps2);
continue;
}
Ready(Err(e)) => {
@@ -509,13 +534,8 @@ impl Stream for EventsStreamScylla {
},
FrState::ReadBack1(ref mut st) => match st.fut.poll_unpin(cx) {
Ready(Ok(item)) => {
if let Some(item) = self.back_1_done(item) {
item.verify();
item.output_info();
Ready(Some(Ok(ChannelEvents::Events(item))))
} else {
continue;
}
self.back_1_done(item);
continue;
}
Ready(Err(e)) => {
self.state = FrState::Done;
@@ -525,13 +545,8 @@ impl Stream for EventsStreamScylla {
},
FrState::ReadBack2(ref mut st) => match st.fut.poll_unpin(cx) {
Ready(Ok(item)) => {
if let Some(item) = self.back_2_done(item) {
item.verify();
item.output_info();
Ready(Some(Ok(ChannelEvents::Events(item))))
} else {
continue;
}
self.back_2_done(item);
continue;
}
Ready(Err(e)) => {
self.state = FrState::Done;
@@ -541,13 +556,14 @@ impl Stream for EventsStreamScylla {
},
FrState::ReadValues(ref mut st) => match st.fut.poll_unpin(cx) {
Ready(Ok(item)) => {
item.verify();
item.output_info();
if !st.next() {
info!("ReadValues exhausted");
trace!("ReadValues exhausted");
self.state = FrState::Done;
}
Ready(Some(Ok(ChannelEvents::Events(item))))
if item.len() > 0 {
self.outqueue.push_back(item);
}
continue;
}
Ready(Err(e)) => Ready(Some(Err(e))),
Pending => Pending,

4
streams/src/collect.rs
View File

@@ -57,7 +57,7 @@ where
break;
}
};
info!("collect_in_span see item");
debug!("collect_in_span see item");
match item {
Ok(item) => match item {
StreamItem::DataItem(item) => match item {
@@ -70,7 +70,7 @@ where
}
}
RangeCompletableItem::Data(mut item) => {
info!("collect_in_span sees {}", item.len());
debug!("collect_in_span sees {}", item.len());
if collector.is_none() {
let c = item.new_collector();
collector = Some(c);

2
streams/src/timebinnedjson.rs
View File

@@ -35,7 +35,7 @@ pub async fn timebinned_json(query: &BinnedQuery, chconf: &ChConf, cluster: &Clu
);
let inps = open_tcp_streams::<_, items_2::channelevents::ChannelEvents>(&rawquery, cluster).await?;
// TODO propagate also the max-buf-len for the first stage event reader:
let stream = { items_2::merger::Merger::new(inps, 1) };
let stream = { items_2::merger::Merger::new(inps, 128) };
let stream = stream::iter([empty]).chain(stream);
let stream = Box::pin(stream);
let stream = crate::timebin::TimeBinnedStream::new(stream, binned_range.edges(), do_time_weight, deadline);