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

Group flush of insert queues

Browse Source
This commit is contained in:
Dominik Werder
2024-05-02 22:33:56 +02:00
parent 3827514900
commit a7970f2559
7 changed files with 101 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netfetch/src/ca/conn.rs
+49 -12
View File
@@ -4,7 +4,6 @@ use super::proto::ReadNotify;
use crate::ca::proto::ChannelClose;
use crate::ca::proto::EventCancel;
use crate::conf::ChannelConfig;
use crate::senderpolling::SenderPolling;
use crate::throttletrace::ThrottleTrace;
use async_channel::Receiver;
use async_channel::Sender;
@@ -31,11 +30,14 @@ use proto::CaProto;
use proto::CreateChan;
use proto::EventAdd;
use scywr::insertqueues::InsertDeques;
use scywr::insertqueues::InsertQueuesTx;
use scywr::insertqueues::InsertSenderPolling;
use scywr::iteminsertqueue as scywriiq;
use scywr::iteminsertqueue::Accounting;
use scywr::iteminsertqueue::DataValue;
use scywr::iteminsertqueue::QueryItem;
use scywr::iteminsertqueue::ShutdownReason;
use scywr::senderpolling::SenderPolling;
use scywriiq::ChannelStatus;
use scywriiq::ChannelStatusClosedReason;
use scywriiq::ChannelStatusItem;
@@ -770,7 +772,7 @@ pub struct CaConn {
ioc_ping_last: Instant,
ioc_ping_next: Instant,
ioc_ping_start: Option<Instant>,
storage_insert_sender: Pin<Box<SenderPolling<VecDeque<QueryItem>>>>,
iqsp: Pin<Box<InsertSenderPolling>>,
ca_conn_event_out_queue: VecDeque<CaConnEvent>,
ca_conn_event_out_queue_max: usize,
thr_msg_poll: ThrottleTrace,
@@ -799,7 +801,7 @@ impl CaConn {
backend: String,
remote_addr_dbg: SocketAddrV4,
local_epics_hostname: String,
storage_insert_tx: Sender<VecDeque<QueryItem>>,
iqtx: InsertQueuesTx,
channel_info_query_tx: Sender<ChannelInfoQuery>,
stats: Arc<CaConnStats>,
ca_proto_stats: Arc<CaProtoStats>,
@@ -836,7 +838,7 @@ impl CaConn {
ioc_ping_last: tsnow,
ioc_ping_next: tsnow + Self::ioc_ping_ivl_rng(&mut rng),
ioc_ping_start: None,
storage_insert_sender: Box::pin(SenderPolling::new(storage_insert_tx)),
iqsp: Box::pin(InsertSenderPolling::new(iqtx)),
ca_conn_event_out_queue: VecDeque::new(),
ca_conn_event_out_queue_max: 2000,
thr_msg_poll: ThrottleTrace::new(Duration::from_millis(10000)),
@@ -2398,15 +2400,16 @@ impl CaConn {
debug!(
"async out flushed iiq {} {} caout {}",
self.iqdqs.len() == 0,
self.storage_insert_sender.is_idle(),
self.iqsp.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()
self.iqdqs.len() == 0 && self.iqsp.is_idle() && self.ca_conn_event_out_queue.is_empty()
}
fn attempt_flush_queue<T, Q, FB, FS>(
qu: &mut VecDeque<T>,
sp: &mut Pin<Box<SenderPolling<Q>>>,
// sp: &mut Pin<Box<SenderPolling<Q>>>,
mut sp: Pin<&mut SenderPolling<Q>>,
qu_to_si: FB,
loop_max: u32,
cx: &mut Context,
@@ -2448,7 +2451,7 @@ impl CaConn {
have_progress = true;
}
Ready(Err(e)) => {
use crate::senderpolling::Error as SpErr;
use scywr::senderpolling::Error as SpErr;
match e {
SpErr::NoSendInProgress => return Err(Error::NotSending),
SpErr::Closed(_) => return Err(Error::ClosedSending),
@@ -2475,7 +2478,7 @@ macro_rules! flush_queue {
($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.$qu;
let sp = &mut obj.$sp;
let sp = obj.$sp.as_mut();
match Self::attempt_flush_queue(qu, sp, $batcher, $loop_max, $cx, $id, $stats) {
Ok(Ready(Some(()))) => {
*$have.0 |= true;
@@ -2493,7 +2496,11 @@ 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;
// let sp = std::pin::pin!(obj.iqsp.$sp);
// let sp = &mut obj.iqsp.$sp;
// let sp = std::pin::pin!(sp);
// let sp = todo!();
let sp = obj.iqsp.as_mut().$sp();
match Self::attempt_flush_queue(qu, sp, $batcher, $loop_max, $cx, $id, $stats) {
Ok(Ready(Some(()))) => {
*$have.0 |= true;
@@ -2577,11 +2584,41 @@ impl Stream for CaConn {
flush_queue_dqs!(
self,
st_rf1_rx,
storage_insert_sender,
st_rf1_sp_pin,
send_batched::<256, _>,
32,
(&mut have_progress, &mut have_pending),
"iq_st_rf1",
"st_rf1_rx",
cx,
stats_fn
);
let stats2 = self.stats.clone();
let stats_fn = move |item: &VecDeque<QueryItem>| {
stats2.iiq_batch_len().ingest(item.len() as u32);
};
flush_queue_dqs!(
self,
st_rf3_rx,
st_rf3_sp_pin,
send_batched::<256, _>,
32,
(&mut have_progress, &mut have_pending),
"st_rf3_rx",
cx,
stats_fn
);
let stats2 = self.stats.clone();
let stats_fn = move |item: &VecDeque<QueryItem>| {
stats2.iiq_batch_len().ingest(item.len() as u32);
};
flush_queue_dqs!(
self,
mt_rf3_rx,
mt_rf3_sp_pin,
send_batched::<256, _>,
32,
(&mut have_progress, &mut have_pending),
"mt_rf3_rx",
cx,
stats_fn
);
netfetch/src/ca/connset.rs
+2 -2
View File
@@ -10,7 +10,6 @@ use crate::conf::ChannelConfig;
use crate::daemon_common::Channel;
use crate::errconv::ErrConv;
use crate::rt::JoinHandle;
use crate::senderpolling::SenderPolling;
use crate::throttletrace::ThrottleTrace;
use async_channel::Receiver;
use async_channel::Sender;
@@ -38,6 +37,7 @@ use netpod::SeriesKind;
use netpod::Shape;
use scywr::iteminsertqueue::ChannelStatusItem;
use scywr::iteminsertqueue::QueryItem;
use scywr::senderpolling::SenderPolling;
use serde::Serialize;
use series::ChannelStatusSeriesId;
use serieswriter::writer::EstablishWorkerJob;
@@ -1047,7 +1047,7 @@ impl CaConnSet {
add.backend.clone(),
addr_v4,
self.local_epics_hostname.clone(),
self.iqtx.st_rf3_tx.clone(),
self.iqtx.clone2(),
self.channel_info_query_tx
.clone()
.ok_or_else(|| Error::with_msg_no_trace("no more channel_info_query_tx available"))?,
netfetch/src/lib.rs
-1
View File
@@ -7,7 +7,6 @@ pub mod metrics;
pub mod netbuf;
pub mod polltimer;
pub mod rt;
pub mod senderpolling;
#[cfg(test)]
pub mod test;
pub mod throttletrace;
netfetch/src/senderpolling.rs
-166
View File
@@ -1,166 +0,0 @@
use async_channel::Send;
use async_channel::SendError;
use async_channel::Sender;
use err::thiserror;
use futures_util::Future;
use pin_project::pin_project;
use std::marker::PhantomPinned;
use std::pin::Pin;
use std::ptr::NonNull;
use std::task::Context;
use std::task::Poll;
use thiserror::Error;
#[derive(Debug, Error)]
pub enum Error<T> {
NoSendInProgress,
Closed(T),
}
#[pin_project]
pub struct SenderPolling<T>
where
T: 'static,
{
sender: Option<Box<Sender<T>>>,
sender_ptr: NonNull<Sender<T>>,
#[pin]
fut: Option<Send<'static, T>>,
_pin: PhantomPinned,
}
fn _require_unpin<T: Unpin>(_: &T) {}
fn _check_unpin() {
let _r: &SenderPolling<String> = err::todoval();
// _require_unpin(_r);
}
unsafe impl<T> core::marker::Send for SenderPolling<T> where T: core::marker::Send {}
impl<T> SenderPolling<T> {
pub fn new(sender: Sender<T>) -> Self {
let mut ret = Self {
sender: Some(Box::new(sender)),
sender_ptr: NonNull::dangling(),
fut: None,
_pin: PhantomPinned,
};
ret.sender_ptr = NonNull::from(ret.sender.as_ref().unwrap().as_ref());
ret
}
pub fn has_sender(&self) -> bool {
self.sender.is_some()
}
pub fn is_idle(&self) -> bool {
self.sender.is_some() && self.fut.is_none()
}
pub fn is_sending(&self) -> bool {
self.fut.is_some()
}
pub fn send_pin(self: Pin<&mut Self>, item: T) {
unsafe { Pin::get_unchecked_mut(self) }.send(item)
}
pub fn send(&mut self, item: T) {
if self.sender.is_none() {
// panic!("send on dropped sender");
// TODO
return;
}
let sender = unsafe { self.sender_ptr.as_mut() };
let s = sender.send(item);
self.fut = Some(s);
}
pub fn close(&self) {
if let Some(tx) = self.sender.as_ref() {
tx.close();
}
}
pub fn drop(self: Pin<&mut Self>) {
unsafe {
let this = self.get_unchecked_mut();
this.fut = None;
this.sender = None;
}
}
pub fn len(&self) -> Option<usize> {
self.sender.as_ref().map(|x| x.len())
}
pub async fn send_async_pin(self: Pin<&mut Self>, item: T) -> Result<(), SendError<T>> {
unsafe { Pin::get_unchecked_mut(self) }.send_async(item).await
}
pub async fn send_async(&mut self, item: T) -> Result<(), SendError<T>> {
if self.is_sending() {
let fut = self.fut.take().unwrap();
if let Err(e) = fut.await {
return Err(e);
}
}
self.sender.as_ref().unwrap().send(item).await
}
unsafe fn reset_fut(futopt: Pin<&mut Option<Send<'_, T>>>) {
let y = futopt.get_unchecked_mut();
let z = y.as_mut().unwrap_unchecked();
std::ptr::drop_in_place(z);
std::ptr::write(y, None);
}
#[allow(unused)]
unsafe fn reset_fut_old(futopt: Pin<&mut Option<Send<'_, T>>>) {
*futopt.get_unchecked_mut() = None;
}
}
impl<T> Future for SenderPolling<T>
where
T: Unpin,
{
type Output = Result<(), Error<T>>;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
use Poll::*;
let mut this = self.project();
match this.fut.as_mut().as_pin_mut() {
Some(mut fut) => match fut.as_mut().poll(cx) {
Ready(Ok(())) => {
unsafe {
Self::reset_fut(this.fut);
}
Ready(Ok(()))
}
Ready(Err(e)) => {
unsafe {
Self::reset_fut(this.fut);
}
Ready(Err(Error::Closed(e.0)))
}
Pending => Pending,
},
None => Ready(Err(Error::NoSendInProgress)),
}
}
}
impl<T> Clone for SenderPolling<T> {
fn clone(&self) -> Self {
let sender = self.sender.as_ref().unwrap().as_ref().clone();
SenderPolling::new(sender)
}
}
impl<T> From<Sender<T>> for SenderPolling<T> {
fn from(value: Sender<T>) -> Self {
SenderPolling::new(value)
}
}