daqbuf/daqbuffer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP best config select

Browse Source
This commit is contained in:
Dominik Werder
2023-06-14 16:35:13 +02:00
parent c60243c646
commit f3bb7c4d6f
3 changed files with 19 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
dbconn/src/scan.rs
View File

@@ -205,16 +205,16 @@ async fn update_db_with_channel_names_inner(
info!("update_db_with_channel_names connection done");
let node_disk_ident = get_node_disk_ident(&node_config, &dbc).await?;
info!("update_db_with_channel_names get_node_disk_ident done");
{
let sql = concat!(
"insert into channels (facility, name) select facility, name from (values ($1::bigint, $2::text)) v1 (facility, name)",
" where not exists (select 1 from channels t1 where t1.facility = v1.facility and t1.name = v1.name)",
" on conflict do nothing",
);
let insert_sql = concat!(
"insert into channels (facility, name) select facility, name from (values ($1::bigint, $2::text)) v1 (facility, name)",
" where not exists (select 1 from channels t1 where t1.facility = v1.facility and t1.name = v1.name)",
" on conflict do nothing",
);
if false {
let fac: i64 = 1;
let ch = format!("tmp_dummy_04");
let ret = dbc
.query(sql, &[&fac, &ch])
.query(insert_sql, &[&fac, &ch])
.await
.err_conv()
.map_err(|e| format!("in channel name insert: {e}"));
@@ -234,13 +234,8 @@ async fn update_db_with_channel_names_inner(
for ch in channel_names {
let fac = node_disk_ident.facility;
crate::delay_io_short().await;
let sql = concat!(
"insert into channels (facility, name) select facility, name from (values ($1::bigint, $2::text)) v1 (facility, name)",
" where not exists (select 1 from channels t1 where t1.facility = v1.facility and t1.name = v1.name)",
" on conflict do nothing",
);
let ret = dbc
.query(sql, &[&fac, &ch])
.query(insert_sql, &[&fac, &ch])
.await
.err_conv()
.map_err(|e| format!("in channel name insert: {e}"));

7
disk/src/channelconfig.rs
View File

@@ -34,12 +34,7 @@ pub async fn channel_config_best_match(
node_config: &NodeConfigCached,
) -> Result<Option<SfDbChConf>, Error> {
let best = config_entry_best_match(&range, channel.clone(), node_config).await?;
let channel_configs = read_local_config(channel.clone(), node_config.clone()).await?;
let entry_res = match extract_matching_config_entry(&range, &channel_configs) {
Ok(k) => k,
Err(e) => return Err(e),
};
match entry_res.best() {
match best {
None => Ok(None),
Some(entry) => {
let shape = match entry.to_shape() {

14
parse/src/channelconfig.rs
View File

@@ -465,6 +465,7 @@ pub fn extract_matching_config_entry<'a>(
.map({
let mut nx = None;
move |(i, x)| {
debug!("MAP ENTRY {i:3} {:?}", x.ts);
let k = nx.clone();
nx = Some(x.ts.clone());
(i, x, k)
@@ -474,11 +475,14 @@ pub fn extract_matching_config_entry<'a>(
let mut c: Vec<_> = b
.into_iter()
.rev()
.map(|(i, e, tsn)| {
if let Some(ts2) = tsn {
//debug!("LOOK AT CONFIG ENTRY {:3} {:?} {:?}", i, e.ts, tsn);
let ret = if let Some(ts2) = tsn.clone() {
if e.ts.ns() < range.end {
let p = if e.ts.ns() < range.beg { range.beg } else { e.ts.ns() };
let q = if ts2.ns() < range.end { ts2.ns() } else { range.end };
//debug!("P/Q {:?} {:?}", TsNano(q - p), TsNano(p - q));
(i, TsNano(q - p), e)
} else {
(i, TsNano(0), e)
@@ -493,14 +497,16 @@ pub fn extract_matching_config_entry<'a>(
} else {
(i, TsNano(0), e)
}
}
};
debug!("LOOK AT CONFIG ENTRY {:3} {:?} {:?} {:?}", i, e.ts, tsn, ret.1);
ret
})
.collect();
c.sort_unstable_by_key(|x| u64::MAX - x.1.ns());
for (i, ts, e) in &c[..c.len().min(3)] {
info!("FOUND CONFIG IN ORDER {} {:?} {:?}", i, ts, e.ts);
for (i, dt, e) in &c[..c.len().min(3)] {
debug!("FOUND CONFIG IN ORDER {} {:?} {:?}", i, dt, e.ts);
}
if let Some(&(i, _, _)) = c.first() {