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

Enable streaming results for channel index

Browse Source
This commit is contained in:
Dominik Werder
2021-06-02 17:18:49 +02:00
parent bfb5c9d28e
commit 79e3a1ea47
4 changed files with 208 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dbconn/src/lib.rs
+7 -7
View File
@@ -1,6 +1,6 @@
use err::Error; use err::Error;
use netpod::log::*; use netpod::log::*;
use netpod::{Channel, NodeConfigCached}; use netpod::{Channel, Database, NodeConfigCached};
use std::time::Duration; use std::time::Duration;
use tokio_postgres::{Client, NoTls}; use tokio_postgres::{Client, NoTls};
@@ -19,8 +19,8 @@ pub async fn delay_io_medium() {
delay_us(2000).await; delay_us(2000).await;
} }
pub async fn create_connection(node_config: &NodeConfigCached) -> Result<Client, Error> { pub async fn create_connection(db_config: &Database) -> Result<Client, Error> {
let d = &node_config.node_config.cluster.database; let d = db_config;
let uri = format!("postgresql://{}:{}@{}:{}/{}", d.user, d.pass, d.host, 5432, d.name); let uri = format!("postgresql://{}:{}@{}:{}/{}", d.user, d.pass, d.host, 5432, d.name);
let (cl, conn) = tokio_postgres::connect(&uri, NoTls).await?; let (cl, conn) = tokio_postgres::connect(&uri, NoTls).await?;
// TODO monitor connection drop. // TODO monitor connection drop.
@@ -34,7 +34,7 @@ pub async fn create_connection(node_config: &NodeConfigCached) -> Result<Client,
} }
pub async fn channel_exists(channel: &Channel, node_config: &NodeConfigCached) -> Result<bool, Error> { pub async fn channel_exists(channel: &Channel, node_config: &NodeConfigCached) -> Result<bool, Error> {
let cl = create_connection(node_config).await?; let cl = create_connection(&node_config.node_config.cluster.database).await?;
let rows = cl let rows = cl
.query("select rowid from channels where name = $1::text", &[&channel.name]) .query("select rowid from channels where name = $1::text", &[&channel.name])
.await?; .await?;
@@ -51,7 +51,7 @@ pub async fn channel_exists(channel: &Channel, node_config: &NodeConfigCached) -
} }
pub async fn database_size(node_config: &NodeConfigCached) -> Result<u64, Error> { pub async fn database_size(node_config: &NodeConfigCached) -> Result<u64, Error> {
let cl = create_connection(node_config).await?; let cl = create_connection(&node_config.node_config.cluster.database).await?;
let rows = cl let rows = cl
.query( .query(
"select pg_database_size($1::text)", "select pg_database_size($1::text)",
@@ -82,7 +82,7 @@ pub async fn table_sizes(node_config: &NodeConfigCached) -> Result<TableSizes, E
"ORDER BY pg_total_relation_size(C.oid) DESC LIMIT 20", "ORDER BY pg_total_relation_size(C.oid) DESC LIMIT 20",
); );
let sql = sql.as_str(); let sql = sql.as_str();
let cl = create_connection(node_config).await?; let cl = create_connection(&node_config.node_config.cluster.database).await?;
let rows = cl.query(sql, &[]).await?; let rows = cl.query(sql, &[]).await?;
let mut sizes = TableSizes { sizes: vec![] }; let mut sizes = TableSizes { sizes: vec![] };
sizes.sizes.push((format!("table"), format!("size"))); sizes.sizes.push((format!("table"), format!("size")));
@@ -94,7 +94,7 @@ pub async fn table_sizes(node_config: &NodeConfigCached) -> Result<TableSizes, E
pub async fn random_channel(node_config: &NodeConfigCached) -> Result<String, Error> { pub async fn random_channel(node_config: &NodeConfigCached) -> Result<String, Error> {
let sql = "select name from channels order by rowid limit 1 offset (random() * (select count(rowid) from channels))::bigint"; let sql = "select name from channels order by rowid limit 1 offset (random() * (select count(rowid) from channels))::bigint";
let cl = create_connection(node_config).await?; let cl = create_connection(&node_config.node_config.cluster.database).await?;
let rows = cl.query(sql, &[]).await?; let rows = cl.query(sql, &[]).await?;
if rows.len() == 0 { if rows.len() == 0 {
Err(Error::with_msg("can not get random channel"))?; Err(Error::with_msg("can not get random channel"))?;
dbconn/src/scan.rs
+161 -97
View File
@@ -3,9 +3,9 @@ use async_channel::{bounded, Receiver};
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use err::Error; use err::Error;
use futures_core::Stream; use futures_core::Stream;
use futures_util::{pin_mut, FutureExt}; use futures_util::{pin_mut, FutureExt, StreamExt};
use netpod::log::*; use netpod::log::*;
use netpod::NodeConfigCached; use netpod::{Database, NodeConfigCached};
use pin_project::pin_project; use pin_project::pin_project;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::future::Future; use std::future::Future;
@@ -203,7 +203,9 @@ impl UpdatedDbWithChannelNamesStream {
channel_inp_done: false, channel_inp_done: false,
clist: vec![], clist: vec![],
}; };
ret.client_fut = Some(Box::pin(crate::create_connection(ret.node_config_ref))); ret.client_fut = Some(Box::pin(crate::create_connection(
&ret.node_config_ref.node_config.cluster.database,
)));
Ok(ret) Ok(ret)
} }
} }
@@ -320,115 +322,177 @@ async fn update_db_with_channel_name_list(list: Vec<String>, backend: i64, dbc:
} }
pub async fn update_db_with_channel_names( pub async fn update_db_with_channel_names(
node_config: &NodeConfigCached, node_config: NodeConfigCached,
db_config: &Database,
) -> Result<Receiver<Result<UpdatedDbWithChannelNames, Error>>, Error> { ) -> Result<Receiver<Result<UpdatedDbWithChannelNames, Error>>, Error> {
let dbc = crate::create_connection(node_config).await?; let (tx, rx) = bounded(16);
let node_disk_ident = get_node_disk_ident(node_config, &dbc).await?; let db_config = db_config.clone();
let c1 = Arc::new(RwLock::new(0u32)); tokio::spawn(async move {
dbc.query("begin", &[]).await?; let dbc = crate::create_connection(&db_config).await?;
let dbc = Arc::new(dbc); let node_disk_ident = get_node_disk_ident(&node_config, &dbc).await?;
find_channel_names_from_config(&node_config.node.data_base_path, |ch| { let c1 = Arc::new(RwLock::new(0u32));
let ch = ch.to_owned(); dbc.query("begin", &[]).await?;
let dbc = dbc.clone(); let dbc = Arc::new(dbc);
let c1 = c1.clone(); let tx = Arc::new(tx);
let fac = node_disk_ident.facility; find_channel_names_from_config(&node_config.node.data_base_path, |ch| {
async move { let ch = ch.to_owned();
crate::delay_io_short().await; let dbc = dbc.clone();
dbc.query( let c1 = c1.clone();
"insert into channels (facility, name) values ($1, $2) on conflict do nothing", let tx = tx.clone();
&[&fac, &ch], let fac = node_disk_ident.facility;
) async move {
.await?; crate::delay_io_short().await;
let c2 = { dbc.query(
let mut g = c1.write()?; "insert into channels (facility, name) values ($1, $2) on conflict do nothing",
*g += 1; &[&fac, &ch],
*g )
}; .await?;
if c2 % 200 == 0 { let c2 = {
trace!("channels {:6} current {}", c2, ch); let mut g = c1.write()?;
dbc.query("commit", &[]).await?; *g += 1;
crate::delay_io_medium().await; *g
dbc.query("begin", &[]).await?; };
if c2 % 200 == 0 {
dbc.query("commit", &[]).await?;
let ret = UpdatedDbWithChannelNames {
msg: format!("current {}", ch),
count: c2,
};
tx.send(Ok(ret)).await?;
crate::delay_io_medium().await;
dbc.query("begin", &[]).await?;
}
Ok(())
} }
Ok(()) })
} .await?;
}) dbc.query("commit", &[]).await?;
.await?; let c2 = *c1.read()?;
dbc.query("commit", &[]).await?; let ret = UpdatedDbWithChannelNames {
let _ret = UpdatedDbWithChannelNames { msg: format!("all done"),
msg: format!("done"), count: c2,
count: *c1.read()?, };
}; tx.send(Ok(ret)).await?;
Ok(bounded(16).1) Ok::<_, Error>(())
});
Ok(rx)
}
pub fn update_db_with_channel_names_3<'a>(
node_config: &'a NodeConfigCached,
) -> impl Stream<Item = Result<UpdatedDbWithChannelNames, Error>> + 'static {
futures_util::future::ready(node_config.node.data_base_path.clone())
.then(|path| tokio::fs::read_dir(path))
.map(Result::unwrap)
.map(|rd| {
futures_util::stream::unfold(rd, move |rd| {
//let fut = rd.next_entry();
futures_util::future::ready(Ok(None)).map(move |item: Result<Option<u32>, Error>| match item {
Ok(Some(item)) => Some((item, rd)),
Ok(None) => None,
Err(_e) => None,
})
})
})
.map(|_conf| Err(Error::with_msg("TODO")))
.into_stream()
} }
#[derive(Debug, Serialize, Deserialize)] #[derive(Debug, Serialize, Deserialize)]
pub struct UpdatedDbWithAllChannelConfigs { pub struct UpdatedDbWithAllChannelConfigs {
msg: String,
count: u32, count: u32,
} }
pub async fn update_db_with_all_channel_configs( pub async fn update_db_with_all_channel_configs(
node_config: &NodeConfigCached, node_config: NodeConfigCached,
) -> Result<UpdatedDbWithAllChannelConfigs, Error> { ) -> Result<Receiver<Result<UpdatedDbWithAllChannelConfigs, Error>>, Error> {
let dbc = crate::create_connection(node_config).await?; let (tx, rx) = bounded(16);
let dbc = Arc::new(dbc); let tx = Arc::new(tx);
let node_disk_ident = &get_node_disk_ident(node_config, &dbc).await?; let tx2 = tx.clone();
let rows = dbc tokio::spawn(
.query( async move {
"select rowid, facility, name from channels where facility = $1 order by facility, name", let node_config = &node_config;
&[&node_config.node.backend], let dbc = crate::create_connection(&node_config.node_config.cluster.database).await?;
) let dbc = Arc::new(dbc);
.await?; let node_disk_ident = &get_node_disk_ident(node_config, &dbc).await?;
let mut c1 = 0; let rows = dbc
dbc.query("begin", &[]).await?; .query(
let mut count_inserted = 0; "select rowid, facility, name from channels where facility = $1 order by facility, name",
let mut count_updated = 0; &[&node_disk_ident.facility],
for row in rows { )
let rowid: i64 = row.try_get(0)?; .await?;
let _facility: i64 = row.try_get(1)?; let mut c1 = 0;
let channel: String = row.try_get(2)?; dbc.query("begin", &[]).await?;
match update_db_with_channel_config( let mut count_inserted = 0;
node_config, let mut count_updated = 0;
node_disk_ident, for row in rows {
rowid, let rowid: i64 = row.try_get(0)?;
&channel, let _facility: i64 = row.try_get(1)?;
dbc.clone(), let channel: String = row.try_get(2)?;
&mut count_inserted, match update_db_with_channel_config(
&mut count_updated, node_config,
) node_disk_ident,
.await rowid,
{ &channel,
/*Err(Error::ChannelConfigdirNotFound { .. }) => { dbc.clone(),
warn!("can not find channel config {}", channel); &mut count_inserted,
crate::delay_io_medium().await; &mut count_updated,
}*/ )
Err(e) => { .await
error!("{:?}", e); {
crate::delay_io_medium().await; /*Err(Error::ChannelConfigdirNotFound { .. }) => {
} warn!("can not find channel config {}", channel);
_ => { crate::delay_io_medium().await;
c1 += 1; }*/
if c1 % 200 == 0 { Err(e) => {
trace!( error!("{:?}", e);
"channel no {:6} inserted {:6} updated {:6}", crate::delay_io_medium().await;
c1, }
count_inserted, _ => {
count_updated c1 += 1;
); if c1 % 200 == 0 {
dbc.query("commit", &[]).await?; dbc.query("commit", &[]).await?;
dbc.query("begin", &[]).await?; let msg = format!(
"channel no {:6} inserted {:6} updated {:6}",
c1, count_inserted, count_updated
);
let ret = UpdatedDbWithAllChannelConfigs { msg, count: c1 };
tx.send(Ok(ret)).await?;
dbc.query("begin", &[]).await?;
}
crate::delay_io_short().await;
}
} }
crate::delay_io_short().await;
} }
dbc.query("commit", &[]).await?;
let msg = format!(
"ALL DONE channel no {:6} inserted {:6} updated {:6}",
c1, count_inserted, count_updated
);
let ret = UpdatedDbWithAllChannelConfigs { msg, count: c1 };
tx.send(Ok(ret)).await?;
Ok::<_, Error>(())
} }
} .then({
dbc.query("commit", &[]).await?; |item| async move {
let ret = UpdatedDbWithAllChannelConfigs { count: c1 }; match item {
Ok(ret) Ok(_) => {}
Err(e) => {
let msg = format!("Seeing error: {:?}", e);
let ret = UpdatedDbWithAllChannelConfigs { msg, count: 0 };
tx2.send(Ok(ret)).await?;
}
}
Ok::<_, Error>(())
}
}),
);
Ok(rx)
} }
pub async fn update_search_cache(node_config: &NodeConfigCached) -> Result<(), Error> { pub async fn update_search_cache(node_config: &NodeConfigCached) -> Result<(), Error> {
let dbc = crate::create_connection(node_config).await?; let dbc = crate::create_connection(&node_config.node_config.cluster.database).await?;
dbc.query("select update_cache()", &[]).await?; dbc.query("select update_cache()", &[]).await?;
Ok(()) Ok(())
} }
@@ -517,7 +581,7 @@ pub async fn update_db_with_all_channel_datafiles(
node_disk_ident: &NodeDiskIdent, node_disk_ident: &NodeDiskIdent,
ks_prefix: &str, ks_prefix: &str,
) -> Result<(), Error> { ) -> Result<(), Error> {
let dbc = Arc::new(crate::create_connection(node_config).await?); let dbc = Arc::new(crate::create_connection(&node_config.node_config.cluster.database).await?);
let rows = dbc let rows = dbc
.query( .query(
"select rowid, facility, name from channels where facility = $1 order by facility, name", "select rowid, facility, name from channels where facility = $1 order by facility, name",
dbconn/src/search.rs
+1 -1
View File
@@ -11,7 +11,7 @@ pub async fn search_channel(
"channel_id, channel_name, source_name, dtype, shape, unit, description, channel_backend", "channel_id, channel_name, source_name, dtype, shape, unit, description, channel_backend",
" from searchext($1, $2, $3, $4)", " from searchext($1, $2, $3, $4)",
)); ));
let cl = create_connection(node_config).await?; let cl = create_connection(&node_config.node_config.cluster.database).await?;
let rows = cl let rows = cl
.query( .query(
sql.as_str(), sql.as_str(),
httpret/src/lib.rs
+39 -7
View File
@@ -443,12 +443,38 @@ pub async fn update_db_with_channel_names(
Some(q) => q.contains("dry"), Some(q) => q.contains("dry"),
None => false, None => false,
}; };
let res = dbconn::scan::UpdatedDbWithChannelNamesStream::new(node_config.clone())?; let res =
//let res = dbconn::scan::update_db_with_channel_names(node_config).await?; dbconn::scan::update_db_with_channel_names(node_config.clone(), &node_config.node_config.cluster.database)
.await?;
let ret = response(StatusCode::OK) let ret = response(StatusCode::OK)
.header(http::header::CONTENT_TYPE, "application/json") .header(http::header::CONTENT_TYPE, "application/jsonlines")
.body(Body::wrap_stream(res.map(|k| match serde_json::to_string(&k) { .body(Body::wrap_stream(res.map(|k| match serde_json::to_string(&k) {
Ok(item) => Ok(item), Ok(mut item) => {
item.push('\n');
Ok(item)
}
Err(e) => Err(e),
})))?;
Ok(ret)
}
pub async fn update_db_with_channel_names_3(
req: Request<Body>,
node_config: &NodeConfigCached,
) -> Result<Response<Body>, Error> {
let (head, _body) = req.into_parts();
let _dry = match head.uri.query() {
Some(q) => q.contains("dry"),
None => false,
};
let res = dbconn::scan::update_db_with_channel_names_3(node_config);
let ret = response(StatusCode::OK)
.header(http::header::CONTENT_TYPE, "application/jsonlines")
.body(Body::wrap_stream(res.map(|k| match serde_json::to_string(&k) {
Ok(mut item) => {
item.push('\n');
Ok(item)
}
Err(e) => Err(e), Err(e) => Err(e),
})))?; })))?;
Ok(ret) Ok(ret)
@@ -463,10 +489,16 @@ pub async fn update_db_with_all_channel_configs(
Some(q) => q.contains("dry"), Some(q) => q.contains("dry"),
None => false, None => false,
}; };
let res = dbconn::scan::update_db_with_all_channel_configs(node_config).await?; let res = dbconn::scan::update_db_with_all_channel_configs(node_config.clone()).await?;
let ret = response(StatusCode::OK) let ret = response(StatusCode::OK)
.header(http::header::CONTENT_TYPE, "application/json") .header(http::header::CONTENT_TYPE, "application/jsonlines")
.body(Body::from(serde_json::to_string(&res)?))?; .body(Body::wrap_stream(res.map(|k| match serde_json::to_string(&k) {
Ok(mut item) => {
item.push('\n');
Ok(item)
}
Err(e) => Err(e),
})))?;
Ok(ret) Ok(ret)
} }