Use coarsetime to build an incoming traffic timestamp.

We need this for the circuit timeout estimator (#57).  It needs to
know "how recently have we got some incoming traffic", so that it
can tell whether a circuit has truly timed out, or whether the
entire network is down.

I'm implementing this with coarsetime, since we need to update these
in response to every single incoming cell, and we need the timestamp
operation to be _fast_.

(This reinstates an earlier commit, f30b2280, which I reverted
because we didn't need it at the time.)

Closes #179.

Cargo.lock

@@ -498,6 +498,18 @@ dependencies = [
  "vec_map",
 ]
 
+[[package]]
+name = "coarsetime"
+version = "0.1.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5b6ec6f6e80e839eb22bd61b18f19a8f2ae3f8bda9cf0fdce9dd96c9c5df8393"
+dependencies = [
+ "libc",
+ "once_cell",
+ "wasi 0.10.2+wasi-snapshot-preview1",
+ "wasm-bindgen",
+]
+
 [[package]]
 name = "concurrent-queue"
 version = "1.2.2"
@@ -2692,6 +2704,7 @@ dependencies = [
  "tor-llcrypto",
  "tor-netdir",
  "tor-persist",
+ "tor-proto",
  "tor-rtcompat",
  "tor-rtmock",
  "tor-units",
@@ -2806,6 +2819,7 @@ dependencies = [
  "asynchronous-codec",
  "bytes",
  "cipher",
+ "coarsetime",
  "crypto-mac",
  "digest",
  "event-listener",

crates/tor-proto/Cargo.toml

@@ -15,6 +15,7 @@ default = []
 hs = []
 ntor_v3 = []
 tokio = [ "tokio-crate", "tokio-util" ]
+traffic-timestamp = [ "coarsetime" ]
 
 [dependencies]
 tor-llcrypto = { path="../tor-llcrypto", version = "0.0.1"}
@@ -47,6 +48,8 @@ zeroize = "1.3.0"
 tokio-crate = { package = "tokio", version = "1.7.0", optional = true }
 tokio-util = { version = "0.6", features = ["compat"], optional = true }
 
+coarsetime = { version = "0.1.20", optional = true }
+
 [dev-dependencies]
 futures-await-test = "0.3.0"
 hex-literal = "0.3.1"

crates/tor-proto/src/channel/handshake.rs

@@ -392,6 +392,14 @@ impl<T: AsyncRead + AsyncWrite + Send + Unpin + 'static> VerifiedChannel<T> {
         Arc<super::Channel>,
         super::reactor::Reactor<stream::SplitStream<CellFrame<T>>>,
     )> {
+        // We treat a completed channel -- that is to say, one where the
+        // authentication is finished -- as incoming traffic.
+        //
+        // TODO: conceivably we should remember the time when we _got_ the
+        // final cell on the handshake, and update the channel completion
+        // time to be no earlier than _that_ timestamp.
+        crate::note_incoming_traffic();
+
         trace!("{}: Sending netinfo cell.", self.unique_id);
         let netinfo = msg::Netinfo::for_client(self.target_addr.as_ref().map(SocketAddr::ip));
         self.tls.send(netinfo.into()).await?;

crates/tor-proto/src/channel/reactor.rs

@@ -163,6 +163,7 @@ where
                     None => return Err(ReactorError::Shutdown), // the TLS connection closed.
                     Some(r) => r.map_err(Error::CellErr)?, // it's a cell.
                 };
+                crate::note_incoming_traffic();
                 self.handle_cell(item).await?;
 
             }

crates/tor-proto/src/lib.rs

@@ -127,3 +127,36 @@ type SecretBytes = zeroize::Zeroizing<Vec<u8>>;
 
 /// A Result type for this crate.
 pub type Result<T> = std::result::Result<T, Error>;
+
+/// Timestamp object that we update whenever we get incoming traffic.
+///
+/// Used to implement [`time_since_last_incoming_traffic`]
+#[cfg(feature = "traffic-timestamp")]
+static LAST_INCOMING_TRAFFIC: util::ts::Timestamp = util::ts::Timestamp::new();
+
+/// Called whenever we receive incoming traffic.
+///
+/// Used to implement [`time_since_last_incoming_traffic`]
+#[inline]
+pub(crate) fn note_incoming_traffic() {
+    #[cfg(feature = "traffic-timestamp")]
+    {
+        LAST_INCOMING_TRAFFIC.update();
+    }
+}
+
+/// Return the amount of time since we last received "incoming traffic".
+///
+/// Requires that the `traffic-timestamp` feature is enabled.
+///
+/// This is a global counter, and is subject to interference from
+/// other users of the `tor_proto`.  Its only permissible use is for
+/// checking how recently we have been definitely able to receive
+/// incoming traffic.
+///
+/// When enabled, this timestamp is updated whenever we receive a valid
+/// cell, and whenever we complete a channel handshake.
+#[cfg(feature = "traffic-timestamp")]
+pub fn time_since_last_incoming_traffic() -> std::time::Duration {
+    LAST_INCOMING_TRAFFIC.time_since_update().into()
+}

crates/tor-proto/src/util.rs

@@ -2,3 +2,5 @@
 
 pub(crate) mod ct;
 pub(crate) mod err;
+#[cfg(feature = "traffic-timestamp")]
+pub(crate) mod ts;

crates/tor-proto/src/util/ts.rs

@@ -0,0 +1,99 @@
+//! Implement a fast 'timestamp' for determining when an event last
+//! happened.
+
+use std::sync::atomic::{AtomicU64, Ordering};
+
+/// An object for determining when an event last happened.
+///
+/// Every `Timestamp` has internal mutability.  A timestamp can move
+/// forward in time, but never backwards.
+///
+/// Internally, it uses the `coarsetime` crate to represent times in a way
+/// that lets us do atomic updates.
+#[derive(Default, Debug)]
+pub(crate) struct Timestamp {
+    /// A timestamp (from `coarsetime`) describing when this timestamp
+    /// was last updated.
+    ///
+    /// I'd rather just use [`coarsetime::Instant`], but that doesn't have
+    /// an atomic form.
+    latest: AtomicU64,
+}
+impl Timestamp {
+    /// Construct a new timestamp that has never been updated.
+    pub(crate) const fn new() -> Self {
+        Timestamp {
+            latest: AtomicU64::new(0),
+        }
+    }
+    /// Update this timestamp to (at least) the current time.
+    pub(crate) fn update(&self) {
+        // TODO: Do we want to use 'Instant::recent() instead,' and
+        // add an updater thread?
+        self.update_to(coarsetime::Instant::now())
+    }
+    /// Update this timestamp to (at least) the time `now`.
+    #[inline]
+    pub(crate) fn update_to(&self, now: coarsetime::Instant) {
+        self.latest.fetch_max(now.as_ticks(), Ordering::Relaxed);
+    }
+
+    /// Return the time since `update` was last called.
+    ///
+    /// Returns 0 if update was never called.
+    pub(crate) fn time_since_update(&self) -> coarsetime::Duration {
+        self.time_since_update_at(coarsetime::Instant::now())
+    }
+
+    /// Return the time between the time when `update` was last
+    /// called, and the time `now`.
+    ///
+    /// Returns 0 if `update` was never called, or if `now` is before
+    /// that time.
+    #[inline]
+    pub(crate) fn time_since_update_at(&self, now: coarsetime::Instant) -> coarsetime::Duration {
+        let earlier = self.latest.load(Ordering::Relaxed);
+        let now = now.as_ticks();
+        if now >= earlier && earlier != 0 {
+            coarsetime::Duration::from_ticks(now - earlier)
+        } else {
+            coarsetime::Duration::from_secs(0)
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn timestamp() {
+        use coarsetime::{Duration, Instant};
+
+        let ts = Timestamp::new();
+
+        let zero = Duration::from_secs(0);
+        let one_sec = Duration::from_secs(1);
+
+        let first = Instant::now();
+        let in_a_bit = first + one_sec * 10;
+        let even_later = first + one_sec * 25;
+
+        assert_eq!(ts.time_since_update_at(first), zero);
+
+        ts.update_to(first);
+        assert_eq!(ts.time_since_update_at(first), zero);
+        assert_eq!(ts.time_since_update_at(in_a_bit), one_sec * 10);
+
+        ts.update_to(in_a_bit);
+        assert_eq!(ts.time_since_update_at(first), zero);
+        assert_eq!(ts.time_since_update_at(in_a_bit), zero);
+        assert_eq!(ts.time_since_update_at(even_later), one_sec * 15);
+
+        // Make sure we can't move backwards.
+        ts.update_to(first);
+        assert_eq!(ts.time_since_update_at(first), zero);
+        assert_eq!(ts.time_since_update_at(in_a_bit), zero);
+        assert_eq!(ts.time_since_update_at(even_later), one_sec * 15);
+    }
+}