feat: add peers used bandwidth calculation and display in TUI (#1770)

* Add peers used bandwidth calculation and display in TUI
* Fix formatting
* Change Mutex to RwLock from peer's used bandwidth statistics in Tracker
* Make used bandwidth column in TUI peers list sort by sum of bytes

Cargo.lock

@@ -655,6 +655,7 @@ dependencies = [
  "grin_servers 0.3.0",
  "grin_util 0.3.0",
  "grin_wallet 0.3.0",
+ "humansize 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "reqwest 0.9.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde 1.0.79 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 1.0.28 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -960,6 +961,11 @@ name = "httparse"
 version = "1.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
+[[package]]
+name = "humansize"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
 [[package]]
 name = "humantime"
 version = "1.1.1"
@@ -2767,6 +2773,7 @@ dependencies = [
 "checksum hmac 0.6.3 (registry+https://github.com/rust-lang/crates.io-index)" = "733e1b3ac906631ca01ebb577e9bb0f5e37a454032b9036b5eaea4013ed6f99a"
 "checksum http 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)" = "24f58e8c2d8e886055c3ead7b28793e1455270b5fb39650984c224bc538ba581"
 "checksum httparse 1.3.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7b6288d7db100340ca12873fd4d08ad1b8f206a9457798dfb17c018a33fee540"
+"checksum humansize 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "b6cab2627acfc432780848602f3f558f7e9dd427352224b0d9324025796d2a5e"
 "checksum humantime 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0484fda3e7007f2a4a0d9c3a703ca38c71c54c55602ce4660c419fd32e188c9e"
 "checksum hyper 0.12.10 (registry+https://github.com/rust-lang/crates.io-index)" = "529d00e4c998cced1a15ffd53bbe203917b39ed6071281c16184ab0014ca6ff3"
 "checksum hyper-rustls 0.14.0 (registry+https://github.com/rust-lang/crates.io-index)" = "68f2aa6b1681795bf4da8063f718cd23145aa0c9a5143d9787b345aa60d38ee4"

Cargo.toml

@@ -20,6 +20,7 @@ chrono = "0.4.4"
 clap = "2.31"
 ctrlc = { version = "3.1", features = ["termination"] }
 cursive = "0.9.0"
+humansize = "1.1.0"
 daemonize = "0.3"
 serde = "1"
 serde_json = "1"

p2p/src/conn.rs

@@ -22,8 +22,9 @@
 
 use std::fs::File;
 use std::io::{self, Read, Write};
+use std::mem::size_of;
 use std::net::TcpStream;
-use std::sync::{mpsc, Arc, Mutex};
+use std::sync::{mpsc, Arc, RwLock};
 use std::{cmp, thread, time};
 
 use core::ser;
@@ -141,12 +142,11 @@ impl<'a> Response<'a> {
 	}
 }
 
-// TODO count sent and received
 pub struct Tracker {
 	/// Bytes we've sent.
-	pub sent_bytes: Arc<Mutex<u64>>,
+	pub sent_bytes: Arc<RwLock<u64>>,
 	/// Bytes we've received.
-	pub received_bytes: Arc<Mutex<u64>>,
+	pub received_bytes: Arc<RwLock<u64>>,
 	/// Channel to allow sending data through the connection
 	pub send_channel: mpsc::SyncSender<Vec<u8>>,
 	/// Channel to close the connection
@@ -161,7 +161,14 @@ impl Tracker {
 		T: ser::Writeable,
 	{
 		let buf = write_to_buf(body, msg_type);
+		let buf_len = buf.len();
 		self.send_channel.try_send(buf)?;
+
+		// Increase sent bytes counter
+		if let Ok(mut sent_bytes) = self.sent_bytes.write() {
+			*sent_bytes += buf_len as u64;
+		}
+
 		Ok(())
 	}
 }
@@ -177,14 +184,24 @@ where
 	let (close_tx, close_rx) = mpsc::channel();
 	let (error_tx, error_rx) = mpsc::channel();
 
+	// Counter of number of bytes received
+	let received_bytes = Arc::new(RwLock::new(0));
+
 	stream
 		.set_nonblocking(true)
 		.expect("Non-blocking IO not available.");
-	poll(stream, handler, send_rx, error_tx, close_rx);
+	poll(
+		stream,
+		handler,
+		send_rx,
+		error_tx,
+		close_rx,
+		received_bytes.clone(),
+	);
 
 	Tracker {
-		sent_bytes: Arc::new(Mutex::new(0)),
-		received_bytes: Arc::new(Mutex::new(0)),
+		sent_bytes: Arc::new(RwLock::new(0)),
+		received_bytes: received_bytes.clone(),
 		send_channel: send_tx,
 		close_channel: close_tx,
 		error_channel: error_rx,
@@ -197,6 +214,7 @@ fn poll<H>(
 	send_rx: mpsc::Receiver<Vec<u8>>,
 	error_tx: mpsc::Sender<Error>,
 	close_rx: mpsc::Receiver<()>,
+	received_bytes: Arc<RwLock<u64>>,
 ) where
 	H: MessageHandler,
 {
@@ -218,6 +236,13 @@ fn poll<H>(
 						msg.header.msg_type,
 						msg.header.msg_len
 					);
+
+					// Increase received bytes counter
+					if let Ok(mut received_bytes) = received_bytes.write() {
+						let header_size = size_of::<MsgHeader>() as u64;
+						*received_bytes += header_size + msg.header.msg_len;
+					}
+
 					if let Some(Some(resp)) = try_break!(error_tx, handler.consume(msg)) {
 						try_break!(error_tx, resp.write());
 					}

p2p/src/peer.rs

@@ -152,6 +152,28 @@ impl Peer {
 		}
 	}
 
+	/// Number of bytes sent to the peer
+	pub fn sent_bytes(&self) -> Option<u64> {
+		if let Some(ref tracker) = self.connection {
+			if let Ok(sent_bytes) = tracker.sent_bytes.read() {
+				return Some(*sent_bytes);
+			}
+		}
+
+		None
+	}
+
+	/// Number of bytes received from the peer
+	pub fn received_bytes(&self) -> Option<u64> {
+		if let Some(ref tracker) = self.connection {
+			if let Ok(received_bytes) = tracker.received_bytes.read() {
+				return Some(*received_bytes);
+			}
+		}
+
+		None
+	}
+
 	/// Set this peer status to banned
 	pub fn set_banned(&self) {
 		*self.state.write().unwrap() = State::Banned;

servers/src/common/stats.rs

@@ -148,6 +148,10 @@ pub struct PeerStats {
 	pub direction: String,
 	/// Last time we saw a ping/pong from this peer.
 	pub last_seen: DateTime<Utc>,
+	/// Number of bytes we've sent to the peer.
+	pub sent_bytes: Option<u64>,
+	/// Number of bytes we've received from the peer.
+	pub received_bytes: Option<u64>,
 }
 
 impl StratumStats {
@@ -181,6 +185,8 @@ impl PeerStats {
 			height: peer.info.height(),
 			direction: direction.to_string(),
 			last_seen: peer.info.last_seen(),
+			sent_bytes: peer.sent_bytes(),
+			received_bytes: peer.received_bytes(),
 		}
 	}
 }

src/bin/tui/mod.rs

@@ -14,6 +14,7 @@
 
 //! Grin TUI
 extern crate chrono;
+extern crate humansize;
 
 mod constants;
 mod menu;

src/bin/tui/peers.rs

@@ -19,6 +19,7 @@ use std::cmp::Ordering;
 use servers::{PeerStats, ServerStats};
 
 use chrono::prelude::*;
+use tui::humansize::{file_size_opts::CONVENTIONAL, FileSize};
 
 use cursive::direction::Orientation;
 use cursive::traits::{Boxable, Identifiable};
@@ -34,6 +35,7 @@ use tui::types::TUIStatusListener;
 enum PeerColumn {
 	Address,
 	State,
+	UsedBandwidth,
 	TotalDifficulty,
 	Direction,
 	Version,
@@ -44,6 +46,7 @@ impl PeerColumn {
 		match *self {
 			PeerColumn::Address => "Address",
 			PeerColumn::State => "State",
+			PeerColumn::UsedBandwidth => "Used bandwidth",
 			PeerColumn::Version => "Version",
 			PeerColumn::TotalDifficulty => "Total Difficulty",
 			PeerColumn::Direction => "Direction",
@@ -53,9 +56,27 @@ impl PeerColumn {
 
 impl TableViewItem<PeerColumn> for PeerStats {
 	fn to_column(&self, column: PeerColumn) -> String {
+		// Converts optional size to human readable size
+		fn size_to_string(size: Option<u64>) -> String {
+			if let Some(n) = size {
+				let size = n.file_size(CONVENTIONAL);
+				match size {
+					Ok(size) => size,
+					Err(_) => "-".to_string(),
+				}
+			} else {
+				"-".to_string()
+			}
+		}
+
 		match column {
 			PeerColumn::Address => self.addr.clone(),
 			PeerColumn::State => self.state.clone(),
+			PeerColumn::UsedBandwidth => format!(
+				"S: {}, R: {}",
+				size_to_string(self.sent_bytes),
+				size_to_string(self.received_bytes),
+			).to_string(),
 			PeerColumn::TotalDifficulty => format!(
 				"{} D @ {} H ({}s)",
 				self.total_difficulty,
@@ -71,9 +92,23 @@ impl TableViewItem<PeerColumn> for PeerStats {
 	where
 		Self: Sized,
 	{
+		// Compares used bandwidth of two peers
+		fn cmp_used_bandwidth(curr: &PeerStats, other: &PeerStats) -> Ordering {
+			let curr_recv_bytes = curr.received_bytes.unwrap_or(0);
+			let curr_sent_bytes = curr.sent_bytes.unwrap_or(0);
+			let other_recv_bytes = other.received_bytes.unwrap_or(0);
+			let other_sent_bytes = other.sent_bytes.unwrap_or(0);
+
+			let curr_sum = curr_recv_bytes + curr_sent_bytes;
+			let other_sum = other_recv_bytes + other_sent_bytes;
+
+			curr_sum.cmp(&other_sum)
+		};
+
 		match column {
 			PeerColumn::Address => self.addr.cmp(&other.addr),
 			PeerColumn::State => self.state.cmp(&other.state),
+			PeerColumn::UsedBandwidth => cmp_used_bandwidth(&self, &other),
 			PeerColumn::TotalDifficulty => self.total_difficulty.cmp(&other.total_difficulty),
 			PeerColumn::Direction => self.direction.cmp(&other.direction),
 			PeerColumn::Version => self.version.cmp(&other.version),
@@ -86,12 +121,14 @@ pub struct TUIPeerView;
 impl TUIStatusListener for TUIPeerView {
 	fn create() -> Box<View> {
 		let table_view = TableView::<PeerStats, PeerColumn>::new()
-			.column(PeerColumn::Address, "Address", |c| c.width_percent(20))
-			.column(PeerColumn::State, "State", |c| c.width_percent(20))
-			.column(PeerColumn::Direction, "Direction", |c| c.width_percent(20))
+			.column(PeerColumn::Address, "Address", |c| c.width_percent(16))
+			.column(PeerColumn::State, "State", |c| c.width_percent(16))
+			.column(PeerColumn::UsedBandwidth, "Used bandwidth", |c| {
+				c.width_percent(16)
+			}).column(PeerColumn::Direction, "Direction", |c| c.width_percent(16))
 			.column(PeerColumn::TotalDifficulty, "Total Difficulty", |c| {
-				c.width_percent(20)
-			}).column(PeerColumn::Version, "Version", |c| c.width_percent(20));
+				c.width_percent(16)
+			}).column(PeerColumn::Version, "Version", |c| c.width_percent(16));
 		let peer_status_view = BoxView::with_full_screen(
 			LinearLayout::new(Orientation::Vertical)
 				.child(