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

Refactor different queues into common type

Browse Source
This commit is contained in:
Dominik Werder
2024-05-02 17:06:05 +02:00
parent 0704ad418c
commit 3827514900
10 changed files with 214 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netfetch/src/ca/beacons.rs
+5 -5
View File
@@ -7,10 +7,10 @@ use log::*;
use netpod::ScalarType;
use netpod::Shape;
use netpod::TsNano;
use scywr::insertqueues::InsertDeques;
use scywr::iteminsertqueue::DataValue;
use scywr::iteminsertqueue::ScalarValue;
use serieswriter::writer::SeriesWriter;
use std::collections::VecDeque;
use std::io::Cursor;
use std::net::Ipv4Addr;
use std::time::SystemTime;
@@ -36,7 +36,7 @@ pub async fn listen_beacons(
sock.set_broadcast(true).unwrap();
let mut buf = Vec::new();
buf.resize(1024 * 4, 0);
let mut item_qu = VecDeque::new();
let mut iqdqs = InsertDeques::new();
loop {
let bb = &mut buf;
let (n, remote) = taskrun::tokio::select! {
@@ -65,12 +65,12 @@ pub async fn listen_beacons(
let ts_local = ts;
let blob = addr_u32 as i64;
let val = DataValue::Scalar(ScalarValue::I64(blob));
writer.write(ts, ts_local, val, &mut item_qu)?;
writer.write(ts, ts_local, val, &mut iqdqs)?;
}
}
if item_qu.len() != 0 {
if iqdqs.len() != 0 {
// TODO deliver to insert queue
item_qu.clear();
iqdqs.clear();
}
}
Ok(())
netfetch/src/ca/conn.rs
+57 -35
View File
@@ -30,6 +30,7 @@ use proto::CaMsgTy;
use proto::CaProto;
use proto::CreateChan;
use proto::EventAdd;
use scywr::insertqueues::InsertDeques;
use scywr::iteminsertqueue as scywriiq;
use scywr::iteminsertqueue::Accounting;
use scywr::iteminsertqueue::DataValue;
@@ -758,7 +759,7 @@ pub struct CaConn {
channel_status_emit_last: Instant,
tick_last_writer: Instant,
init_state_count: u64,
insert_item_queue: VecDeque<QueryItem>,
iqdqs: InsertDeques,
remote_addr_dbg: SocketAddrV4,
local_epics_hostname: String,
stats: Arc<CaConnStats>,
@@ -824,7 +825,7 @@ impl CaConn {
cid_by_sid: HashMap::new(),
channel_status_emit_last: tsnow,
tick_last_writer: tsnow,
insert_item_queue: VecDeque::new(),
iqdqs: InsertDeques::new(),
remote_addr_dbg,
local_epics_hostname,
stats,
@@ -906,7 +907,8 @@ impl CaConn {
};
self.channel_state_on_shutdown(channel_reason);
let addr = self.remote_addr_dbg.clone();
self.insert_item_queue
self.iqdqs
.lt_rf3_rx
.push_back(QueryItem::ConnectionStatus(ConnectionStatusItem {
ts: self.tmp_ts_poll,
addr,
@@ -1004,7 +1006,7 @@ impl CaConn {
cssid: st2.channel.cssid.clone(),
status: ChannelStatus::Opened,
});
self.insert_item_queue.push_back(item);
self.iqdqs.lt_rf3_rx.push_back(item);
}
let name = conf.conf.name();
if name.starts_with("TEST:PEAKING:") {
@@ -1171,7 +1173,7 @@ impl CaConn {
cssid: cssid.clone(),
status: ChannelStatus::Closed(channel_reason.clone()),
});
self.insert_item_queue.push_back(item);
self.iqdqs.lt_rf3_rx.push_back(item);
*chst = ChannelState::Ended(cssid);
}
ChannelState::Error(..) => {
@@ -1337,9 +1339,9 @@ impl CaConn {
});
let crst = &mut st.channel;
let writer = &mut st.writer;
let iiq = &mut self.insert_item_queue;
let iqdqs = &mut self.iqdqs;
let stats = self.stats.as_ref();
Self::event_add_ingest(ev.payload_len, ev.value, crst, writer, iiq, tsnow, stnow, stats)?;
Self::event_add_ingest(ev.payload_len, ev.value, crst, writer, iqdqs, tsnow, stnow, stats)?;
}
ReadingState::Monitoring(st2) => {
match &mut st2.mon2state {
@@ -1353,9 +1355,9 @@ impl CaConn {
}
let crst = &mut st.channel;
let writer = &mut st.writer;
let iiq = &mut self.insert_item_queue;
let iqdqs = &mut self.iqdqs;
let stats = self.stats.as_ref();
Self::event_add_ingest(ev.payload_len, ev.value, crst, writer, iiq, tsnow, stnow, stats)?;
Self::event_add_ingest(ev.payload_len, ev.value, crst, writer, iqdqs, tsnow, stnow, stats)?;
}
ReadingState::StopMonitoringForPolling(st2) => {
// TODO count for metrics
@@ -1483,9 +1485,9 @@ impl CaConn {
// TODO maintain histogram of read-notify latencies
self.read_ioids.remove(ioid);
st2.tick = PollTickState::Idle(tsnow);
let iiq = &mut self.insert_item_queue;
let iqdqs = &mut self.iqdqs;
let stats = self.stats.as_ref();
Self::read_notify_res_for_write(ev, st, iiq, stnow, tsnow, stats)?;
Self::read_notify_res_for_write(ev, st, iqdqs, stnow, tsnow, stats)?;
}
},
ReadingState::EnableMonitoring(..) => {
@@ -1506,9 +1508,9 @@ impl CaConn {
}
self.read_ioids.remove(&ioid);
st2.mon2state = Monitoring2State::Passive(Monitoring2PassiveState { tsbeg: tsnow });
let iiq = &mut self.insert_item_queue;
let iqdqs = &mut self.iqdqs;
let stats = self.stats.as_ref();
Self::read_notify_res_for_write(ev, st, iiq, stnow, tsnow, stats)?;
Self::read_notify_res_for_write(ev, st, iqdqs, stnow, tsnow, stats)?;
}
},
ReadingState::StopMonitoringForPolling(..) => {
@@ -1531,14 +1533,14 @@ impl CaConn {
fn read_notify_res_for_write(
ev: proto::ReadNotifyRes,
st: &mut WritableState,
iiq: &mut VecDeque<QueryItem>,
iqdqs: &mut InsertDeques,
stnow: SystemTime,
tsnow: Instant,
stats: &CaConnStats,
) -> Result<(), Error> {
let crst = &mut st.channel;
let writer = &mut st.writer;
Self::event_add_ingest(ev.payload_len, ev.value, crst, writer, iiq, tsnow, stnow, stats)?;
Self::event_add_ingest(ev.payload_len, ev.value, crst, writer, iqdqs, tsnow, stnow, stats)?;
Ok(())
}
@@ -1547,7 +1549,7 @@ impl CaConn {
value: CaEventValue,
crst: &mut CreatedState,
writer: &mut SeriesWriter,
iiq: &mut VecDeque<QueryItem>,
iqdqs: &mut InsertDeques,
tsnow: Instant,
stnow: SystemTime,
stats: &CaConnStats,
@@ -1577,7 +1579,7 @@ impl CaConn {
Self::check_ev_value_data(&value.data, writer.scalar_type())?;
{
let val: DataValue = value.data.into();
writer.write(TsNano::from_ns(ts), TsNano::from_ns(ts_local), val, iiq)?;
writer.write(TsNano::from_ns(ts), TsNano::from_ns(ts_local), val, iqdqs)?;
}
}
if false {
@@ -2144,7 +2146,8 @@ impl CaConn {
Ok(Ok(tcp)) => {
self.stats.tcp_connected.inc();
let addr = addr.clone();
self.insert_item_queue
self.iqdqs
.lt_rf3_rx
.push_back(QueryItem::ConnectionStatus(ConnectionStatusItem {
ts: self.tmp_ts_poll,
addr,
@@ -2164,7 +2167,8 @@ impl CaConn {
Ok(Err(e)) => {
debug!("error connect to {addr} {e}");
let addr = addr.clone();
self.insert_item_queue
self.iqdqs
.lt_rf3_rx
.push_back(QueryItem::ConnectionStatus(ConnectionStatusItem {
ts: self.tmp_ts_poll,
addr,
@@ -2177,7 +2181,8 @@ impl CaConn {
// TODO log with exponential backoff
debug!("timeout connect to {addr} {e}");
let addr = addr.clone();
self.insert_item_queue
self.iqdqs
.lt_rf3_rx
.push_back(QueryItem::ConnectionStatus(ConnectionStatusItem {
ts: self.tmp_ts_poll,
addr,
@@ -2237,7 +2242,7 @@ impl CaConn {
self.stats.loop2_count.inc();
if self.is_shutdown() {
break;
} else if self.insert_item_queue.len() >= self.opts.insert_queue_max {
} else if self.iqdqs.len() >= self.opts.insert_queue_max {
break;
} else {
match self.handle_conn_state(tsnow, cx) {
@@ -2365,7 +2370,7 @@ impl CaConn {
count,
bytes,
});
self.insert_item_queue.push_back(item);
self.iqdqs.lt_rf3_rx.push_back(item);
}
}
}
@@ -2379,7 +2384,7 @@ impl CaConn {
for (_, chconf) in &mut self.channels {
let chst = &mut chconf.state;
if let ChannelState::Writable(st2) = chst {
st2.writer.tick(&mut self.insert_item_queue)?;
st2.writer.tick(&mut self.iqdqs)?;
}
}
Ok(())
@@ -2392,13 +2397,11 @@ impl CaConn {
fn queues_out_flushed(&self) -> bool {
debug!(
"async out flushed iiq {} {} caout {}",
self.insert_item_queue.is_empty(),
self.iqdqs.len() == 0,
self.storage_insert_sender.is_idle(),
self.ca_conn_event_out_queue.is_empty()
);
self.insert_item_queue.is_empty()
&& self.storage_insert_sender.is_idle()
&& self.ca_conn_event_out_queue.is_empty()
self.iqdqs.len() == 0 && self.storage_insert_sender.is_idle() && self.ca_conn_event_out_queue.is_empty()
}
fn attempt_flush_queue<T, Q, FB, FS>(
@@ -2415,9 +2418,10 @@ impl CaConn {
FB: Fn(&mut VecDeque<T>) -> Option<Q>,
FS: Fn(&Q),
{
let self_name = "attempt_flush_queue";
use Poll::*;
if qu.len() != 0 {
trace_flush_queue!("attempt_flush_queue id {:7} len {}", id, qu.len());
trace_flush_queue!("{self_name} id {:10} len {}", id, qu.len());
}
let mut have_progress = false;
let mut i = 0;
@@ -2440,7 +2444,7 @@ impl CaConn {
if sp.is_sending() {
match sp.poll_unpin(cx) {
Ready(Ok(())) => {
trace_flush_queue!("attempt_flush_queue id {:7} send done", id);
trace_flush_queue!("{self_name} id {:10} send done", id);
have_progress = true;
}
Ready(Err(e)) => {
@@ -2485,6 +2489,24 @@ macro_rules! flush_queue {
};
}
macro_rules! flush_queue_dqs {
($self:expr, $qu:ident, $sp:ident, $batcher:expr, $loop_max:expr, $have:expr, $id:expr, $cx:expr, $stats:expr) => {
let obj = $self.as_mut().get_mut();
let qu = &mut obj.iqdqs.$qu;
let sp = &mut obj.$sp;
match Self::attempt_flush_queue(qu, sp, $batcher, $loop_max, $cx, $id, $stats) {
Ok(Ready(Some(()))) => {
*$have.0 |= true;
}
Ok(Ready(None)) => {}
Ok(Pending) => {
*$have.1 |= true;
}
Err(e) => break Ready(Some(CaConnEvent::err_now(e))),
}
};
}
fn send_individual<T>(qu: &mut VecDeque<T>) -> Option<T> {
qu.pop_front()
}
@@ -2514,7 +2536,7 @@ impl Stream for CaConn {
let lts1 = Instant::now();
self.stats.poll_loop_begin().inc();
let qlen = self.insert_item_queue.len();
let qlen = self.iqdqs.len();
if qlen >= self.opts.insert_queue_max * 2 / 3 {
self.stats.insert_item_queue_pressure().inc();
} else if qlen >= self.opts.insert_queue_max {
@@ -2543,8 +2565,8 @@ impl Stream for CaConn {
}
{
let iiq = &self.insert_item_queue;
self.stats.iiq_len().ingest(iiq.len() as u32);
let n = self.iqdqs.len();
self.stats.iiq_len().ingest(n as u32);
}
{
@@ -2552,14 +2574,14 @@ impl Stream for CaConn {
let stats_fn = move |item: &VecDeque<QueryItem>| {
stats2.iiq_batch_len().ingest(item.len() as u32);
};
flush_queue!(
flush_queue_dqs!(
self,
insert_item_queue,
st_rf1_rx,
storage_insert_sender,
send_batched::<256, _>,
32,
(&mut have_progress, &mut have_pending),
"strg",
"iq_st_rf1",
cx,
stats_fn
);
netfetch/src/conf.rs
+22 -5
View File
@@ -246,6 +246,7 @@ async fn parse_channel_config_txt(fname: &Path, re_p: Regex, re_n: Regex) -> Res
short_term: Some(ChannelReadConfig::Monitor),
medium_term: None,
long_term: None,
is_polled: false,
},
};
conf.channels.push(item);
@@ -274,6 +275,8 @@ pub struct IngestConfigArchiving {
#[serde(default, skip_serializing_if = "Option::is_none")]
#[serde(with = "serde_option_channel_read_config")]
long_term: Option<ChannelReadConfig>,
#[serde(default, skip_serializing_if = "bool_is_false")]
is_polled: bool,
}
fn bool_is_false(x: &bool) -> bool {
@@ -368,7 +371,7 @@ mod serde_option_channel_read_config {
type Value = Option<ChannelReadConfig>;
fn expecting(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
write!(fmt, "keyword `Monitor`, an integer, or not this field at all")
write!(fmt, "keyword `Monitor`, keyword `None`, an integer, or missing")
}
fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
@@ -390,8 +393,9 @@ mod serde_option_channel_read_config {
where
E: de::Error,
{
if v < 1 || v > 108000 {
let e = E::custom(format!("unsupported value {v:?}, polling must be in range 1..108000"));
let max = 108000;
if v < 1 || v > max {
let e = E::custom(format!("unsupported value {v:?}, polling must be in range 1..{max:?}"));
return Err(e);
}
Ok(Some(ChannelReadConfig::Poll(Duration::from_secs(v as u64))))
@@ -401,8 +405,9 @@ mod serde_option_channel_read_config {
where
E: de::Error,
{
if v < 1 || v > 108000 {
let e = E::custom(format!("unsupported value {v:?}, polling must be in range 1..108000"));
let max = 108000;
if v < 1 || v > max {
let e = E::custom(format!("unsupported value {v:?}, polling must be in range 1..{max:?}"));
return Err(e);
}
self.visit_u64(v as u64)
@@ -436,6 +441,17 @@ CH-02:
short_term: Monitor
CH-03:
archiving_configuration:
CH-04:
archiving_configuration:
short_term: None
medium_term: None
long_term: 3600
is_polled: true
CH-05:
archiving_configuration:
short_term: None
medium_term: None
long_term: Monitor
"###;
let x: BTreeMap<String, ChannelConfigParse> = serde_yaml::from_str(inp).unwrap();
assert_eq!(
@@ -501,6 +517,7 @@ impl ChannelConfig {
short_term: Some(ChannelReadConfig::Monitor),
medium_term: None,
long_term: None,
is_polled: false,
},
}
}
netfetch/src/metrics/postingest.rs
+12 -13
View File
@@ -7,6 +7,8 @@ use mrucache::mucache::MuCache;
use netpod::ScalarType;
use netpod::Shape;
use netpod::TsNano;
use scywr::insertqueues::InsertDeques;
use scywr::insertqueues::InsertQueuesTx;
use scywr::iteminsertqueue::DataValue;
use scywr::iteminsertqueue::QueryItem;
use scywr::iteminsertqueue::ScalarValue;
@@ -44,11 +46,11 @@ pub async fn process_api_query_items(
backend: String,
item_rx: Receiver<EventValueItem>,
info_worker_tx: Sender<ChannelInfoQuery>,
iiq_tx: Sender<VecDeque<QueryItem>>,
mut iqtx: InsertQueuesTx,
) -> Result<(), Error> {
// TODO so far arbitrary upper limit on the number of ad-hoc channels:
let mut mucache: MuCache<String, SeriesWriter> = MuCache::new(2000);
let mut item_qu = VecDeque::new();
let mut iqdqs = InsertDeques::new();
let mut sw_tick_last = Instant::now();
#[allow(irrefutable_let_patterns)]
@@ -56,7 +58,7 @@ pub async fn process_api_query_items(
let tsnow = Instant::now();
if tsnow.saturating_duration_since(sw_tick_last) >= Duration::from_millis(5000) {
sw_tick_last = tsnow;
tick_writers(mucache.all_ref_mut(), &mut item_qu)?;
tick_writers(mucache.all_ref_mut(), &mut iqdqs)?;
}
let item = match item {
Ok(Ok(item)) => item,
@@ -81,26 +83,23 @@ pub async fn process_api_query_items(
stnow,
)
.await?;
let sw = &mut sw;
sw.write(item.ts, item.ts, item.val, &mut item_qu)?;
let item = core::mem::replace(&mut item_qu, VecDeque::new());
iiq_tx.send(item).await?;
sw.write(item.ts, item.ts, item.val, &mut iqdqs)?;
iqtx.send_all(&mut iqdqs).await.map_err(|_| Error::SendError)?;
}
finish_writers(mucache.all_ref_mut(), &mut item_qu)?;
finish_writers(mucache.all_ref_mut(), &mut iqdqs)?;
Ok(())
}
fn tick_writers(sws: Vec<&mut SeriesWriter>, iiq: &mut VecDeque<QueryItem>) -> Result<(), Error> {
fn tick_writers(sws: Vec<&mut SeriesWriter>, iqdqs: &mut InsertDeques) -> Result<(), Error> {
for sw in sws {
sw.tick(iiq)?;
sw.tick(iqdqs)?;
}
Ok(())
}
fn finish_writers(sws: Vec<&mut SeriesWriter>, iiq: &mut VecDeque<QueryItem>) -> Result<(), Error> {
fn finish_writers(sws: Vec<&mut SeriesWriter>, iqdqs: &mut InsertDeques) -> Result<(), Error> {
for sw in sws {
sw.tick(iiq)?;
sw.tick(iqdqs)?;
}
Ok(())
}
netfetch/src/metrics/status.rs
+28
View File
@@ -18,6 +18,8 @@ struct ChannelState {
ioc_address: Option<SocketAddr>,
connection: ConnectionState,
archiving_configuration: ChannelConfig,
recv_count: u64,
recv_bytes: u64,
}
#[derive(Debug, Serialize)]
@@ -58,6 +60,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: None,
connection: ConnectionState::Connecting,
archiving_configuration: st1.config,
recv_count: 0,
recv_bytes: 0,
};
states.channels.insert(k, chst);
}
@@ -66,6 +70,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: None,
connection: ConnectionState::Connecting,
archiving_configuration: st1.config,
recv_count: 0,
recv_bytes: 0,
};
states.channels.insert(k, chst);
}
@@ -77,6 +83,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: None,
connection: ConnectionState::Connecting,
archiving_configuration: st1.config,
recv_count: 0,
recv_bytes: 0,
};
states.channels.insert(k, chst);
}
@@ -88,6 +96,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: Some(SocketAddr::V4(addr)),
connection: ConnectionState::Connecting,
archiving_configuration: st1.config,
recv_count: 0,
recv_bytes: 0,
};
states.channels.insert(k, chst);
}
@@ -99,10 +109,14 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: Some(SocketAddr::V4(addr)),
connection: ConnectionState::Connecting,
archiving_configuration: st1.config,
recv_count: 0,
recv_bytes: 0,
};
states.channels.insert(k, chst);
}
ConnectionStateValue::ChannelStateInfo(st6) => {
let recv_count = st6.recv_count.unwrap_or(0);
let recv_bytes = st6.recv_bytes.unwrap_or(0);
use crate::ca::conn::ChannelConnectedInfo;
match st6.channel_connected_info {
ChannelConnectedInfo::Disconnected => {
@@ -112,6 +126,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
// TODO config is stored in two places
// conf: st6.conf,
archiving_configuration: st1.config,
recv_count,
recv_bytes,
};
states.channels.insert(k, chst);
}
@@ -120,6 +136,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: Some(SocketAddr::V4(addr)),
connection: ConnectionState::Connecting,
archiving_configuration: st1.config,
recv_count,
recv_bytes,
};
states.channels.insert(k, chst);
}
@@ -128,6 +146,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: Some(SocketAddr::V4(addr)),
connection: ConnectionState::Connected,
archiving_configuration: st1.config,
recv_count,
recv_bytes,
};
states.channels.insert(k, chst);
}
@@ -136,6 +156,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: Some(SocketAddr::V4(addr)),
connection: ConnectionState::Error,
archiving_configuration: st1.config,
recv_count,
recv_bytes,
};
states.channels.insert(k, chst);
}
@@ -150,6 +172,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: None,
connection: ConnectionState::Connecting,
archiving_configuration: st1.config,
recv_count: 0,
recv_bytes: 0,
};
states.channels.insert(k, chst);
}
@@ -158,6 +182,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: None,
connection: ConnectionState::Unreachable,
archiving_configuration: st1.config,
recv_count: 0,
recv_bytes: 0,
};
states.channels.insert(k, chst);
}
@@ -166,6 +192,8 @@ pub async fn channel_states(params: HashMap<String, String>, tx: Sender<CaConnSe
ioc_address: None,
connection: ConnectionState::Unreachable,
archiving_configuration: st1.config,
recv_count: 0,
recv_bytes: 0,
};
states.channels.insert(k, chst);
}