daqbuf/daqingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Factor out SeriesWriter

Browse Source
This commit is contained in:
Dominik Werder
2024-01-11 08:23:24 +01:00
parent 722af64220
commit 0b4a5c0a34
15 changed files with 615 additions and 563 deletions
Download Patch File Download Diff File Expand all files Collapse all files

664
netfetch/src/ca/conn.rs
View File

File diff suppressed because it is too large Load Diff

68
netfetch/src/ca/connset.rs
View File

@@ -30,14 +30,13 @@ use futures_util::FutureExt;
use futures_util::Stream;
use futures_util::StreamExt;
use log::*;
use netpod::Database;
use scywr::iteminsertqueue::ChannelInfoItem;
use scywr::iteminsertqueue::ChannelStatus;
use scywr::iteminsertqueue::ChannelStatusItem;
use scywr::iteminsertqueue::QueryItem;
use serde::Serialize;
use series::series::CHANNEL_STATUS_DUMMY_SCALAR_TYPE;
use series::ChannelStatusSeriesId;
use serieswriter::writer::EstablishWorkerJob;
use statemap::ActiveChannelState;
use statemap::CaConnStateValue;
use statemap::ChannelState;
@@ -58,7 +57,6 @@ use std::collections::HashMap;
use std::collections::VecDeque;
use std::net::SocketAddr;
use std::net::SocketAddrV4;
use std::pin::pin;
use std::pin::Pin;
use std::sync::atomic;
use std::sync::Arc;
@@ -345,6 +343,7 @@ pub struct CaConnSet {
local_epics_hostname: String,
ca_conn_ress: BTreeMap<SocketAddr, CaConnRes>,
channel_states: ChannelStateMap,
channel_by_cssid: HashMap<ChannelStatusSeriesId, Channel>,
connset_inp_rx: Pin<Box<Receiver<CaConnSetEvent>>>,
channel_info_query_queue: VecDeque<ChannelInfoQuery>,
channel_info_query_sender: Pin<Box<SenderPolling<ChannelInfoQuery>>>,
@@ -373,7 +372,7 @@ pub struct CaConnSet {
ca_proto_stats: Arc<CaProtoStats>,
rogue_channel_count: u64,
connect_fail_count: usize,
name_by_cssid: HashMap<ChannelStatusSeriesId, String>,
establish_worker_tx: async_channel::Sender<EstablishWorkerJob>,
}
impl CaConnSet {
@@ -387,6 +386,7 @@ impl CaConnSet {
storage_insert_tx: Sender<VecDeque<QueryItem>>,
channel_info_query_tx: Sender<ChannelInfoQuery>,
ingest_opts: CaIngestOpts,
establish_worker_tx: async_channel::Sender<EstablishWorkerJob>,
) -> CaConnSetCtrl {
let (ca_conn_res_tx, ca_conn_res_rx) = async_channel::bounded(200);
let (connset_inp_tx, connset_inp_rx) = async_channel::bounded(200);
@@ -410,6 +410,7 @@ impl CaConnSet {
local_epics_hostname,
ca_conn_ress: BTreeMap::new(),
channel_states: ChannelStateMap::new(),
channel_by_cssid: HashMap::new(),
connset_inp_rx: Box::pin(connset_inp_rx),
channel_info_query_queue: VecDeque::new(),
channel_info_query_sender: Box::pin(SenderPolling::new(channel_info_query_tx.clone())),
@@ -439,7 +440,7 @@ impl CaConnSet {
ca_proto_stats: ca_proto_stats.clone(),
rogue_channel_count: 0,
connect_fail_count: 0,
name_by_cssid: HashMap::new(),
establish_worker_tx,
};
// TODO await on jh
let jh = tokio::spawn(CaConnSet::run(connset));
@@ -558,7 +559,53 @@ impl CaConnSet {
CaConnEventValue::EndOfStream => self.handle_ca_conn_eos(addr),
CaConnEventValue::ConnectFail => self.handle_connect_fail(addr),
CaConnEventValue::ChannelStatus(st) => {
error!("TODO handle_ca_conn_event update channel status view");
self.apply_ca_conn_health_update(addr, st)?;
// let sst = &mut self.channel_states;
// for (k, v) in st.channel_statuses {
// if let Some(ch) = self.channel_by_cssid.get(&k) {
// // Only when the channel is active we expect to receive status updates.
// if let Some(st) = sst.get_mut(ch) {
// if let ChannelStateValue::Active(st2) = &mut st.value {
// if let ActiveChannelState::WithStatusSeriesId {
// status_series_id,
// state: st3,
// } = st2
// {
// if let WithStatusSeriesIdStateInner::WithAddress { addr, state: st4 } =
// &mut st3.inner
// {
// if let WithAddressState::Assigned(st5) = st4 {
// } else {
// }
// } else {
// }
// } else {
// }
// } else {
// }
// st.value = ChannelStateValue::Active(ActiveChannelState::WithStatusSeriesId {
// status_series_id: (),
// state: WithStatusSeriesIdState {
// addr_find_backoff: todo!(),
// inner: todo!(),
// },
// });
// } else {
// // TODO this should be an error.
// }
// match v.channel_connected_info {
// conn::ChannelConnectedInfo::Disconnected => {}
// conn::ChannelConnectedInfo::Connecting => todo!(),
// conn::ChannelConnectedInfo::Connected => todo!(),
// conn::ChannelConnectedInfo::Error => todo!(),
// conn::ChannelConnectedInfo::Ended => todo!(),
// }
// } else {
// warn!("we do not know {:?}", k);
// }
// }
Ok(())
}
}
@@ -572,7 +619,8 @@ impl CaConnSet {
match res {
Ok(res) => {
let cssid = ChannelStatusSeriesId::new(res.series.into_inner().id());
self.name_by_cssid.insert(cssid.clone(), res.channel.clone());
self.channel_by_cssid
.insert(cssid.clone(), Channel::new(res.channel.clone()));
let add = ChannelAddWithStatusId {
backend: res.backend,
name: res.channel,
@@ -842,12 +890,11 @@ impl CaConnSet {
let tsnow = SystemTime::now();
self.rogue_channel_count = 0;
for (k, v) in res.channel_statuses {
let name = if let Some(x) = self.name_by_cssid.get(&v.cssid) {
let ch = if let Some(x) = self.channel_by_cssid.get(&k) {
x
} else {
return Err(Error::with_msg_no_trace(format!("unknown cssid {:?}", v.cssid)));
};
let ch = Channel::new(name.clone());
if let Some(st1) = self.channel_states.get_mut(&ch) {
if let ChannelStateValue::Active(st2) = &mut st1.value {
if let ActiveChannelState::WithStatusSeriesId {
@@ -995,6 +1042,7 @@ impl CaConnSet {
.ok_or_else(|| Error::with_msg_no_trace("no more channel_info_query_tx available"))?,
self.ca_conn_stats.clone(),
self.ca_proto_stats.clone(),
self.establish_worker_tx.clone(),
);
let conn_tx = conn.conn_command_tx();
let conn_stats = conn.stats();
@@ -1466,7 +1514,7 @@ impl CaConnSet {
}
fn handle_own_ticker_tick(mut self: Pin<&mut Self>, cx: &mut Context) -> Result<(), Error> {
debug!("handle_own_ticker_tick {}", Self::self_name());
// debug!("handle_own_ticker_tick {}", Self::self_name());
if !self.ready_for_end_of_stream() {
self.ticker = Self::new_self_ticker();
let _ = self.ticker.poll_unpin(cx);

85
netfetch/src/ca/proto.rs
View File

@@ -218,91 +218,6 @@ impl From<CaDataScalarValue> for scywr::iteminsertqueue::ScalarValue {
}
}
pub trait GetValHelp<T> {
type ScalTy: Clone;
fn get(&self) -> Result<&Self::ScalTy, Error>;
}
impl GetValHelp<i8> for CaDataValue {
type ScalTy = i8;
fn get(&self) -> Result<&Self::ScalTy, Error> {
match self {
CaDataValue::Scalar(v) => match v {
CaDataScalarValue::I8(v) => Ok(v),
_ => {
//let ty = any::type_name::<Self::ScalTy>();
Err(Error::GetValHelpInnerTypeMismatch)
}
},
_ => Err(Error::GetValHelpTodoWaveform),
}
}
}
impl GetValHelp<i16> for CaDataValue {
type ScalTy = i16;
fn get(&self) -> Result<&Self::ScalTy, Error> {
match self {
CaDataValue::Scalar(v) => match v {
CaDataScalarValue::I16(v) => Ok(v),
_ => {
//let ty = any::type_name::<Self::ScalTy>();
Err(Error::GetValHelpInnerTypeMismatch)
}
},
_ => Err(Error::GetValHelpTodoWaveform),
}
}
}
impl GetValHelp<i32> for CaDataValue {
type ScalTy = i32;
fn get(&self) -> Result<&Self::ScalTy, Error> {
match self {
CaDataValue::Scalar(v) => match v {
CaDataScalarValue::I32(v) => Ok(v),
_ => {
//let ty = any::type_name::<Self::ScalTy>();
Err(Error::GetValHelpInnerTypeMismatch)
}
},
_ => Err(Error::GetValHelpTodoWaveform),
}
}
}
impl GetValHelp<f32> for CaDataValue {
type ScalTy = f32;
fn get(&self) -> Result<&Self::ScalTy, Error> {
match self {
CaDataValue::Scalar(v) => match v {
CaDataScalarValue::F32(v) => Ok(v),
_ => {
//let ty = any::type_name::<Self::ScalTy>();
Err(Error::GetValHelpInnerTypeMismatch)
}
},
_ => Err(Error::GetValHelpTodoWaveform),
}
}
}
impl GetValHelp<f64> for CaDataValue {
type ScalTy = f64;
fn get(&self) -> Result<&Self::ScalTy, Error> {
match self {
CaDataValue::Scalar(v) => match v {
CaDataScalarValue::F64(v) => Ok(v),
_ => {
//let ty = any::type_name::<Self::ScalTy>();
Err(Error::GetValHelpInnerTypeMismatch)
}
},
_ => Err(Error::GetValHelpTodoWaveform),
}
}
}
#[derive(Clone, Debug)]
pub enum CaDataArrayValue {
I8(Vec<i8>),

2
netfetch/src/lib.rs
View File

@@ -5,11 +5,9 @@ pub mod errconv;
pub mod linuxhelper;
pub mod metrics;
pub mod netbuf;
pub mod patchcollect;
pub mod polltimer;
pub mod rt;
pub mod senderpolling;
#[cfg(test)]
pub mod test;
pub mod throttletrace;
pub mod timebin;

123
netfetch/src/patchcollect.rs
View File

@@ -1,123 +0,0 @@
use err::Error;
use items_0::timebin::TimeBinned;
use netpod::log::*;
use netpod::timeunits::SEC;
use netpod::TsNano;
use std::collections::VecDeque;
use std::mem;
pub struct PatchCollect {
patch_len: TsNano,
bin_len: TsNano,
bin_count: u64,
coll: Option<Box<dyn TimeBinned>>,
locked: bool,
outq: VecDeque<Box<dyn TimeBinned>>,
}
impl PatchCollect {
pub fn new(bin_len: TsNano, bin_count: u64) -> Self {
Self {
patch_len: TsNano(bin_len.0 * bin_count),
bin_len,
bin_count,
coll: None,
locked: false,
outq: VecDeque::new(),
}
}
pub fn patch_len(&self) -> TsNano {
self.patch_len.clone()
}
pub fn bin_len(&self) -> TsNano {
self.bin_len.clone()
}
pub fn bin_count(&self) -> u64 {
self.bin_count
}
pub fn ingest(&mut self, item: &mut dyn TimeBinned) -> Result<(), Error> {
let mut n1 = 0;
let mut item_len_exp = item.len();
loop {
n1 += 1;
if n1 > 20 {
return Err(Error::with_msg_no_trace("patchcollect too many iterations"));
}
info!("ingest loop item len {}", item.len());
if item.len() != item_len_exp {
return Err(Error::with_msg_no_trace(format!(
"patchcollect item_len_exp mismatch {} vs {}",
item.len(),
item_len_exp
)));
}
if item.len() == 0 {
break;
}
let coll = self.coll.get_or_insert_with(|| item.empty_like_self_box_time_binned());
let (ts1s, ts2s) = item.edges_slice();
let mut discard = false;
let mut emit = false;
let i1 = 0;
let mut i3 = item.len();
for (i2, (ts1, ts2)) in ts1s.iter().zip(ts2s).enumerate() {
info!("EDGE {}", ts1 / SEC);
if self.locked {
if ts2 % self.patch_len.0 == 0 {
info!("FOUND PATCH EDGE-END at {}", ts2 / SEC);
i3 = i2 + 1;
emit = true;
}
} else {
if ts1 % self.patch_len.0 == 0 {
info!("FOUND PATCH EDGE-BEG at {}", ts1 / SEC);
self.locked = true;
i3 = i2;
discard = true;
break;
}
}
}
if !self.locked {
info!("drain all");
item_len_exp = 0;
item.reset();
} else if discard {
let range = i1..i3;
info!("discard range-len {}", range.len());
item_len_exp -= range.len();
item.drain_into_tb(coll.as_mut(), range)?;
coll.reset();
} else if emit {
let range = i1..i3;
info!("take and emit range-len {}", range.len());
item_len_exp -= range.len();
item.drain_into_tb(coll.as_mut(), range)?;
if coll.len() != self.bin_count as usize {
error!("PatchCollect bin count mismatch {} vs {}", coll.len(), self.bin_count);
}
//info!("Patch EMIT {coll:?}");
let k = self.coll.take().unwrap();
self.outq.push_back(k);
} else {
let range = i1..i3;
info!("take all range-len {}", range.len());
item_len_exp = 0;
item.drain_into_tb(coll.as_mut(), range)?;
}
}
Ok(())
}
pub fn outq_len(&self) -> usize {
self.outq.len()
}
pub fn take_outq(&mut self) -> VecDeque<Box<dyn TimeBinned>> {
mem::replace(&mut self.outq, VecDeque::new())
}
}

311
netfetch/src/timebin.rs
View File

@@ -1,311 +0,0 @@
use crate::ca::proto;
use crate::ca::proto::CaDataValue;
use crate::ca::proto::CaEventValue;
use crate::patchcollect::PatchCollect;
use err::Error;
use items_0::scalar_ops::ScalarOps;
use items_0::timebin::TimeBinner;
use items_0::Appendable;
use items_0::Empty;
use items_0::Events;
use items_0::Resettable;
use items_2::binsdim0::BinsDim0;
use items_2::eventsdim0::EventsDim0;
use netpod::log::*;
use netpod::timeunits::SEC;
use netpod::BinnedRange;
use netpod::BinnedRangeEnum;
use netpod::ScalarType;
use netpod::Shape;
use netpod::TsNano;
use scywr::iteminsertqueue::QueryItem;
use scywr::iteminsertqueue::TimeBinPatchSimpleF32;
use series::SeriesId;
use std::any;
use std::any::Any;
use std::collections::VecDeque;
use std::time::SystemTime;
#[allow(unused)]
macro_rules! trace2 {
($($arg:tt)*) => {
if false {
trace!($($arg)*);
}
};
}
struct TickParams<'a> {
series: SeriesId,
acc: &'a mut Box<dyn Any + Send>,
tb: &'a mut Box<dyn TimeBinner>,
pc: &'a mut PatchCollect,
iiq: &'a mut VecDeque<QueryItem>,
}
pub struct ConnTimeBin {
did_setup: bool,
series: SeriesId,
acc: Box<dyn Any + Send>,
push_fn: Box<dyn Fn(SeriesId, &mut Box<dyn Any + Send>, u64, &CaEventValue) -> Result<(), Error> + Send>,
tick_fn: Box<dyn Fn(TickParams) -> Result<(), Error> + Send>,
events_binner: Option<Box<dyn TimeBinner>>,
patch_collect: PatchCollect,
}
impl ConnTimeBin {
pub fn empty() -> Self {
Self {
did_setup: false,
series: SeriesId::new(0),
acc: Box::new(()),
push_fn: Box::new(push::<i32>),
tick_fn: Box::new(tick::<i32>),
events_binner: None,
patch_collect: PatchCollect::new(TsNano(SEC * 60), 1),
}
}
pub fn setup_for(&mut self, series: SeriesId, scalar_type: &ScalarType, shape: &Shape) -> Result<(), Error> {
use ScalarType::*;
self.series = series;
let tsnow = SystemTime::now();
let ts0 = SEC * tsnow.duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs();
let bin_len = self.patch_collect.bin_len();
let range1 = BinnedRange {
bin_off: ts0 / bin_len.ns(),
bin_cnt: u64::MAX / bin_len.ns() - 10,
bin_len,
};
let binrange = BinnedRangeEnum::Time(range1);
//info!("binrange {binrange:?}");
let do_time_weight = true;
match shape {
Shape::Scalar => {
type Cont<T> = EventsDim0<T>;
match scalar_type {
I8 => {
type ST = i8;
trace2!("SCALAR {}", any::type_name::<ST>());
let cont = Cont::<ST>::empty();
self.events_binner =
Some(cont.as_time_binnable_ref().time_binner_new(binrange, do_time_weight));
self.acc = Box::new(cont);
self.push_fn = Box::new(push::<ST>);
self.tick_fn = Box::new(tick::<ST>);
self.did_setup = true;
}
I16 => {
type ST = i16;
trace2!("SCALAR {}", std::any::type_name::<ST>());
let cont = Cont::<ST>::empty();
self.events_binner =
Some(cont.as_time_binnable_ref().time_binner_new(binrange, do_time_weight));
self.acc = Box::new(cont);
self.push_fn = Box::new(push::<ST>);
self.tick_fn = Box::new(tick::<ST>);
self.did_setup = true;
}
I32 => {
type ST = i32;
trace2!("SCALAR {}", std::any::type_name::<ST>());
let cont = Cont::<ST>::empty();
self.events_binner =
Some(cont.as_time_binnable_ref().time_binner_new(binrange, do_time_weight));
self.acc = Box::new(cont);
self.push_fn = Box::new(push::<ST>);
self.tick_fn = Box::new(tick::<ST>);
self.did_setup = true;
}
F32 => {
type ST = f32;
trace2!("SCALAR {}", std::any::type_name::<ST>());
let cont = Cont::<ST>::empty();
self.events_binner =
Some(cont.as_time_binnable_ref().time_binner_new(binrange, do_time_weight));
self.acc = Box::new(cont);
self.push_fn = Box::new(push::<ST>);
self.tick_fn = Box::new(tick::<ST>);
self.did_setup = true;
}
F64 => {
type ST = f64;
trace2!("SCALAR {}", std::any::type_name::<ST>());
let cont = Cont::<ST>::empty();
self.events_binner =
Some(cont.as_time_binnable_ref().time_binner_new(binrange, do_time_weight));
self.acc = Box::new(cont);
self.push_fn = Box::new(push::<ST>);
self.tick_fn = Box::new(tick::<ST>);
self.did_setup = true;
}
STRING => {
trace2!("TODO setup_event_acc {:?} {:?}", scalar_type, shape);
}
_ => {
trace2!("TODO setup_event_acc {:?} {:?}", scalar_type, shape);
}
}
}
Shape::Wave(..) => {
//type Cont<T> = EventsDim1<T>;
match scalar_type {
_ => {
trace2!("TODO setup_event_acc {:?} {:?}", scalar_type, shape);
}
}
}
_ => {
trace2!("TODO setup_event_acc {:?} {:?}", scalar_type, shape);
}
}
Ok(())
}
pub fn push(&mut self, ts: u64, value: &CaEventValue) -> Result<(), Error> {
if !self.did_setup {
//return Err(Error::with_msg_no_trace("ConnTimeBin not yet set up"));
return Ok(());
}
let (f, acc) = (&self.push_fn, &mut self.acc);
f(self.series.clone(), acc, ts, value)
}
pub fn tick(&mut self, insert_item_queue: &mut VecDeque<QueryItem>) -> Result<(), Error> {
if !self.did_setup {
return Ok(());
}
let (f,) = (&self.tick_fn,);
let params = TickParams {
series: self.series.clone(),
acc: &mut self.acc,
tb: self.events_binner.as_mut().unwrap(),
pc: &mut self.patch_collect,
iiq: insert_item_queue,
};
f(params)
}
}
fn store_patch(series: SeriesId, pc: &mut PatchCollect, iiq: &mut VecDeque<QueryItem>) -> Result<(), Error> {
for item in pc.take_outq() {
if let Some(k) = item.as_any_ref().downcast_ref::<BinsDim0<f32>>() {
let ts0 = if let Some(x) = k.ts1s.front() {
*x
} else {
return Err(Error::with_msg_no_trace("patch contains no bins"));
};
let off = ts0 / pc.patch_len().0;
let off_msp = off / 1000;
let off_lsp = off % 1000;
let item = TimeBinPatchSimpleF32 {
series: series.clone(),
bin_len_sec: (pc.bin_len().ns() / SEC) as u32,
bin_count: pc.bin_count() as u32,
off_msp: off_msp as u32,
off_lsp: off_lsp as u32,
counts: k.counts.iter().map(|x| *x as i64).collect(),
mins: k.mins.iter().map(|x| *x).collect(),
maxs: k.maxs.iter().map(|x| *x).collect(),
avgs: k.avgs.iter().map(|x| *x).collect(),
};
let item = QueryItem::TimeBinPatchSimpleF32(item);
iiq.push_back(item);
} else {
error!("unexpected container!");
return Err(Error::with_msg_no_trace("timebin store_patch unexpected container"));
}
}
Ok(())
}
fn push<STY>(series: SeriesId, acc: &mut Box<dyn Any + Send>, ts: u64, ev: &CaEventValue) -> Result<(), Error>
where
STY: ScalarOps,
CaDataValue: proto::GetValHelp<STY, ScalTy = STY>,
{
let v = match proto::GetValHelp::<STY>::get(&ev.data) {
Ok(x) => x,
Err(e) => {
let msg = format!(
"GetValHelp mismatch: series {:?} STY {} data {:?} {e}",
series,
any::type_name::<STY>(),
ev.data
);
error!("{msg}");
return Err(Error::with_msg_no_trace(msg));
}
};
if let Some(c) = acc.downcast_mut::<EventsDim0<STY>>() {
c.push(ts, 0, v.clone());
Ok(())
} else {
// TODO report once and error out
error!("unexpected container");
//Err(Error::with_msg_no_trace("unexpected container"))
Ok(())
}
}
fn tick<STY>(params: TickParams) -> Result<(), Error>
where
STY: ScalarOps,
{
use items_0::WithLen;
let acc = params.acc;
let tb = params.tb;
let pc = params.pc;
let iiq = params.iiq;
if let Some(c) = acc.downcast_mut::<EventsDim0<STY>>() {
if c.len() >= 1 {
//info!("push events len {}", c.len());
tb.ingest(c);
c.reset();
if tb.bins_ready_count() >= 1 {
info!("store bins len {}", tb.bins_ready_count());
if let Some(mut bins) = tb.bins_ready() {
//info!("store bins {bins:?}");
let mut bins = bins.to_simple_bins_f32();
pc.ingest(bins.as_mut())?;
if pc.outq_len() != 0 {
store_patch(params.series.clone(), pc, iiq)?;
for item in pc.take_outq() {
if let Some(k) = item.as_any_ref().downcast_ref::<BinsDim0<f32>>() {
// TODO
//let off_msp =
let item = TimeBinPatchSimpleF32 {
series: params.series.clone(),
bin_len_sec: (pc.bin_len().ns() / SEC) as u32,
bin_count: pc.bin_count() as u32,
off_msp: 0,
off_lsp: 0,
counts: k.counts.iter().map(|x| *x as i64).collect(),
mins: k.mins.iter().map(|x| *x).collect(),
maxs: k.maxs.iter().map(|x| *x).collect(),
avgs: k.avgs.iter().map(|x| *x).collect(),
};
let item = QueryItem::TimeBinPatchSimpleF32(item);
iiq.push_back(item);
} else {
error!("unexpected container!");
}
}
}
Ok(())
} else {
error!("have bins but none returned");
Err(Error::with_msg_no_trace("have bins but none returned"))
}
} else {
Ok(())
}
} else {
Ok(())
}
} else {
error!("unexpected container");
//Err(Error::with_msg_no_trace("unexpected container"))
Ok(())
}
}