Reduce worker count to reduce postgres connections

daqingest/src/daemon.rs

@@ -93,7 +93,7 @@ impl Daemon {
         // TODO keep join handles and await later
         let (channel_info_query_tx, jhs, jh) = dbpg::seriesbychannel::start_lookup_workers::<
             dbpg::seriesbychannel::SalterRandom,
-        >(4, &opts.pgconf, series_by_channel_stats.clone())
+        >(2, &opts.pgconf, series_by_channel_stats.clone())
         .await
         .map_err(|e| Error::with_msg_no_trace(e.to_string()))?;
 

dbpg/src/seriesbychannel.rs

@@ -154,6 +154,7 @@ impl Worker {
         stats: Arc<SeriesByChannelStats>,
     ) -> Result<Self, Error> {
         use tokio_postgres::types::Type;
+        debug!("Worker  make_pg_client");
         let (pg, pg_client_jh) = crate::conn::make_pg_client(db).await?;
         let sql = concat!(
             "with q1 as (",
@@ -643,6 +644,7 @@ impl HashSalter for SalterRandom {
     }
 }
 
+#[cfg(test)]
 async fn psql_play(db: &Database) -> Result<(), Error> {
     use tokio_postgres::types::ToSql;
     use tokio_postgres::types::Type;

netfetch/src/ca/finder.rs

@@ -22,7 +22,7 @@ use std::time::Instant;
 use taskrun::tokio;
 use tokio::task::JoinHandle;
 
-const SEARCH_DB_PIPELINE_LEN: usize = 4;
+const SEARCH_DB_PIPELINE_LEN: usize = 2;
 
 #[allow(unused)]
 macro_rules! debug_batch {
@@ -144,6 +144,7 @@ async fn finder_worker_single(
     db: Database,
     stats: Arc<IocFinderStats>,
 ) -> Result<(), Error> {
+    debug!("finder_worker_single  make_pg_client");
     let (pg, jh) = make_pg_client(&db)
         .await
         .map_err(|e| Error::with_msg_no_trace(e.to_string()))?;

netfetch/src/metrics/ingest.rs

@@ -11,8 +11,8 @@ use err::ThisError;
 use futures_util::StreamExt;
 use futures_util::TryStreamExt;
 use items_2::eventsdim0::EventsDim0;
-use items_2::eventsdim0::EventsDim0Enum;
 use items_2::eventsdim0::EventsDim0NoPulse;
+use items_2::eventsdim0enum::EventsDim0Enum;
 use items_2::eventsdim1::EventsDim1;
 use items_2::eventsdim1::EventsDim1NoPulse;
 use netpod::log::*;

serieswriter/src/timebin.rs

@@ -418,7 +418,7 @@ fn store_bins(
                     .ts1s
                     .iter()
                     .zip(k.ts2s.iter())
-                    .zip(k.counts.iter())
+                    .zip(k.cnts.iter())
                     .zip(k.mins.iter())
                     .zip(k.maxs.iter())
                     .zip(k.avgs.iter())