explicit multithreading and logging tweak

src/bin/wind_service.rs

@@ -6,7 +6,7 @@ use clap::{arg, Command};
 use davis_rpi::{api::WindServer, davis::Davis};
 use tokio::signal;
 
-#[tokio::main]
+#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
 async fn main() -> Result<()> {
     tracing_subscriber::fmt::init();
 

src/davis.rs

@@ -160,6 +160,7 @@ pub async fn cleanup_loop(db: Arc<DB>, t: Duration) {
 
 /// at every period, convert counter value to wind speed using formula from manufacturer
 pub async fn fetch_data_loop(period: Duration, counter: Arc<AtomicU64>, db: Arc<DB>) {
+    let mut prev_count = 0;
     let mut interval = interval(period);
     interval.tick().await; // we skip first tick to get some data from sensor first
     loop {
@@ -168,11 +169,18 @@ pub async fn fetch_data_loop(period: Duration, counter: Arc<AtomicU64>, db: Arc<
         let wind_speed_mph = count as f64 * (2.25 / period.as_secs_f64());
         tracing::debug!("Number of IO edges: {:?}", &count);
         let vel = wind_speed_mph * 0.44704;
-        if vel > 30.0 {
+        if vel > 25.0 {
+            tracing::warn!(
+                "Filtering out very high edge count: {:?}, previous was {:?}, calculated vel is: {:?}",
+                &count,
+                &prev_count,
+                vel,
+            );
             // filter out too high values, something is wrong
             continue;
         }
-        tracing::info!("Read vel: {:?}", &vel);
+        prev_count = count;
+        tracing::debug!("Read vel: {:?}", &vel);
         if let Err(err) = db.insert_measurement(vel, 0).await {
             tracing::error!("Failed to write measurement in DB!, {:?}", err);
         }

src/db.rs

@@ -149,15 +149,13 @@ impl DB {
             };
             measurements.push(mesurement);
         }
-        tracing::info!("Sending range: {:?}", &measurements);
         Ok(measurements)
     }
 
     pub async fn last_data(&self) -> Result<Measurement> {
         let row = sqlx::query!("SELECT ts, vel, direction  FROM wind ORDER BY ts DESC LIMIT 1",)
             .fetch_one(&self.pool)
             .await?;
-        tracing::info!("Sending last data: {:?}", &row);
         Ok(Measurement {
             ts: row.ts.ok_or_else(|| anyhow::anyhow!("not found"))?,
             vel: row.vel.ok_or_else(|| anyhow::anyhow!("Not found"))?,
@@ -172,7 +170,6 @@ impl DB {
         let row = sqlx::query!("SELECT ts, vel, direction  FROM wind ORDER BY ts ASC LIMIT 1",)
             .fetch_one(&self.pool)
             .await?;
-        tracing::info!("Sending oldest data: {:?}", &row);
         Ok(Measurement {
             ts: row.ts.ok_or_else(|| anyhow::anyhow!("not found"))?,
             vel: row.vel.ok_or_else(|| anyhow::anyhow!("Not found"))?,