Fixes average stats bug (#436)

* Add test

* Fix test

* Add fix

Author: zainkabani

src/stats.rs

@@ -107,8 +107,19 @@ impl Collector {
             loop {
                 interval.tick().await;
 
-                for stats in SERVER_STATS.read().values() {
-                    stats.address_stats().update_averages();
+                // Hold read lock for duration of update to retain all server stats
+                let server_stats = SERVER_STATS.read();
+
+                for stats in server_stats.values() {
+                    if !stats.check_address_stat_average_is_updated_status() {
+                        stats.address_stats().update_averages();
+                        stats.set_address_stat_average_is_updated_status(true);
+                    }
+                }
+
+                // Reset to false for next update
+                for stats in server_stats.values() {
+                    stats.set_address_stat_average_is_updated_status(false);
                 }
             }
         });

src/stats/address.rs

@@ -30,6 +30,9 @@ pub struct AddressStats {
     pub avg_xact_time: Arc<AtomicU64>,
     pub avg_xact_count: Arc<AtomicU64>,
     pub avg_wait_time: Arc<AtomicU64>,
+
+    // Determines if the averages have been updated since the last time they were reported
+    pub averages_updated: Arc<AtomicBool>,
 }
 
 impl IntoIterator for AddressStats {
@@ -114,15 +117,14 @@ impl AddressStats {
 
     pub fn update_averages(&self) {
         let (totals, averages, old_totals) = self.fields_iterators();
-        for data in itertools::izip!(totals, averages, old_totals) {
-            let (total, average, old_total) = data;
-            let total = total.load(Ordering::Relaxed);
-            let old = old_total.load(Ordering::Relaxed);
+        for (total, average, old_total) in itertools::izip!(totals, averages, old_totals) {
+            let total_value = total.load(Ordering::Relaxed);
+            let old_total_value = old_total.load(Ordering::Relaxed);
             average.store(
-                (total - old) / (crate::stats::STAT_PERIOD / 1_000),
+                (total_value - old_total_value) / (crate::stats::STAT_PERIOD / 1_000),
                 Ordering::Relaxed,
             ); // Avg / second
-            old_total.store(total, Ordering::Relaxed);
+            old_total.store(total_value, Ordering::Relaxed);
         }
     }
 

src/stats/server.rs

@@ -139,6 +139,17 @@ impl ServerStats {
         self.address.stats.clone()
     }
 
+    pub fn check_address_stat_average_is_updated_status(&self) -> bool {
+        self.address.stats.averages_updated.load(Ordering::Relaxed)
+    }
+
+    pub fn set_address_stat_average_is_updated_status(&self, is_checked: bool) {
+        self.address
+            .stats
+            .averages_updated
+            .store(is_checked, Ordering::Relaxed);
+    }
+
     // Helper methods for show_servers
     pub fn pool_name(&self) -> String {
         self.pool_stats.database()

tests/ruby/admin_spec.rb

@@ -14,19 +14,20 @@
   describe "SHOW STATS" do
     context "clients connect and make one query" do
       it "updates *_query_time and *_wait_time" do
-        connection = PG::connect("#{pgcat_conn_str}?application_name=one_query")
-        connection.async_exec("SELECT pg_sleep(0.25)")
-        connection.async_exec("SELECT pg_sleep(0.25)")
-        connection.async_exec("SELECT pg_sleep(0.25)")
-        connection.close
+        connections = Array.new(3) { PG::connect("#{pgcat_conn_str}?application_name=one_query") }
+        connections.each do |c|
+          Thread.new { c.async_exec("SELECT pg_sleep(0.25)") }
+        end
+        sleep(1)
+        connections.map(&:close)
 
         # wait for averages to be calculated, we shouldn't do this too often
         sleep(15.5)
         admin_conn = PG::connect(processes.pgcat.admin_connection_string)
         results = admin_conn.async_exec("SHOW STATS")[0]
         admin_conn.close
         expect(results["total_query_time"].to_i).to be_within(200).of(750)
-        expect(results["avg_query_time"].to_i).to_not eq(0)
+        expect(results["avg_query_time"].to_i).to be_within(20).of(50)
 
         expect(results["total_wait_time"].to_i).to_not eq(0)
         expect(results["avg_wait_time"].to_i).to_not eq(0)