diff --git a/Cargo.lock b/Cargo.lock index 49e150649a..e938994bf7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -273,7 +273,7 @@ dependencies = [ "idna 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", "publicsuffix 1.5.4 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", "time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)", "try_from 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)", @@ -513,6 +513,29 @@ name = "futures" version = "0.1.29" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "futures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-channel" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "futures-core" version = "0.3.1" @@ -527,6 +550,60 @@ dependencies = [ "num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "futures-executor" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-io" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-macro" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.8 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-sink" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-task" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-util" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "memchr 2.2.1 (registry+https://github.com/rust-lang/crates.io-index)", + "pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)", + "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "generic-array" version = "0.12.3" @@ -911,9 +988,12 @@ dependencies = [ "clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)", "curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)", "dirs 2.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", "pem 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)", + "rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)", + "rand_distr 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)", "reqwest 0.9.24 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", "sfw-provider-requests 0.1.0", "sphinx 0.1.0", "tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)", @@ -927,6 +1007,7 @@ dependencies = [ "base64 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)", "clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)", "curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)", + "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", "nym-client 0.1.0", "sphinx 0.1.0", "tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)", @@ -1016,6 +1097,11 @@ name = "pin-project-lite" version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "pin-utils" +version = "0.1.0-alpha.4" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "pkg-config" version = "0.3.17" @@ -1026,6 +1112,21 @@ name = "ppv-lite86" version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "proc-macro-hack" +version = "0.5.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.8 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "proc-macro-nested" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "proc-macro2" version = "1.0.6" @@ -1262,7 +1363,7 @@ dependencies = [ "mime 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)", "mime_guess 2.0.1 (registry+https://github.com/rust-lang/crates.io-index)", "native-tls 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", "serde_urlencoded 0.5.5 (registry+https://github.com/rust-lang/crates.io-index)", "time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)", @@ -1352,15 +1453,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index" [[package]] name = "serde" -version = "1.0.103" +version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ - "serde_derive 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] name = "serde_derive" -version = "1.0.103" +version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", @@ -1375,7 +1476,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)", "ryu 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -1385,7 +1486,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" dependencies = [ "dtoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)", "itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)", - "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", "url 1.7.2 (registry+https://github.com/rust-lang/crates.io-index)", ] @@ -1963,8 +2064,16 @@ dependencies = [ "checksum fuchsia-zircon 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2e9763c69ebaae630ba35f74888db465e49e259ba1bc0eda7d06f4a067615d82" "checksum fuchsia-zircon-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7" "checksum futures 0.1.29 (registry+https://github.com/rust-lang/crates.io-index)" = "1b980f2816d6ee8673b6517b52cb0e808a180efc92e5c19d02cdda79066703ef" +"checksum futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b6f16056ecbb57525ff698bb955162d0cd03bee84e6241c27ff75c08d8ca5987" +"checksum futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "fcae98ca17d102fd8a3603727b9259fcf7fa4239b603d2142926189bc8999b86" "checksum futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "79564c427afefab1dfb3298535b21eda083ef7935b4f0ecbfcb121f0aec10866" "checksum futures-cpupool 0.1.8 (registry+https://github.com/rust-lang/crates.io-index)" = "ab90cde24b3319636588d0c35fe03b1333857621051837ed769faefb4c2162e4" +"checksum futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1e274736563f686a837a0568b478bdabfeaec2dca794b5649b04e2fe1627c231" +"checksum futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "e676577d229e70952ab25f3945795ba5b16d63ca794ca9d2c860e5595d20b5ff" +"checksum futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "52e7c56c15537adb4f76d0b7a76ad131cb4d2f4f32d3b0bcabcbe1c7c5e87764" +"checksum futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "171be33efae63c2d59e6dbba34186fe0d6394fb378069a76dfd80fdcffd43c16" +"checksum futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0bae52d6b29cf440e298856fec3965ee6fa71b06aa7495178615953fd669e5f9" +"checksum futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "c0d66274fb76985d3c62c886d1da7ac4c0903a8c9f754e8fe0f35a6a6cc39e76" "checksum generic-array 0.12.3 (registry+https://github.com/rust-lang/crates.io-index)" = "c68f0274ae0e023facc3c97b2e00f076be70e254bc851d972503b328db79b2ec" "checksum getrandom 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)" = "e7db7ca94ed4cd01190ceee0d8a8052f08a247aa1b469a7f68c6a3b71afcf407" "checksum h2 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)" = "a5b34c246847f938a410a03c5458c7fee2274436675e76d8b903c08efc29c462" @@ -2015,8 +2124,11 @@ dependencies = [ "checksum percent-encoding 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "31010dd2e1ac33d5b46a5b413495239882813e0369f8ed8a5e266f173602f831" "checksum percent-encoding 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e" "checksum pin-project-lite 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f0af6cbca0e6e3ce8692ee19fb8d734b641899e07b68eb73e9bbbd32f1703991" +"checksum pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)" = "5894c618ce612a3fa23881b152b608bafb8c56cfc22f434a3ba3120b40f7b587" "checksum pkg-config 0.3.17 (registry+https://github.com/rust-lang/crates.io-index)" = "05da548ad6865900e60eaba7f589cc0783590a92e940c26953ff81ddbab2d677" "checksum ppv-lite86 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "74490b50b9fbe561ac330df47c08f3f33073d2d00c150f719147d7c54522fa1b" +"checksum proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)" = "ecd45702f76d6d3c75a80564378ae228a85f0b59d2f3ed43c91b4a69eb2ebfc5" +"checksum proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "369a6ed065f249a159e06c45752c780bda2fb53c995718f9e484d08daa9eb42e" "checksum proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "9c9e470a8dc4aeae2dee2f335e8f533e2d4b347e1434e5671afc49b054592f27" "checksum publicsuffix 1.5.4 (registry+https://github.com/rust-lang/crates.io-index)" = "3bbaa49075179162b49acac1c6aa45fb4dafb5f13cf6794276d77bc7fd95757b" "checksum quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "053a8c8bcc71fcce321828dc897a98ab9760bef03a4fc36693c231e5b3216cfe" @@ -2052,8 +2164,8 @@ dependencies = [ "checksum security-framework-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "e31493fc37615debb8c5090a7aeb4a9730bc61e77ab10b9af59f1a202284f895" "checksum semver 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1d7eb9ef2c18661902cc47e535f9bc51b78acd254da71d375c2f6720d9a40403" "checksum semver-parser 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3" -"checksum serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)" = "1217f97ab8e8904b57dd22eb61cde455fa7446a9c1cf43966066da047c1f3702" -"checksum serde_derive 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)" = "a8c6faef9a2e64b0064f48570289b4bf8823b7581f1d6157c1b52152306651d0" +"checksum serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)" = "414115f25f818d7dfccec8ee535d76949ae78584fc4f79a6f45a904bf8ab4449" +"checksum serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)" = "128f9e303a5a29922045a830221b8f78ec74a5f544944f3d5984f8ec3895ef64" "checksum serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)" = "48c575e0cc52bdd09b47f330f646cf59afc586e9c4e3ccd6fc1f625b8ea1dad7" "checksum serde_urlencoded 0.5.5 (registry+https://github.com/rust-lang/crates.io-index)" = "642dd69105886af2efd227f75a520ec9b44a820d65bc133a9131f7d229fd165a" "checksum sha-1 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)" = "23962131a91661d643c98940b20fcaffe62d776a823247be80a48fcb8b6fce68" diff --git a/Cargo.toml b/Cargo.toml index 47f4f140e2..0adfb88178 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,7 @@ edition = "2018" base64 = "0.11.0" clap = "2.33.0" curve25519-dalek = "1.2.3" +futures = "0.3.1" nym-client = { path = "../nym-client" } sphinx = { path = "../sphinx" } tokio = { version = "0.2", features = ["full"] } diff --git a/src/mix_peer.rs b/src/mix_peer.rs index c2fbde8477..0e5b235c29 100644 --- a/src/mix_peer.rs +++ b/src/mix_peer.rs @@ -9,6 +9,7 @@ pub struct MixPeer { impl MixPeer { // note that very soon `next_hop_address` will be changed to `next_hop_metadata` pub fn new(next_hop_address: [u8; 32]) -> MixPeer { + println!("constructing next hop from: {:?}", next_hop_address); let b = next_hop_address; let host = Ipv4Addr::new(b[0], b[1], b[2], b[3]); let port: u16 = u16::from_be_bytes([b[4], b[5]]); @@ -26,6 +27,7 @@ impl MixPeer { } pub fn to_string(&self) -> String { + println!("going to report metrics for: {:?}. original: {:?}", self.connection.to_string(), self.connection); self.connection.to_string() } } diff --git a/src/node/metrics.rs b/src/node/metrics.rs new file mode 100644 index 0000000000..afacc9ad08 --- /dev/null +++ b/src/node/metrics.rs @@ -0,0 +1,86 @@ +use std::sync::{RwLock, Arc}; +use futures::lock::Mutex; +use futures::channel::mpsc; +use futures::{StreamExt, SinkExt}; +use std::time::Duration; +use nym_client::clients::directory; +use nym_client::clients::directory::DirectoryClient; +use nym_client::clients::directory::requests::metrics_mixes_post::MetricsMixPoster; +use nym_client::clients::directory::metrics::MixMetric; +use curve25519_dalek::montgomery::MontgomeryPoint; +use std::collections::HashMap; + +const METRICS_INTERVAL: u64 = 3; + +#[derive(Debug)] +pub struct MetricsReporter { + received: u64, + sent: HashMap +} + +impl MetricsReporter { + pub(crate) fn new() -> Self { + MetricsReporter{ + received: 0, + sent: HashMap::new(), + } + } + + pub(crate) fn add_arc_mutex(self) -> Arc> { + Arc::new(Mutex::new(self)) + } + + async fn increment_received_metrics(mut metrics: Arc>) { + let mut unlocked = metrics.lock().await; + unlocked.received += 1; + println!("new metrics: {:?}", unlocked); + } + + pub(crate) async fn run_received_metrics_control(metrics: Arc>, mut rx: mpsc::Receiver<()>) { + while let Some(received_metric) = rx.next().await { + println!("[METRIC] - received new packet!"); + MetricsReporter::increment_received_metrics(metrics.clone()).await; + } + } + + async fn increment_sent_metrics(mut metrics: Arc>, sent_to: String) { + let mut unlocked = metrics.lock().await; + let receiver_count = unlocked.sent.entry(sent_to).or_insert(0); + *receiver_count += 1; + +// unlocked.sent += 1; + println!("new metrics: {:?}", unlocked); + } + + pub(crate) async fn run_sent_metrics_control(metrics: Arc>, mut rx: mpsc::Receiver) { + while let Some(sent_metric) = rx.next().await { + println!("[METRIC] - sent new packet!"); + MetricsReporter::increment_sent_metrics(metrics.clone(), sent_metric).await; + } + } + + async fn acquire_and_reset_metrics(mut metrics: Arc>) -> (u64, HashMap) { + let mut unlocked = metrics.lock().await; + let received = unlocked.received; + + println!("RESETTING METRICS!"); + let sent = std::mem::replace(&mut unlocked.sent, HashMap::new()); + unlocked.received = 0; + + (received, sent) + } + + pub(crate) async fn run_metrics_sender(metrics: Arc>, cfg: directory::Config, pub_key_str: String) { + let delay_duration = Duration::from_secs(METRICS_INTERVAL); + let directory_client= directory::Client::new(cfg); + loop { + tokio::time::delay_for(delay_duration).await; + let (received, sent) = MetricsReporter::acquire_and_reset_metrics(metrics.clone()).await; + directory_client.metrics_post.post(&MixMetric{ + pub_key: pub_key_str.clone(), + received, + sent, + }); + } + } +} diff --git a/src/node/mod.rs b/src/node/mod.rs index 36b1674b44..776259248c 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -1,18 +1,23 @@ use std::net::SocketAddr; use std::sync::{Arc, RwLock}; use std::time::Duration; - +use futures::channel::mpsc; use curve25519_dalek::montgomery::MontgomeryPoint; use curve25519_dalek::scalar::Scalar; use sphinx::header::delays::Delay as SphinxDelay; use sphinx::{ProcessedPacket, SphinxPacket}; use tokio::prelude::*; use tokio::runtime::Runtime; +use futures::lock::Mutex; use crate::mix_peer::MixPeer; +use crate::node::metrics::MetricsReporter; +use futures::SinkExt; +use nym_client::clients::directory; mod presence; pub mod runner; +mod metrics; pub struct Config { directory_server: String, @@ -25,8 +30,7 @@ pub struct Config { impl Config { pub fn public_key_string(&self) -> String { let key_bytes = self.public_key.to_bytes().to_vec(); - let b64 = base64::encode_config(&key_bytes, base64::URL_SAFE); - b64.to_string() + base64::encode_config(&key_bytes, base64::URL_SAFE) } } @@ -50,15 +54,18 @@ struct ForwardingData { packet: SphinxPacket, delay: SphinxDelay, recipient: MixPeer, + sent_metrics_tx: mpsc::Sender, } // TODO: this will need to be changed if MixPeer will live longer than our Forwarding Data impl ForwardingData { - fn new(packet: SphinxPacket, delay: SphinxDelay, recipient: MixPeer) -> Self { + fn new(packet: SphinxPacket, delay: SphinxDelay, recipient: MixPeer, sent_metrics_tx: mpsc::Sender, + ) -> Self { ForwardingData { packet, delay, recipient, + sent_metrics_tx, } } } @@ -66,28 +73,36 @@ impl ForwardingData { // ProcessingData defines all data required to correctly unwrap sphinx packets struct ProcessingData { secret_key: Scalar, + received_metrics_tx: mpsc::Sender<()>, + sent_metrics_tx: mpsc::Sender, } impl ProcessingData { - fn new(secret_key: Scalar) -> Self { - ProcessingData { secret_key } + fn new(secret_key: Scalar, received_metrics_tx: mpsc::Sender<()>, sent_metrics_tx: mpsc::Sender) -> Self { + ProcessingData { secret_key, received_metrics_tx, sent_metrics_tx } } - fn add_arc_rwlock(self) -> Arc> { - Arc::new(RwLock::new(self)) + fn add_arc_mutex(self) -> Arc> { + Arc::new(Mutex::new(self)) } } struct PacketProcessor {} impl PacketProcessor { - pub fn process_sphinx_data_packet( + pub async fn process_sphinx_data_packet( packet_data: &[u8], - processing_data: Arc>, + processing_data: Arc>, ) -> Result { + // we received something resembling a sphinx packet, report it! + let processing_data = processing_data.lock().await; + let mut received_sender = processing_data.received_metrics_tx.clone(); + + received_sender.send(()).await; + let packet = SphinxPacket::from_bytes(packet_data.to_vec())?; let (next_packet, next_hop_address, delay) = - match packet.process(processing_data.read().unwrap().secret_key) { + match packet.process(processing_data.secret_key) { ProcessedPacket::ProcessedPacketForwardHop(packet, address, delay) => { (packet, address, delay) } @@ -96,35 +111,38 @@ impl PacketProcessor { let next_mix = MixPeer::new(next_hop_address); - let fwd_data = ForwardingData::new(next_packet, delay, next_mix); + let fwd_data = ForwardingData::new(next_packet, delay, next_mix, processing_data.sent_metrics_tx.clone()); Ok(fwd_data) } - async fn wait_and_forward(forwarding_data: ForwardingData) { + async fn wait_and_forward(mut forwarding_data: ForwardingData) { let delay_duration = Duration::from_nanos(forwarding_data.delay.get_value()); println!("client says to wait for {:?}", delay_duration); tokio::time::delay_for(delay_duration).await; - println!("waited {:?} - time to forward the packet!", delay_duration); + println!("waited {:?} - time to forward the packet! (and update metrics!)", delay_duration); + forwarding_data.sent_metrics_tx.send(forwarding_data.recipient.to_string()).await; match forwarding_data .recipient .send(forwarding_data.packet.to_bytes()) .await - { - Ok(()) => (), - Err(e) => { - println!( - "failed to write bytes to next mix peer. err = {:?}", - e.to_string() - ); + { + Ok(()) => (), + Err(e) => { + println!( + "failed to write bytes to next mix peer. err = {:?}", + e.to_string() + ); + } } - } } } // the MixNode will live for whole duration of this program pub struct MixNode { + directory_server: String, network_address: SocketAddr, + public_key: MontgomeryPoint, secret_key: Scalar, layer: usize, } @@ -132,15 +150,17 @@ pub struct MixNode { impl MixNode { pub fn new(config: &Config) -> Self { MixNode { + directory_server: config.directory_server.clone(), network_address: config.socket_address, secret_key: config.secret_key, + public_key: config.public_key, layer: config.layer, } } async fn process_socket_connection( mut socket: tokio::net::TcpStream, - processing_data: Arc>, + processing_data: Arc>, ) { // NOTE: processing_data is copied here!! let mut buf = [0u8; sphinx::PACKET_SIZE]; @@ -157,8 +177,8 @@ impl MixNode { let fwd_data = PacketProcessor::process_sphinx_data_packet( buf.as_ref(), processing_data.clone(), - ) - .unwrap(); + ).await + .unwrap(); PacketProcessor::wait_and_forward(fwd_data).await; } Err(e) => { @@ -175,14 +195,37 @@ impl MixNode { } } - pub fn start_listening(&self) -> Result<(), Box> { + pub fn start(&self) -> Result<(), Box> { // Create the runtime, probably later move it to MixNode itself? let mut rt = Runtime::new()?; + let (received_tx, received_rx) = mpsc::channel(1024); + let (sent_tx, sent_rx) = mpsc::channel(1024); + + let directory_cfg = directory::Config { base_url: self.directory_server.clone() }; + let pub_key_str = base64::encode_config(&self.public_key.to_bytes().to_vec(), base64::URL_SAFE); + + let metrics = MetricsReporter::new().add_arc_mutex(); + rt.spawn(MetricsReporter::run_received_metrics_control(metrics.clone(), received_rx)); + rt.spawn(MetricsReporter::run_sent_metrics_control(metrics.clone(), sent_rx)); + rt.spawn(MetricsReporter::run_metrics_sender(metrics, directory_cfg, pub_key_str)); +// let mut foo = received_tx.clone(); +// let mut bar = sent_tx.clone(); + // let's increase some metrics! +// rt.spawn(async move { +// loop { +// tokio::time::delay_for(Duration::from_secs(2)).await; +// foo.send(()).await; +// bar.send("bar".to_string()).await; +// foo.send(()).await; +// } +// }); +// let (sent_tx, sent_rx) = mpsc::channel(1024); + // Spawn the root task rt.block_on(async { let mut listener = tokio::net::TcpListener::bind(self.network_address).await?; - let processing_data = ProcessingData::new(self.secret_key).add_arc_rwlock(); + let processing_data = ProcessingData::new(self.secret_key, received_tx, sent_tx).add_arc_mutex(); loop { let (socket, _) = listener.accept().await?; diff --git a/src/node/presence.rs b/src/node/presence.rs index c2acff1fc2..848cb70e06 100644 --- a/src/node/presence.rs +++ b/src/node/presence.rs @@ -20,7 +20,7 @@ impl Notifier { let presence = MixNodePresence { host: node_config.socket_address.to_string(), // note: the directory server formats the real incoming IP itself pub_key: node_config.public_key_string(), - layer: 666, + layer: node_config.layer as u64, last_seen: 666, }; Notifier { diff --git a/src/node/runner.rs b/src/node/runner.rs index c841069dd9..3823592084 100644 --- a/src/node/runner.rs +++ b/src/node/runner.rs @@ -21,7 +21,7 @@ pub fn start(matches: &ArgMatches) { let notifier = presence::Notifier::new(&config); notifier.run(); }); - mix.start_listening().unwrap(); + mix.start().unwrap(); } fn new_config(matches: &ArgMatches) -> node::Config {