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

WIP before remove of channelwriter

Browse Source
This commit is contained in:
Dominik Werder
2023-08-24 20:42:50 +02:00
parent 26ee621a84
commit 9b03cfe666
20 changed files with 385 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
netfetch/src/ca/search.rs
View File

@@ -163,7 +163,7 @@ pub async fn ca_search(opts: CaIngestOpts, channels: &Vec<String>) -> Result<(),
gw_addrs.push(addr);
}
Err(e) => {
error!("can not resolve {s} {e}");
warn!("can not resolve {s} {e}");
}
}
}

238
netfetch/src/channelwriter.rs
View File

@@ -8,14 +8,6 @@ use netpod::timeunits::SEC;
use netpod::ByteOrder;
use netpod::ScalarType;
use netpod::Shape;
use scylla::batch::Batch;
use scylla::batch::BatchType;
use scylla::frame::value::BatchValues;
use scylla::frame::value::ValueList;
use scylla::prepared_statement::PreparedStatement;
use scylla::transport::errors::QueryError;
use scylla::QueryResult;
use scylla::Session as ScySession;
use std::mem;
use std::pin::Pin;
use std::sync::Arc;
@@ -24,236 +16,6 @@ use std::task::Poll;
use std::time::Duration;
use std::time::Instant;
pub struct ScyQueryFut<'a> {
fut: Pin<Box<dyn Future<Output = Result<QueryResult, QueryError>> + Send + 'a>>,
}
impl<'a> ScyQueryFut<'a> {
pub fn new<V>(scy: &'a ScySession, query: Option<&'a PreparedStatement>, values: V) -> Self
where
V: ValueList + Send + 'static,
{
//let fut = scy.execute(query, values);
let fut = futures_util::future::ready(Err(QueryError::TimeoutError));
Self { fut: Box::pin(fut) }
}
}
impl<'a> Future for ScyQueryFut<'a> {
type Output = Result<(), Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
use Poll::*;
match self.fut.poll_unpin(cx) {
Ready(k) => match k {
Ok(_) => Ready(Ok(())),
Err(e) => Ready(Err(e).err_conv()),
},
Pending => Pending,
}
}
}
pub struct ScyBatchFut<'a> {
fut: Pin<Box<dyn Future<Output = Result<QueryResult, QueryError>> + 'a>>,
polled: usize,
ts_create: Instant,
ts_poll_start: Instant,
}
impl<'a> ScyBatchFut<'a> {
pub fn new<V>(scy: &'a ScySession, batch: &'a Batch, values: V) -> Self
where
V: BatchValues + Send + Sync + 'static,
{
let fut = scy.batch(batch, values);
let tsnow = Instant::now();
Self {
fut: Box::pin(fut),
polled: 0,
ts_create: tsnow,
ts_poll_start: tsnow,
}
}
}
impl<'a> Future for ScyBatchFut<'a> {
type Output = Result<(), Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
use Poll::*;
if self.polled == 0 {
self.ts_poll_start = Instant::now();
}
self.polled += 1;
match self.fut.poll_unpin(cx) {
Ready(k) => match k {
Ok(_) => {
trace!("ScyBatchFut done Ok");
Ready(Ok(()))
}
Err(e) => {
let tsnow = Instant::now();
let dt_created = tsnow.duration_since(self.ts_create).as_secs_f32() * 1e3;
let dt_polled = tsnow.duration_since(self.ts_poll_start).as_secs_f32() * 1e3;
warn!(
"ScyBatchFut polled {} dt_created {:6.2} ms dt_polled {:6.2} ms",
self.polled, dt_created, dt_polled
);
warn!("ScyBatchFut done Err {e:?}");
Ready(Err(e).err_conv())
}
},
Pending => Pending,
}
}
}
pub struct ScyBatchFutGen<'a> {
fut: Pin<Box<dyn Future<Output = Result<QueryResult, QueryError>> + Send + 'a>>,
polled: usize,
ts_create: Instant,
ts_poll_start: Instant,
}
impl<'a> ScyBatchFutGen<'a> {
pub fn new<V>(scy: &'a ScySession, batch: &'a Batch, values: V) -> Self
where
V: BatchValues + Send + Sync + 'static,
{
let fut = scy.batch(batch, values);
let tsnow = Instant::now();
Self {
fut: Box::pin(fut),
polled: 0,
ts_create: tsnow,
ts_poll_start: tsnow,
}
}
}
impl<'a> Future for ScyBatchFutGen<'a> {
type Output = Result<(), Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
use Poll::*;
if self.polled == 0 {
self.ts_poll_start = Instant::now();
}
self.polled += 1;
match self.fut.poll_unpin(cx) {
Ready(k) => match k {
Ok(_) => {
trace!("ScyBatchFutGen done Ok");
Ready(Ok(()))
}
Err(e) => {
let tsnow = Instant::now();
let dt_created = tsnow.duration_since(self.ts_create).as_secs_f32() * 1e3;
let dt_polled = tsnow.duration_since(self.ts_poll_start).as_secs_f32() * 1e3;
warn!(
"ScyBatchFutGen polled {} dt_created {:6.2} ms dt_polled {:6.2} ms",
self.polled, dt_created, dt_polled
);
warn!("ScyBatchFutGen done Err {e:?}");
Ready(Err(e).err_conv())
}
},
Pending => Pending,
}
}
}
pub struct InsertLoopFut<'a> {
futs: Vec<Pin<Box<dyn Future<Output = Result<QueryResult, QueryError>> + Send + 'a>>>,
fut_ix: usize,
polled: usize,
ts_create: Instant,
ts_poll_start: Instant,
}
impl<'a> InsertLoopFut<'a> {
pub fn new<V>(scy: &'a ScySession, query: Option<&'a PreparedStatement>, values: Vec<V>, skip_insert: bool) -> Self
where
V: ValueList + Send + Sync + 'static,
{
let mut values = values;
if skip_insert {
values.clear();
}
// TODO
// Can I store the values in some better generic form?
// Or is it acceptable to generate all insert futures right here and poll them later?
let futs: Vec<_> = values
.into_iter()
.map(|vs| {
//let fut = scy.execute(query, vs);
let fut = futures_util::future::ready(Err(QueryError::TimeoutError));
Box::pin(fut) as _
})
.collect();
let tsnow = Instant::now();
Self {
futs,
fut_ix: 0,
polled: 0,
ts_create: tsnow,
ts_poll_start: tsnow,
}
}
}
impl<'a> Future for InsertLoopFut<'a> {
type Output = Result<(), Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
use Poll::*;
if self.polled == 0 {
self.ts_poll_start = Instant::now();
}
self.polled += 1;
if self.futs.is_empty() {
return Ready(Ok(()));
}
loop {
let fut_ix = self.fut_ix;
break match self.futs[fut_ix].poll_unpin(cx) {
Ready(k) => match k {
Ok(_) => {
self.fut_ix += 1;
if self.fut_ix >= self.futs.len() {
if false {
let tsnow = Instant::now();
let dt_created = tsnow.duration_since(self.ts_create).as_secs_f32() * 1e3;
let dt_polled = tsnow.duration_since(self.ts_poll_start).as_secs_f32() * 1e3;
info!(
"InsertLoopFut polled {} dt_created {:6.2} ms dt_polled {:6.2} ms",
self.polled, dt_created, dt_polled
);
}
continue;
} else {
Ready(Ok(()))
}
}
Err(e) => {
let tsnow = Instant::now();
let dt_created = tsnow.duration_since(self.ts_create).as_secs_f32() * 1e3;
let dt_polled = tsnow.duration_since(self.ts_poll_start).as_secs_f32() * 1e3;
warn!(
"InsertLoopFut polled {} dt_created {:6.2} ms dt_polled {:6.2} ms",
self.polled, dt_created, dt_polled
);
warn!("InsertLoopFut done Err {e:?}");
Ready(Err(e).err_conv())
}
},
Pending => Pending,
};
}
}
}
pub struct ChannelWriteRes {
pub nrows: u32,
pub dt: Duration,

45
netfetch/src/scylla.rs
View File

@@ -3,53 +3,8 @@ use futures_util::StreamExt;
#[allow(unused)]
use netpod::log::*;
use netpod::ScyllaConfig;
use scylla::execution_profile::ExecutionProfileBuilder;
use scylla::statement::Consistency;
use scylla::transport::errors::DbError;
use scylla::transport::errors::QueryError;
use scylla::Session;
use std::sync::Arc;
pub async fn create_session(scyconf: &ScyllaConfig) -> Result<Arc<Session>, Error> {
let scy = scylla::SessionBuilder::new()
.known_nodes(&scyconf.hosts)
.use_keyspace(&scyconf.keyspace, true)
.default_execution_profile_handle(
ExecutionProfileBuilder::default()
.consistency(Consistency::LocalOne)
.build()
.into_handle(),
)
.build()
.await
.map_err(|e| Error::from(format!("{e}")))?;
let scy = Arc::new(scy);
Ok(scy)
}
async fn has_table(name: &str, scy: &Session, scyconf: &ScyllaConfig) -> Result<bool, Error> {
let ks = scy
.get_keyspace()
.ok_or_else(|| Error::with_msg_no_trace("session is not using a keyspace yet"))?;
let mut res = scy
.query_iter(
"select table_name from system_schema.tables where keyspace_name = ?",
(ks.as_ref(),),
)
.await
.map_err(|e| e.to_string())
.map_err(Error::from)?;
while let Some(k) = res.next().await {
let row = k.map_err(|e| e.to_string()).map_err(Error::from)?;
if let Some(table_name) = row.columns[0].as_ref().unwrap().as_text() {
if table_name == name {
return Ok(true);
}
}
}
Ok(false)
}
async fn check_table_exist(name: &str, scy: &Session) -> Result<bool, Error> {
match scy.query(format!("select * from {} limit 1", name), ()).await {
Ok(_) => Ok(true),

2
netfetch/src/zmtp.rs
View File

@@ -17,7 +17,7 @@ use futures_util::FutureExt;
use futures_util::StreamExt;
use futures_util::TryFutureExt;
use log::*;
use scylla::Session as ScySession;
use scywr::session::ScySession;
use std::io;
use std::net::SocketAddr;
use std::pin::Pin;