net: add timestamp to msg log. util: create unix timestamp tool.

error: error handling for timestamp

src/error.rs

@@ -300,6 +300,9 @@ pub enum Error {
     #[error("Unsupported OS")]
     UnsupportedOS,
 
+    #[error("System clock went backwards")]
+    BackwardsTime(std::time::SystemTimeError),
+
     // ==============================================
     // Wrappers for other error types in this library
     // ==============================================
@@ -397,6 +400,12 @@ impl From<std::io::Error> for Error {
     }
 }
 
+impl From<std::time::SystemTimeError> for Error {
+    fn from(err: std::time::SystemTimeError) -> Self {
+        Self::BackwardsTime(err)
+    }
+}
+
 impl From<std::convert::Infallible> for Error {
     fn from(err: std::convert::Infallible) -> Self {
         Self::InfallibleError(err.to_string())

src/net/channel.rs

@@ -1,4 +1,5 @@
 use async_std::sync::{Arc, Mutex};
+use std::sync::atomic::{AtomicBool, Ordering};
 
 use futures::{
     io::{ReadHalf, WriteHalf},
@@ -12,6 +13,7 @@ use url::Url;
 
 use crate::{
     system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
+    util::time,
     Error, Result,
 };
 
@@ -29,7 +31,7 @@ struct ChannelInfo {
     last_msg: String,
     last_status: String,
     // Message log which is cleared on querying get_info
-    log: Mutex<Vec<(String, String)>>,
+    log: Mutex<Vec<(u64, String, String)>>,
 }
 
 impl ChannelInfo {
@@ -62,7 +64,7 @@ pub struct Channel {
     message_subsystem: MessageSubsystem,
     stop_subscriber: SubscriberPtr<Error>,
     receive_task: StoppableTaskPtr,
-    stopped: Mutex<bool>,
+    stopped: AtomicBool,
     info: Mutex<ChannelInfo>,
 }
 
@@ -85,7 +87,7 @@ impl Channel {
             message_subsystem,
             stop_subscriber: Subscriber::new(),
             receive_task: StoppableTask::new(),
-            stopped: Mutex::new(false),
+            stopped: AtomicBool::new(false),
             info: Mutex::new(ChannelInfo::new()),
         })
     }
@@ -114,15 +116,13 @@ impl Channel {
     /// the channel has been closed.
     pub async fn stop(&self) {
         debug!(target: "net", "Channel::stop() [START, address={}]", self.address());
-        let mut stopped = self.stopped.lock().await;
-        if !*stopped {
-            *stopped = true;
-            self.stop_subscriber.notify(Error::ChannelStopped).await;
-            self.receive_task.stop().await;
-            self.message_subsystem.trigger_error(Error::ChannelStopped).await;
-            debug!(target: "net", "Channel::stop() [END, address={}]", self.address());
-        }
-        drop(stopped);
+        assert!(!self.stopped.load(Ordering::Relaxed));
+        // Changes memory ordering to relaxed. We don't need strict thread locking here.
+        self.stopped.store(false, Ordering::Relaxed);
+        self.stop_subscriber.notify(Error::ChannelStopped).await;
+        self.receive_task.stop().await;
+        self.message_subsystem.trigger_error(Error::ChannelStopped).await;
+        debug!(target: "net", "Channel::stop() [END, address={}]", self.address());
     }
 
     /// Creates a subscription to a stopped signal.
@@ -150,13 +150,8 @@ impl Channel {
          M::name(),
          self.address()
         );
-
-        // TODO can we use RwLock here instead of Mutex
-        {
-            let stopped = *self.stopped.lock().await;
-            if stopped {
-                return Err(Error::ChannelStopped)
-            }
+        if self.stopped.load(Ordering::Relaxed) {
+            return Err(Error::ChannelStopped)
         }
 
         // Catch failure and stop channel, return a net error
@@ -191,10 +186,11 @@ impl Channel {
         let mut payload = Vec::new();
         message.encode(&mut payload)?;
         let packet = message::Packet { command: String::from(M::name()), payload };
+        let time = time::unix_timestamp()?;
 
         {
             let info = &mut *self.info.lock().await;
-            info.log.lock().await.push(("send".to_string(), packet.command.clone()));
+            info.log.lock().await.push((time, "send".to_string(), packet.command.clone()));
         }
 
         let stream = &mut *self.writer.lock().await;
@@ -276,7 +272,8 @@ impl Channel {
                 let info = &mut *self.info.lock().await;
                 info.last_msg = packet.command.clone();
                 info.last_status = "recv".to_string();
-                info.log.lock().await.push(("recv".to_string(), packet.command.clone()));
+                let time = time::unix_timestamp()?;
+                info.log.lock().await.push((time, "recv".to_string(), packet.command.clone()));
             }
 
             // Send result to our subscribers

src/util/mod.rs

@@ -17,4 +17,4 @@ pub use async_util::sleep;
 pub use net_name::NetworkName;
 pub use parse::{decode_base10, encode_base10};
 pub use path::{expand_path, join_config_path, load_keypair_to_str};
-pub use time::{check_clock, Timestamp};
+pub use time::{check_clock, unix_timestamp, Timestamp};

src/util/time.rs

@@ -1,3 +1,5 @@
+use std::time::SystemTime;
+
 use async_std::{
     io::{ReadExt, WriteExt},
     net::TcpStream,
@@ -158,3 +160,7 @@ pub fn timestamp_to_date(timestamp: i64, dt: &str) -> String {
         _ => "".to_string(),
     }
 }
+
+pub fn unix_timestamp() -> Result<u64> {
+    Ok(SystemTime::now().duration_since(std::time::UNIX_EPOCH)?.as_secs())
+}