Minimize read/write cache contention (#676)

* Add timeout to cache.write() call

* mixnode_cache -> validator_cache

* Fix rename

* Minimize read/write contention

validator-api/src/cache.rs

@@ -1,27 +1,33 @@
 use anyhow::Result;
 use mixnet_contract::{GatewayBond, MixNodeBond};
-use std::time::Instant;
+use serde::Serialize;
+use std::time::{SystemTime, UNIX_EPOCH};
+use tokio::sync::RwLock;
 use validator_client::Client;
 
 pub struct ValidatorCache {
-    mixnodes: Cache<Vec<MixNodeBond>>,
-    gateways: Cache<Vec<GatewayBond>>,
+    mixnodes: RwLock<Cache<Vec<MixNodeBond>>>,
+    gateways: RwLock<Cache<Vec<GatewayBond>>>,
     validator_client: Client,
 }
 
-#[derive(Default)]
-struct Cache<T> {
+#[derive(Default, Serialize, Clone)]
+pub struct Cache<T> {
     value: T,
     #[allow(dead_code)]
-    as_at: Option<Instant>,
+    as_at: u64,
 }
 
 impl<T: Clone> Cache<T> {
     fn set(&mut self, value: T) {
         self.value = value;
-        self.as_at = Some(Instant::now())
+        self.as_at = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs()
     }
 
+    #[allow(dead_code)]
     pub fn get(&self) -> T {
         self.value.clone()
     }
@@ -32,27 +38,28 @@ impl ValidatorCache {
         let config = validator_client::Config::new(validators_rest_uris, mixnet_contract);
         let validator_client = validator_client::Client::new(config);
         ValidatorCache {
-            mixnodes: Cache::default(),
-            gateways: Cache::default(),
+            mixnodes: RwLock::new(Cache::default()),
+            gateways: RwLock::new(Cache::default()),
             validator_client,
         }
     }
 
-    pub async fn cache(&mut self) -> Result<()> {
+    pub async fn refresh_cache(&self) -> Result<()> {
         let (mixnodes, gateways) = tokio::join!(
             self.validator_client.get_mix_nodes(),
             self.validator_client.get_gateways()
         );
-        self.mixnodes.set(mixnodes?);
-        self.gateways.set(gateways?);
+        self.mixnodes.write().await.set(mixnodes?);
+        self.gateways.write().await.set(gateways?);
+
         Ok(())
     }
 
-    pub fn mixnodes(&self) -> Vec<MixNodeBond> {
-        self.mixnodes.get()
+    pub async fn mixnodes(&self) -> Cache<Vec<MixNodeBond>> {
+        self.mixnodes.read().await.clone()
     }
 
-    pub fn gateways(&self) -> Vec<GatewayBond> {
-        self.gateways.get()
+    pub async fn gateways(&self) -> Cache<Vec<GatewayBond>> {
+        self.gateways.read().await.clone()
     }
 }

validator-api/src/main.rs

@@ -23,7 +23,7 @@ use crate::tested_network::good_topology::parse_topology_file;
 use crate::tested_network::TestedNetwork;
 use ::config::NymConfig;
 use anyhow::Result;
-use cache::ValidatorCache;
+use cache::{Cache, ValidatorCache};
 use clap::{App, Arg, ArgMatches};
 use crypto::asymmetric::{encryption, identity};
 use futures::channel::mpsc;
@@ -32,7 +32,6 @@ use mixnet_contract::{GatewayBond, MixNodeBond};
 use nymsphinx::addressing::clients::Recipient;
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::sync::RwLock;
 use tokio::time;
 use topology::NymTopology;
 
@@ -259,15 +258,13 @@ fn check_if_up_to_date(v4_topology: &NymTopology, v6_topology: &NymTopology) {
 }
 
 #[get("/mixnodes")]
-async fn get_mixnodes(cache: &State<Arc<RwLock<ValidatorCache>>>) -> Json<Vec<MixNodeBond>> {
-    let cache = cache.read().await;
-    Json(cache.mixnodes())
+async fn get_mixnodes(cache: &State<Arc<ValidatorCache>>) -> Json<Cache<Vec<MixNodeBond>>> {
+    Json(cache.mixnodes().await)
 }
 
 #[get("/gateways")]
-async fn get_gateways(cache: &State<Arc<RwLock<ValidatorCache>>>) -> Json<Vec<GatewayBond>> {
-    let cache = cache.read().await;
-    Json(cache.gateways())
+async fn get_gateways(cache: &State<Arc<ValidatorCache>>) -> Json<Cache<Vec<GatewayBond>>> {
+    Json(cache.gateways().await)
 }
 
 fn override_config(mut config: Config, matches: &ArgMatches) -> Config {
@@ -442,24 +439,18 @@ async fn main() -> Result<()> {
         info!("Network monitoring is disabled.")
     }
 
-    let mixnode_cache = Arc::new(RwLock::new(ValidatorCache::init(
+    let validator_cache = Arc::new(ValidatorCache::init(
         config.get_validators_urls(),
         config.get_mixnet_contract_address(),
-    )));
+    ));
 
-    let write_mixnode_cache = Arc::clone(&mixnode_cache);
+    let write_validator_cache = Arc::clone(&validator_cache);
 
     tokio::spawn(async move {
         let mut interval = time::interval(config.get_caching_interval());
         loop {
             interval.tick().await;
-            {
-                match write_mixnode_cache.try_write() {
-                    Ok(mut w) => w.cache().await.unwrap(),
-                    // If we don't get the write lock skip a tick
-                    Err(e) => error!("Could not aquire write lock on cache: {}", e),
-                }
-            }
+            write_validator_cache.refresh_cache().await.unwrap()
         }
     });
 
@@ -481,7 +472,7 @@ async fn main() -> Result<()> {
     rocket::build()
         .attach(cors)
         .mount("/v1", routes![get_mixnodes, get_gateways])
-        .manage(mixnode_cache)
+        .manage(validator_cache)
         .ignite()
         .await?
         .launch()