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

Update deps

Browse Source
This commit is contained in:
Dominik Werder
2023-05-10 09:42:10 +02:00
parent c43a558c0e
commit 8818fb865f
8 changed files with 163 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,4 +1,6 @@
use super::proto;
use super::proto::CaDataValue;
use super::proto::CaEventValue;
use super::proto::CaItem;
use super::proto::CaMsg;
use super::proto::CaMsgTy;
@@ -30,9 +32,11 @@ use futures_util::Future;
use futures_util::FutureExt;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::scalar_ops::ScalarOps;
use items_0::Appendable;
use items_0::Empty;
use items_2::eventsdim0::EventsDim0;
use items_2::eventsdim1::EventsDim1;
use log::*;
use netpod::timeunits::*;
use netpod::ScalarType;
@@ -440,6 +444,7 @@ pub struct CaConn {
Pin<Box<dyn Future<Output = Result<(Cid, u32, u16, u16, Existence<SeriesId>), Error>> + Send>>,
>,
events_acc: Box<dyn Any + Send>,
events_acc_func: Box<dyn Fn(&mut CaConn, u64, CaEventValue) -> Result<(), Error> + Send>,
}
impl CaConn {
@@ -493,6 +498,7 @@ impl CaConn {
series_lookup_schedule: BTreeMap::new(),
series_lookup_futs: FuturesUnordered::new(),
events_acc: Box::new(()),
events_acc_func: Box::new(Self::event_acc_push::<i32>),
}
}
@@ -1029,29 +1035,16 @@ impl CaConn {
}
}
// TODO write generic.
// TODO store a pointer to the type-resolved function for faster usage.
fn event_acc_push(&mut self, ts: u64, ev: proto::EventAddRes) -> Result<(), Error> {
match &ev.value.data {
proto::CaDataValue::Scalar(k) => {
use proto::CaDataScalarValue::*;
match k {
F64(v) => {
if let Some(c) = self.events_acc.downcast_mut::<EventsDim0<f64>>() {
c.push(ts, 0, v.clone());
// TODO check for a max length.
// Also check at every check-tick.
} else {
}
}
_ => {
// TODO
}
}
}
proto::CaDataValue::Array(_) => {
// TODO
}
fn event_acc_push<STY>(this: &mut CaConn, ts: u64, ev: CaEventValue) -> Result<(), Error>
where
STY: ScalarOps,
CaDataValue: proto::GetValHelp<STY, ScalTy = STY>,
{
let v = proto::GetValHelp::<STY>::get(&ev.data)?;
if let Some(c) = this.events_acc.downcast_mut::<EventsDim0<STY>>() {
c.push(ts, 0, v.clone());
// TODO check for a max length.
// Also check at every check-tick.
}
Ok(())
}
@@ -1059,16 +1052,63 @@ impl CaConn {
fn setup_event_acc(&mut self, scalar_type: &ScalarType, shape: &Shape) -> Result<(), Error> {
use ScalarType::*;
match shape {
Shape::Scalar => match scalar_type {
F64 => {
self.events_acc = Box::new(EventsDim0::<f64>::empty());
Shape::Scalar => {
type Cont<T> = EventsDim0<T>;
match scalar_type {
I8 => {
self.events_acc = Box::new(Cont::<i8>::empty());
}
I16 => {
self.events_acc = Box::new(Cont::<i16>::empty());
}
I32 => {
type ST = i32;
self.events_acc = Box::new(Cont::<ST>::empty());
let f: Box<dyn Fn(&mut CaConn, u64, CaEventValue) -> Result<(), Error>> =
Box::new(Self::event_acc_push::<ST>);
}
F32 => {
type ST = f32;
self.events_acc = Box::new(Cont::<ST>::empty());
let f: Box<dyn Fn(&mut CaConn, u64, CaEventValue) -> Result<(), Error>> =
Box::new(Self::event_acc_push::<ST>);
}
F64 => {
type ST = f64;
self.events_acc = Box::new(Cont::<ST>::empty());
let f: Box<dyn Fn(&mut CaConn, u64, CaEventValue) -> Result<(), Error>> =
Box::new(Self::event_acc_push::<ST>);
}
_ => {
warn!("TODO setup_event_acc {:?} {:?}", scalar_type, shape);
}
}
_ => {
// TODO
}
Shape::Wave(..) => {
type Cont<T> = EventsDim1<T>;
match scalar_type {
I8 => {
self.events_acc = Box::new(Cont::<i8>::empty());
}
I16 => {
self.events_acc = Box::new(Cont::<i16>::empty());
}
I32 => {
self.events_acc = Box::new(Cont::<i32>::empty());
}
F32 => {
self.events_acc = Box::new(Cont::<f32>::empty());
}
F64 => {
self.events_acc = Box::new(Cont::<f64>::empty());
}
_ => {
warn!("TODO setup_event_acc {:?} {:?}", scalar_type, shape);
}
}
},
}
_ => {
// TODO
warn!("TODO setup_event_acc {:?} {:?}", scalar_type, shape);
}
}
Ok(())

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

@@ -166,6 +166,50 @@ pub enum CaDataScalarValue {
Bool(bool),
}
pub trait GetValHelp<T> {
type ScalTy: Clone;
fn get(&self) -> Result<&Self::ScalTy, Error>;
}
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),
_ => Err(Error::with_msg_no_trace("GetValHelp inner type mismatch")),
},
_ => Err(Error::with_msg_no_trace("GetValHelp inner type mismatch")),
}
}
}
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),
_ => Err(Error::with_msg_no_trace("GetValHelp inner type mismatch")),
},
_ => Err(Error::with_msg_no_trace("GetValHelp inner type mismatch")),
}
}
}
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),
_ => Err(Error::with_msg_no_trace("GetValHelp inner type mismatch")),
},
_ => Err(Error::with_msg_no_trace("GetValHelp inner type mismatch")),
}
}
}
#[derive(Clone, Debug)]
pub enum CaDataArrayValue {
I8(Vec<i8>),

8
netfetch/src/ca/store.rs
View File

@@ -1,6 +1,7 @@
use err::Error;
use futures_util::StreamExt;
use netpod::ScyllaConfig;
use scylla::execution_profile::ExecutionProfileBuilder;
use scylla::prepared_statement::PreparedStatement;
use scylla::statement::Consistency;
use scylla::Session as ScySession;
@@ -59,8 +60,13 @@ impl DataStore {
pub async fn new(scyconf: &ScyllaConfig) -> Result<Self, Error> {
let scy = scylla::SessionBuilder::new()
.known_nodes(&scyconf.hosts)
.default_consistency(Consistency::LocalOne)
.use_keyspace(&scyconf.keyspace, true)
.default_execution_profile_handle(
ExecutionProfileBuilder::default()
.consistency(Consistency::LocalOne)
.build()
.into_handle(),
)
.build()
.await
.map_err(|e| Error::with_msg_no_trace(format!("{e:?}")))?;

1
netfetch/src/store.rs
View File

@@ -62,6 +62,7 @@ impl IntoSimplerError for QueryError {
QueryError::TooManyOrphanedStreamIds(e) => Error::DbError(e.to_string()),
QueryError::UnableToAllocStreamId => Error::DbError(e.to_string()),
QueryError::RequestTimeout(e) => Error::DbError(e.to_string()),
QueryError::TranslationError(e) => Error::DbError(e.to_string()),
}
}
}