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

WIP parse and send 2M batched

Browse Source
This commit is contained in:
Dominik Werder
2023-11-16 17:07:13 +01:00
parent 298e9b4faa
commit e2d8f389b4
18 changed files with 576 additions and 343 deletions
Download Patch File Download Diff File Expand all files Collapse all files
netfetch/src/ca/conn.rs
+101 -48
View File
@@ -41,11 +41,15 @@ use scywriiq::QueryItem;
use serde::Serialize;
use series::ChannelStatusSeriesId;
use series::SeriesId;
use stats::rand_xoshiro::rand_core::RngCore;
use stats::rand_xoshiro::rand_core::SeedableRng;
use stats::rand_xoshiro::Xoshiro128StarStar;
use stats::CaConnStats;
use stats::CaProtoStats;
use stats::IntervalEma;
use stats::XorShift32;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::collections::VecDeque;
use std::net::SocketAddrV4;
use std::ops::ControlFlow;
@@ -149,10 +153,10 @@ fn ser_instant<S: serde::Serializer>(val: &Option<Instant>, ser: S) -> Result<S:
}
}
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
struct Cid(pub u32);
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
struct Subid(pub u32);
#[derive(Clone, Debug)]
@@ -521,11 +525,11 @@ pub struct CaConn {
proto: Option<CaProto>,
cid_store: CidStore,
subid_store: SubidStore,
channels: BTreeMap<Cid, ChannelState>,
cid_by_name: BTreeMap<String, Cid>,
cid_by_subid: BTreeMap<Subid, Cid>,
name_by_cid: BTreeMap<Cid, String>,
time_binners: BTreeMap<Cid, ConnTimeBin>,
channels: HashMap<Cid, ChannelState>,
cid_by_name: HashMap<String, Cid>,
cid_by_subid: HashMap<Subid, Cid>,
name_by_cid: HashMap<Cid, String>,
time_binners: HashMap<Cid, ConnTimeBin>,
init_state_count: u64,
insert_item_queue: VecDeque<QueryItem>,
remote_addr_dbg: SocketAddrV4,
@@ -541,14 +545,14 @@ pub struct CaConn {
ioc_ping_last: Instant,
ioc_ping_next: Instant,
ioc_ping_start: Option<Instant>,
storage_insert_sender: Pin<Box<SenderPolling<QueryItem>>>,
storage_insert_sender: Pin<Box<SenderPolling<VecDeque<QueryItem>>>>,
ca_conn_event_out_queue: VecDeque<CaConnEvent>,
channel_info_query_queue: VecDeque<ChannelInfoQuery>,
channel_info_query_sending: Pin<Box<SenderPolling<ChannelInfoQuery>>>,
thr_msg_poll: ThrottleTrace,
ca_proto_stats: Arc<CaProtoStats>,
weird_count: usize,
rng: XorShift32,
rng: Xoshiro128StarStar,
}
#[cfg(DISABLED)]
@@ -564,13 +568,13 @@ impl CaConn {
backend: String,
remote_addr_dbg: SocketAddrV4,
local_epics_hostname: String,
storage_insert_tx: Sender<QueryItem>,
storage_insert_tx: Sender<VecDeque<QueryItem>>,
channel_info_query_tx: Sender<ChannelInfoQuery>,
stats: Arc<CaConnStats>,
ca_proto_stats: Arc<CaProtoStats>,
) -> Self {
let (cq_tx, cq_rx) = async_channel::bounded(32);
let mut rng = XorShift32::new_from_time();
let mut rng = stats::xoshiro_from_time();
Self {
opts,
backend,
@@ -580,16 +584,16 @@ impl CaConn {
cid_store: CidStore::new_from_time(),
subid_store: SubidStore::new_from_time(),
init_state_count: 0,
channels: BTreeMap::new(),
cid_by_name: BTreeMap::new(),
cid_by_subid: BTreeMap::new(),
name_by_cid: BTreeMap::new(),
time_binners: BTreeMap::new(),
channels: HashMap::new(),
cid_by_name: HashMap::new(),
cid_by_subid: HashMap::new(),
name_by_cid: HashMap::new(),
time_binners: HashMap::new(),
insert_item_queue: VecDeque::new(),
remote_addr_dbg,
local_epics_hostname,
stats,
insert_ivl_min_mus: 1000 * 6,
insert_ivl_min_mus: 1000 * 4,
conn_command_tx: Box::pin(cq_tx),
conn_command_rx: Box::pin(cq_rx),
conn_backoff: 0.02,
@@ -610,8 +614,8 @@ impl CaConn {
}
}
fn ioc_ping_ivl_rng(rng: &mut XorShift32) -> Duration {
IOC_PING_IVL * 100 / (70 + (rng.next() % 60))
fn ioc_ping_ivl_rng(rng: &mut Xoshiro128StarStar) -> Duration {
IOC_PING_IVL * 100 / (70 + (rng.next_u32() % 60))
}
fn new_self_ticker() -> Pin<Box<tokio::time::Sleep>> {
@@ -813,7 +817,7 @@ impl CaConn {
fn handle_conn_command(&mut self, cx: &mut Context) -> Result<Poll<Option<()>>, Error> {
// TODO if this loops for too long time, yield and make sure we get wake up again.
use Poll::*;
self.stats.caconn_loop3_count.inc();
self.stats.loop3_count.inc();
if self.is_shutdown() {
Ok(Ready(None))
} else {
@@ -896,11 +900,11 @@ impl CaConn {
fn channel_remove_expl(
name: String,
channels: &mut BTreeMap<Cid, ChannelState>,
cid_by_name: &mut BTreeMap<String, Cid>,
name_by_cid: &mut BTreeMap<Cid, String>,
channels: &mut HashMap<Cid, ChannelState>,
cid_by_name: &mut HashMap<String, Cid>,
name_by_cid: &mut HashMap<Cid, String>,
cid_store: &mut CidStore,
time_binners: &mut BTreeMap<Cid, ConnTimeBin>,
time_binners: &mut HashMap<Cid, ConnTimeBin>,
) {
let cid = Self::cid_by_name_expl(&name, cid_by_name, name_by_cid, cid_store);
if channels.contains_key(&cid) {
@@ -924,8 +928,8 @@ impl CaConn {
fn cid_by_name_expl(
name: &str,
cid_by_name: &mut BTreeMap<String, Cid>,
name_by_cid: &mut BTreeMap<Cid, String>,
cid_by_name: &mut HashMap<String, Cid>,
name_by_cid: &mut HashMap<Cid, String>,
cid_store: &mut CidStore,
) -> Cid {
if let Some(cid) = cid_by_name.get(name) {
@@ -1214,9 +1218,7 @@ impl CaConn {
let ema = em.ema();
let ivl_min = (insert_ivl_min_mus as f32) * 1e-6;
let dt = (ivl_min - ema).max(0.) / em.k();
st.insert_next_earliest = tsnow
.checked_add(Duration::from_micros((dt * 1e6) as u64))
.ok_or_else(|| Error::with_msg_no_trace("time overflow in next insert"))?;
st.insert_next_earliest = tsnow + Duration::from_micros((dt * 1e6) as u64);
let ts_msp_last = st.ts_msp_last;
// TODO get event timestamp from channel access field
let ts_msp_grid = (ts / TS_MSP_GRID_UNIT / TS_MSP_GRID_SPACING * TS_MSP_GRID_SPACING) as u32;
@@ -1555,7 +1557,7 @@ impl CaConn {
}
CaMsgTy::EventAddRes(k) => {
trace4!("got EventAddRes: {k:?}");
self.stats.caconn_recv_data.inc();
self.stats.event_add_res_recv.inc();
let res = Self::handle_event_add_res(self, k, tsnow);
let ts2 = Instant::now();
self.stats
@@ -1866,7 +1868,7 @@ impl CaConn {
let tsnow = Instant::now();
let mut have_progress = false;
for _ in 0..64 {
self.stats.caconn_loop2_count.inc();
self.stats.loop2_count.inc();
if self.is_shutdown() {
break;
} else if self.insert_item_queue.len() >= self.opts.insert_queue_max {
@@ -1963,21 +1965,50 @@ impl CaConn {
fn attempt_flush_storage_queue(mut self: Pin<&mut Self>, cx: &mut Context) -> Result<Poll<Option<()>>, Error> {
use Poll::*;
let (qu, sd, stats) = Self::storage_queue_vars(&mut self);
{
let n = qu.len();
if n >= 128 {
stats.storage_queue_above_128().inc();
} else if n >= 32 {
stats.storage_queue_above_32().inc();
} else if n >= 8 {
stats.storage_queue_above_8().inc();
}
}
let mut have_progress = false;
for _ in 0..128 {
let sd = &mut self.storage_insert_sender;
let mut i = 0;
loop {
i += 1;
if i > 120 {
break;
}
if !sd.has_sender() {
return Err(Error::with_msg_no_trace("attempt_flush_storage_queue no more sender"));
}
if sd.is_idle() {
if let Some(item) = self.insert_item_queue.pop_front() {
self.storage_insert_sender.as_mut().send_pin(item);
if qu.len() != 0 {
let item: VecDeque<_> = qu.drain(..).collect();
stats.storage_queue_send().add(item.len() as _);
sd.as_mut().send_pin(item);
} else {
break;
}
}
if self.storage_insert_sender.is_sending() {
match self.storage_insert_sender.poll_unpin(cx) {
if sd.is_sending() {
match sd.poll_unpin(cx) {
Ready(Ok(())) => {
have_progress = true;
}
Ready(Err(_)) => return Err(Error::with_msg_no_trace("can not send into channel")),
Pending => return Ok(Pending),
Ready(Err(_)) => {
return Err(Error::with_msg_no_trace(
"attempt_flush_storage_queue can not send into channel",
));
}
Pending => {
stats.storage_queue_pending().inc();
return Ok(Pending);
}
}
}
}
@@ -1988,12 +2019,32 @@ impl CaConn {
}
}
// TODO refactor, put together in separate type:
fn storage_queue_vars(
this: &mut CaConn,
) -> (
&mut VecDeque<QueryItem>,
&mut Pin<Box<SenderPolling<VecDeque<QueryItem>>>>,
&CaConnStats,
) {
(
&mut this.insert_item_queue,
&mut this.storage_insert_sender,
&this.stats,
)
}
fn attempt_flush_channel_info_query(mut self: Pin<&mut Self>, cx: &mut Context) -> Result<Poll<Option<()>>, Error> {
use Poll::*;
if self.is_shutdown() {
Ok(Ready(None))
} else {
let sd = self.channel_info_query_sending.as_mut();
if !sd.has_sender() {
return Err(Error::with_msg_no_trace(
"attempt_flush_channel_info_query no more sender",
));
}
if sd.is_idle() {
if let Some(item) = self.channel_info_query_queue.pop_front() {
trace3!("send series query {item:?}");
@@ -2005,7 +2056,9 @@ impl CaConn {
if sd.is_sending() {
match sd.poll_unpin(cx) {
Ready(Ok(())) => Ok(Ready(Some(()))),
Ready(Err(_)) => Err(Error::with_msg_no_trace("can not send into channel")),
Ready(Err(_)) => Err(Error::with_msg_no_trace(
"attempt_flush_channel_info_query can not send into channel",
)),
Pending => Ok(Pending),
}
} else {
@@ -2020,11 +2073,11 @@ impl Stream for CaConn {
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
self.stats.caconn_poll_count.inc();
self.stats.poll_count().inc();
let poll_ts1 = Instant::now();
self.stats.ca_conn_poll_fn_begin().inc();
self.stats.poll_fn_begin().inc();
let ret = loop {
self.stats.ca_conn_poll_loop_begin().inc();
self.stats.poll_loop_begin().inc();
let qlen = self.insert_item_queue.len();
if qlen >= self.opts.insert_queue_max * 2 / 3 {
self.stats.insert_item_queue_pressure().inc();
@@ -2108,10 +2161,10 @@ impl Stream for CaConn {
} else {
// debug!("queues_out_flushed false");
if have_progress {
self.stats.ca_conn_poll_reloop().inc();
self.stats.poll_reloop().inc();
continue;
} else if have_pending {
self.stats.ca_conn_poll_pending().inc();
self.stats.poll_pending().inc();
Pending
} else {
// TODO error
@@ -2123,13 +2176,13 @@ impl Stream for CaConn {
}
} else {
if have_progress {
self.stats.ca_conn_poll_reloop().inc();
self.stats.poll_reloop().inc();
continue;
} else if have_pending {
self.stats.ca_conn_poll_pending().inc();
self.stats.poll_pending().inc();
Pending
} else {
self.stats.ca_conn_poll_no_progress_no_pending().inc();
self.stats.poll_no_progress_no_pending().inc();
let e = Error::with_msg_no_trace("no progress no pending");
Ready(Some(Err(e)))
}
netfetch/src/ca/connset.rs
+64 -53
View File
@@ -261,7 +261,13 @@ impl CaConnSetCtrl {
pub async fn check_health(&self) -> Result<(), Error> {
let cmd = ConnSetCmd::CheckHealth(Instant::now());
let n = self.tx.len();
if n > 0 {
debug!("check_health self.tx.len() {:?}", n);
}
let s = format!("{:?}", cmd);
self.tx.send(CaConnSetEvent::ConnSetCmd(cmd)).await?;
debug!("check_health enqueued {s}");
Ok(())
}
@@ -345,9 +351,9 @@ pub struct CaConnSet {
find_ioc_query_queue: VecDeque<IocAddrQuery>,
find_ioc_query_sender: Pin<Box<SenderPolling<IocAddrQuery>>>,
find_ioc_res_rx: Pin<Box<Receiver<VecDeque<FindIocRes>>>>,
storage_insert_tx: Pin<Box<Sender<QueryItem>>>,
storage_insert_queue: VecDeque<QueryItem>,
storage_insert_sender: Pin<Box<SenderPolling<QueryItem>>>,
storage_insert_tx: Pin<Box<Sender<VecDeque<QueryItem>>>>,
storage_insert_queue: VecDeque<VecDeque<QueryItem>>,
storage_insert_sender: Pin<Box<SenderPolling<VecDeque<QueryItem>>>>,
ca_conn_res_tx: Pin<Box<Sender<(SocketAddr, CaConnEvent)>>>,
ca_conn_res_rx: Pin<Box<Receiver<(SocketAddr, CaConnEvent)>>>,
connset_out_queue: VecDeque<CaConnSetItem>,
@@ -361,7 +367,6 @@ pub struct CaConnSet {
await_ca_conn_jhs: VecDeque<(SocketAddr, JoinHandle<Result<(), Error>>)>,
thr_msg_poll_1: ThrottleTrace,
thr_msg_storage_len: ThrottleTrace,
did_connset_out_queue: bool,
ca_proto_stats: Arc<CaProtoStats>,
rogue_channel_count: u64,
connect_fail_count: usize,
@@ -371,7 +376,7 @@ impl CaConnSet {
pub fn start(
backend: String,
local_epics_hostname: String,
storage_insert_tx: Sender<QueryItem>,
storage_insert_tx: Sender<VecDeque<QueryItem>>,
channel_info_query_tx: Sender<ChannelInfoQuery>,
ingest_opts: CaIngestOpts,
) -> CaConnSetCtrl {
@@ -422,7 +427,6 @@ impl CaConnSet {
await_ca_conn_jhs: VecDeque::new(),
thr_msg_poll_1: ThrottleTrace::new(Duration::from_millis(2000)),
thr_msg_storage_len: ThrottleTrace::new(Duration::from_millis(1000)),
did_connset_out_queue: false,
ca_proto_stats: ca_proto_stats.clone(),
rogue_channel_count: 0,
connect_fail_count: 0,
@@ -491,7 +495,8 @@ impl CaConnSet {
CaConnEventValue::EchoTimeout => Ok(()),
CaConnEventValue::ConnCommandResult(x) => self.handle_conn_command_result(addr, x),
CaConnEventValue::QueryItem(item) => {
self.storage_insert_queue.push_back(item);
todo!("remove this insert case");
// self.storage_insert_queue.push_back(item);
Ok(())
}
CaConnEventValue::ChannelCreateFail(x) => self.handle_channel_create_fail(addr, x),
@@ -743,7 +748,7 @@ impl CaConnSet {
}
fn handle_check_health(&mut self, ts1: Instant) -> Result<(), Error> {
debug!("handle_check_health");
trace2!("handle_check_health");
if self.shutdown_stopping {
return Ok(());
}
@@ -754,15 +759,11 @@ impl CaConnSet {
self.check_channel_states()?;
// Trigger already the next health check, but use the current data that we have.
// TODO try to deliver a command to CaConn
// Add some queue for commands to CaConn to the ress.
// Fail here if that queue gets too long.
// Try to push the commands periodically.
// TODO do the full check before sending the reply to daemon.
for (_, res) in self.ca_conn_ress.iter_mut() {
let item = ConnCommand::check_health();
res.cmd_queue.push_back(item);
debug!(
trace2!(
"handle_check_health pushed check command {:?} {:?}",
res.cmd_queue.len(),
res.sender.len()
@@ -822,7 +823,7 @@ impl CaConnSet {
}
fn apply_ca_conn_health_update(&mut self, addr: SocketAddr, res: CheckHealthResult) -> Result<(), Error> {
debug!("apply_ca_conn_health_update {addr}");
trace2!("apply_ca_conn_health_update {addr}");
let tsnow = SystemTime::now();
self.rogue_channel_count = 0;
for (k, v) in res.channel_statuses {
@@ -993,7 +994,7 @@ impl CaConnSet {
async fn ca_conn_item_merge(
conn: CaConn,
tx1: Sender<(SocketAddr, CaConnEvent)>,
tx2: Sender<QueryItem>,
tx2: Sender<VecDeque<QueryItem>>,
addr: SocketAddr,
stats: Arc<CaConnSetStats>,
) -> Result<(), Error> {
@@ -1005,10 +1006,13 @@ impl CaConnSet {
while let Some(item) = conn.next().await {
match item {
Ok(item) => {
connstats.conn_item_count.inc();
connstats.item_count.inc();
match item.value {
CaConnEventValue::QueryItem(x) => {
if let Err(_) = tx2.send(x).await {
warn!("ca_conn_item_merge should not go here often");
let mut v = VecDeque::new();
v.push_back(x);
if let Err(_) = tx2.send(v).await {
break;
}
}
@@ -1076,7 +1080,9 @@ impl CaConnSet {
};
}
let item = QueryItem::ChannelStatus(item);
self.storage_insert_queue.push_back(item);
let mut v = VecDeque::new();
v.push_back(item);
self.storage_insert_queue.push_back(v);
Ok(())
}
@@ -1423,35 +1429,37 @@ impl CaConnSet {
(search_pending, assigned_without_health_update)
}
fn try_push_ca_conn_cmds(&mut self, cx: &mut Context) {
fn try_push_ca_conn_cmds(&mut self, cx: &mut Context) -> Result<(), Error> {
use Poll::*;
for (_, v) in self.ca_conn_ress.iter_mut() {
'level2: loop {
let tx = &mut v.sender;
if v.cmd_queue.len() != 0 || tx.is_sending() {
debug!("try_push_ca_conn_cmds {:?} {:?}", v.cmd_queue.len(), tx.len());
let tx = &mut v.sender;
loop {
if false {
if v.cmd_queue.len() != 0 || tx.is_sending() {
debug!("try_push_ca_conn_cmds {:?} {:?}", v.cmd_queue.len(), tx.len());
}
}
loop {
break if tx.is_sending() {
match tx.poll_unpin(cx) {
Ready(Ok(())) => {
self.stats.try_push_ca_conn_cmds_sent.inc();
continue;
}
Ready(Err(e)) => {
error!("try_push_ca_conn_cmds {e}");
}
Pending => {
break 'level2;
}
break if tx.is_sending() {
match tx.poll_unpin(cx) {
Ready(Ok(())) => {
self.stats.try_push_ca_conn_cmds_sent.inc();
continue;
}
} else if let Some(item) = v.cmd_queue.pop_front() {
tx.as_mut().send_pin(item);
continue;
};
}
Ready(Err(e)) => {
error!("try_push_ca_conn_cmds {e}");
return Err(Error::with_msg_no_trace(format!("{e}")));
}
Pending => (),
}
} else if let Some(item) = v.cmd_queue.pop_front() {
tx.as_mut().send_pin(item);
continue;
} else {
()
};
}
}
Ok(())
}
}
@@ -1459,9 +1467,11 @@ impl Stream for CaConnSet {
type Item = CaConnSetItem;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
trace4!("CaConnSet poll begin");
use Poll::*;
self.stats.poll_fn_begin().inc();
loop {
let res = loop {
trace4!("CaConnSet poll loop");
self.stats.poll_loop_begin().inc();
self.stats.storage_insert_tx_len.set(self.storage_insert_tx.len() as _);
@@ -1485,15 +1495,12 @@ impl Stream for CaConnSet {
let mut have_pending = false;
let mut have_progress = false;
self.try_push_ca_conn_cmds(cx);
if let Err(e) = self.try_push_ca_conn_cmds(cx) {
break Ready(Some(CaConnSetItem::Error(e)));
}
if self.did_connset_out_queue {
self.did_connset_out_queue = false;
} else {
if let Some(item) = self.connset_out_queue.pop_front() {
self.did_connset_out_queue = true;
break Ready(Some(item));
}
if let Some(item) = self.connset_out_queue.pop_front() {
break Ready(Some(item));
}
if let Some((addr, jh)) = self.await_ca_conn_jhs.front_mut() {
@@ -1634,7 +1641,9 @@ impl Stream for CaConnSet {
}
Err(e) => break Ready(Some(CaConnSetItem::Error(e))),
},
Ready(None) => {}
Ready(None) => {
warn!("connset_inp_rx broken?")
}
Pending => {
have_pending = true;
}
@@ -1663,6 +1672,8 @@ impl Stream for CaConnSet {
}
}
};
}
};
trace4!("CaConnSet poll done");
res
}
}
netfetch/src/ca/proto.rs
+164 -146
View File
@@ -1,13 +1,12 @@
use crate::netbuf;
use err::thiserror;
use err::ThisError;
use futures_util::pin_mut;
use futures_util::Stream;
use log::*;
use netpod::timeunits::*;
use slidebuf::SlideBuf;
use stats::CaProtoStats;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::collections::VecDeque;
use std::io;
use std::net::SocketAddrV4;
@@ -46,6 +45,8 @@ pub enum Error {
ParseAttemptInDoneState,
UnexpectedHeader,
ExtendedHeaderBadCount,
NoReadBufferSpace,
NeitherPendingNorProgress,
}
const CA_PROTO_VERSION: u16 = 13;
@@ -1016,8 +1017,9 @@ pub struct CaProto {
outbuf: SlideBuf,
out: VecDeque<CaMsg>,
array_truncate: usize,
logged_proto_error_for_cid: BTreeMap<u32, bool>,
logged_proto_error_for_cid: HashMap<u32, bool>,
stats: Arc<CaProtoStats>,
resqu: VecDeque<CaItem>,
}
impl CaProto {
@@ -1026,12 +1028,13 @@ impl CaProto {
tcp,
remote_addr_dbg,
state: CaState::StdHead,
buf: SlideBuf::new(1024 * 1024 * 4),
buf: SlideBuf::new(1024 * 1024 * 8),
outbuf: SlideBuf::new(1024 * 128),
out: VecDeque::new(),
array_truncate,
logged_proto_error_for_cid: BTreeMap::new(),
logged_proto_error_for_cid: HashMap::new(),
stats,
resqu: VecDeque::with_capacity(256),
}
}
@@ -1063,14 +1066,14 @@ impl CaProto {
}
}
fn attempt_output(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Error>> {
fn attempt_output(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<usize, Error>> {
use Poll::*;
let (w, b) = self.outbuf_conn();
pin_mut!(w);
let w = Pin::new(w);
match w.poll_write(cx, b) {
Ready(k) => match k {
Ok(k) => match self.outbuf.adv(k) {
Ok(()) => Ready(Ok(())),
Ok(()) => Ready(Ok(k)),
Err(e) => {
error!("advance error {:?}", e);
Ready(Err(e.into()))
@@ -1085,13 +1088,12 @@ impl CaProto {
}
}
fn loop_body(mut self: Pin<&mut Self>, cx: &mut Context) -> Result<Option<Poll<CaItem>>, Error> {
fn loop_body(mut self: Pin<&mut Self>, cx: &mut Context) -> Result<Poll<()>, Error> {
use Poll::*;
let mut have_pending = false;
let mut have_progress = false;
let tsnow = Instant::now();
let output_res_1: Option<Poll<()>> = 'll1: loop {
if self.out.len() == 0 {
break None;
}
'l1: while self.out.len() != 0 {
while let Some((msg, buf)) = self.out_msg_buf() {
let msglen = msg.len();
if msglen > buf.len() {
@@ -1103,168 +1105,184 @@ impl CaProto {
self.out.pop_front();
}
}
while self.outbuf.len() > 0 {
while self.outbuf.len() != 0 {
match Self::attempt_output(self.as_mut(), cx)? {
Ready(()) => {}
Ready(n) => {
if n != 0 {
have_progress = true;
} else {
// Should not occur to begin with. TODO restructure.
break 'l1;
}
}
Pending => {
break 'll1 Some(Pending);
have_pending = true;
break 'l1;
}
}
}
};
let output_res_2: Option<Poll<()>> = if let Some(Pending) = output_res_1 {
Some(Pending)
} else {
loop {
if self.outbuf.len() == 0 {
break None;
}
'l1: while self.outbuf.len() != 0 {
match Self::attempt_output(self.as_mut(), cx)? {
Ready(n) => {
if n != 0 {
have_progress = true;
} else {
// Should not occur to begin with. TODO restructure.
break 'l1;
}
}
match Self::attempt_output(self.as_mut(), cx)? {
Ready(()) => {}
Pending => break Some(Pending),
Pending => {
have_pending = true;
break 'l1;
}
}
};
}
let need_min = self.state.need_min();
let read_res = {
if self.buf.cap() < need_min {
self.state = CaState::Done;
let e = Error::BufferTooSmallForNeedMin(self.buf.cap(), self.state.need_min());
Err(e)
} else if self.buf.len() < need_min {
let (w, mut rbuf) = self.inpbuf_conn(need_min)?;
pin_mut!(w);
match w.poll_read(cx, &mut rbuf) {
Ready(k) => match k {
Ok(()) => {
let nf = rbuf.filled().len();
if nf == 0 {
info!(
"EOF peer {:?} {:?} {:?}",
self.tcp.peer_addr(),
self.remote_addr_dbg,
self.state
);
// TODO may need another state, if not yet done when input is EOF.
self.state = CaState::Done;
Ok(Some(Ready(CaItem::empty())))
} else {
if false {
info!("received {} bytes", rbuf.filled().len());
let t = rbuf.filled().len().min(32);
info!("received data {:?}", &rbuf.filled()[0..t]);
if self.buf.cap() < need_min {
self.state = CaState::Done;
let e = Error::BufferTooSmallForNeedMin(self.buf.cap(), self.state.need_min());
return Err(e);
}
if self.buf.len() < need_min {
let (w, mut rbuf) = self.inpbuf_conn(need_min)?;
if rbuf.remaining() == 0 {
return Err(Error::NoReadBufferSpace);
}
let w = Pin::new(w);
match w.poll_read(cx, &mut rbuf) {
Ready(k) => match k {
Ok(()) => {
let nf = rbuf.filled().len();
if nf == 0 {
info!(
"EOF peer {:?} {:?} {:?}",
self.tcp.peer_addr(),
self.remote_addr_dbg,
self.state
);
// TODO may need another state, if not yet done when input is EOF.
self.state = CaState::Done;
} else {
if false {
info!("received {} bytes", rbuf.filled().len());
let t = rbuf.filled().len().min(32);
info!("received data {:?}", &rbuf.filled()[0..t]);
}
match self.buf.wadv(nf) {
Ok(()) => {
have_progress = true;
self.stats.tcp_recv_bytes().add(nf as _);
self.stats.tcp_recv_count().inc();
}
match self.buf.wadv(nf) {
Ok(()) => {
self.stats.tcp_recv_bytes().add(nf as _);
self.stats.tcp_recv_count().inc();
Ok(Some(Ready(CaItem::empty())))
}
Err(e) => {
error!("netbuf wadv fail nf {nf}");
Err(e.into())
}
Err(e) => {
error!("netbuf wadv fail nf {nf}");
return Err(e.into());
}
}
}
Err(e) => Err(e.into()),
},
Pending => Ok(Some(Pending)),
}
Err(e) => {
return Err(e.into());
}
},
Pending => {
have_pending = true;
}
}
}
while self.resqu.len() < self.resqu.capacity() {
if let Some(item) = self.parse_item(tsnow)? {
have_progress = true;
self.resqu.push_back(item);
} else {
Ok(None)
break;
}
}?;
let parse_res: Option<CaItem> = self.parse_item(tsnow)?;
match (output_res_2, read_res, parse_res) {
(_, _, Some(item)) => Ok(Some(Ready(item))),
(Some(Pending), _, _) => Ok(Some(Pending)),
(_, Some(Pending), _) => Ok(Some(Pending)),
(_, None, None) => {
// TODO constrain how often we can go to this case consecutively.
Ok(None)
}
(_, Some(_), None) => Ok(None),
}
if have_progress {
Ok(Ready(()))
} else if have_pending {
Ok(Pending)
} else {
Err(Error::NeitherPendingNorProgress)
}
}
fn parse_item(&mut self, tsnow: Instant) -> Result<Option<CaItem>, Error> {
loop {
if self.buf.len() < self.state.need_min() {
break Ok(None);
}
break match &self.state {
CaState::StdHead => {
let hi = HeadInfo::from_netbuf(&mut self.buf)?;
if hi.cmdid == 1 || hi.cmdid == 15 {
let sid = hi.param1;
if hi.payload_size == 0xffff {
if hi.data_count != 0 {
warn!("protocol error: {hi:?}");
return Err(Error::ExtendedHeaderBadCount);
}
if self.buf.len() < self.state.need_min() {
return Ok(None);
}
match &self.state {
CaState::StdHead => {
let hi = HeadInfo::from_netbuf(&mut self.buf)?;
if hi.cmdid == 1 || hi.cmdid == 15 {
let sid = hi.param1;
if hi.payload_size == 0xffff {
if hi.data_count != 0 {
warn!("protocol error: {hi:?}");
return Err(Error::ExtendedHeaderBadCount);
}
if hi.payload_size == 0xffff {
} else if hi.payload_size > 16368 {
self.stats.payload_std_too_large().inc();
}
}
if hi.cmdid > 26 {
// TODO count as logic error
self.stats.protocol_issue().inc();
}
if hi.payload_size == 0xffff {
self.state = CaState::ExtHead(hi);
Ok(None)
} else {
// For extended messages, ingest on receive of extended header
self.stats.payload_size().ingest(hi.payload_len() as u32);
if hi.payload_size == 0 {
self.state = CaState::StdHead;
let msg = CaMsg::from_proto_infos(&hi, &[], tsnow, self.array_truncate)?;
Ok(Some(CaItem::Msg(msg)))
} else {
self.state = CaState::Payload(hi);
Ok(None)
}
} else if hi.payload_size > 16368 {
self.stats.payload_std_too_large().inc();
}
}
CaState::ExtHead(hi) => {
let payload_size = self.buf.read_u32_be()?;
let data_count = self.buf.read_u32_be()?;
if hi.cmdid > 26 {
// TODO count as logic error
self.stats.protocol_issue().inc();
}
if hi.payload_size == 0xffff {
self.state = CaState::ExtHead(hi);
Ok(None)
} else {
// For extended messages, ingest on receive of extended header
self.stats.payload_size().ingest(hi.payload_len() as u32);
if payload_size > 1024 * 1024 * 32 {
self.stats.payload_ext_very_large().inc();
if false {
warn!(
"ExtHead data_type {} payload_size {payload_size} data_count {data_count}",
hi.data_type
);
}
if hi.payload_size == 0 {
self.state = CaState::StdHead;
let msg = CaMsg::from_proto_infos(&hi, &[], tsnow, self.array_truncate)?;
Ok(Some(CaItem::Msg(msg)))
} else {
self.state = CaState::Payload(hi);
Ok(None)
}
if payload_size <= 16368 {
self.stats.payload_ext_but_small().inc();
}
}
CaState::ExtHead(hi) => {
let payload_size = self.buf.read_u32_be()?;
let data_count = self.buf.read_u32_be()?;
self.stats.payload_size().ingest(hi.payload_len() as u32);
if payload_size > 1024 * 1024 * 32 {
self.stats.payload_ext_very_large().inc();
if false {
warn!(
"ExtHead data_type {} payload_size {payload_size} data_count {data_count}",
hi.data_type
);
}
let hi = hi.clone().with_ext(payload_size, data_count);
self.state = CaState::Payload(hi);
Ok(None)
}
CaState::Payload(hi) => {
let g = self.buf.read_bytes(hi.payload_len())?;
let msg = CaMsg::from_proto_infos(hi, g, tsnow, self.array_truncate)?;
// data-count is only reasonable for event messages
if let CaMsgTy::EventAddRes(e) = &msg.ty {
self.stats.data_count().ingest(hi.data_count() as u32);
}
self.state = CaState::StdHead;
Ok(Some(CaItem::Msg(msg)))
if payload_size <= 16368 {
self.stats.payload_ext_but_small().inc();
warn!(
"ExtHead data_type {} payload_size {payload_size} data_count {data_count}",
hi.data_type
);
}
CaState::Done => Err(Error::ParseAttemptInDoneState),
};
let hi = hi.clone().with_ext(payload_size, data_count);
self.state = CaState::Payload(hi);
Ok(None)
}
CaState::Payload(hi) => {
let g = self.buf.read_bytes(hi.payload_len())?;
let msg = CaMsg::from_proto_infos(hi, g, tsnow, self.array_truncate)?;
// data-count is only reasonable for event messages
if let CaMsgTy::EventAddRes(e) = &msg.ty {
self.stats.data_count().ingest(hi.data_count() as u32);
}
self.state = CaState::StdHead;
Ok(Some(CaItem::Msg(msg)))
}
CaState::Done => Err(Error::ParseAttemptInDoneState),
}
}
}
@@ -1274,16 +1292,16 @@ impl Stream for CaProto {
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
loop {
break if let CaState::Done = self.state {
break if let Some(item) = self.resqu.pop_front() {
Ready(Some(Ok(item)))
} else if let CaState::Done = self.state {
Ready(None)
} else {
let k = Self::loop_body(self.as_mut(), cx);
match k {
Ok(Some(Ready(k))) => Ready(Some(Ok(k))),
Ok(Some(Pending)) => Pending,
Ok(None) => continue,
Ok(Ready(())) => continue,
Ok(Pending) => Pending,
Err(e) => Ready(Some(Err(e))),
}
};
netfetch/src/conf.rs
+6 -1
View File
@@ -28,6 +28,7 @@ pub struct CaIngestOpts {
scylla: ScyllaConfig,
array_truncate: Option<u64>,
insert_worker_count: Option<usize>,
insert_worker_concurrency: Option<usize>,
insert_scylla_sessions: Option<usize>,
insert_queue_max: Option<usize>,
insert_item_queue_cap: Option<usize>,
@@ -76,7 +77,11 @@ impl CaIngestOpts {
}
pub fn insert_worker_count(&self) -> usize {
self.insert_worker_count.unwrap_or(800)
self.insert_worker_count.unwrap_or(4)
}
pub fn insert_worker_concurrency(&self) -> usize {
self.insert_worker_concurrency.unwrap_or(32)
}
pub fn insert_scylla_sessions(&self) -> usize {
netfetch/src/metrics.rs
+9
View File
@@ -223,6 +223,15 @@ fn make_routes(dcom: Arc<DaemonComm>, connset_cmd_tx: Sender<CaConnSetEvent>, st
|| async move {
axum::Json(serde_json::json!({
"v1": 42_u32,
"o1": {
"v2": 56,
"o2": {
"v3": "test",
},
},
"o5": {
"v6": 89,
},
}))
}
}),
netfetch/src/senderpolling.rs
+19 -3
View File
@@ -43,6 +43,10 @@ impl<T> SenderPolling<T> {
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()
}
@@ -97,6 +101,18 @@ impl<T> SenderPolling<T> {
}
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>
@@ -109,16 +125,16 @@ where
use Poll::*;
let mut this = self.project();
match this.fut.as_mut().as_pin_mut() {
Some(fut) => match fut.poll(cx) {
Some(mut fut) => match fut.as_mut().poll(cx) {
Ready(Ok(())) => {
unsafe {
*this.fut.get_unchecked_mut() = None;
Self::reset_fut(this.fut);
}
Ready(Ok(()))
}
Ready(Err(e)) => {
unsafe {
*this.fut.get_unchecked_mut() = None;
Self::reset_fut(this.fut);
}
Ready(Err(Error::Closed(e.0)))
}