clippy: Disallow lock and instant types from std (#1458)

We use `parking_lot` locks throughout our code. This change disallows the
introduction of std::sync's locks.

This change also enforces the use of `tokio::time::Instant`, which
allows for a mockable time source in tests.

Signed-off-by: Oliver Gould <[email protected]>
(cherry picked from commit 8aa474e)
Signed-off-by: Oliver Gould <[email protected]>

Author: olix0r

.clippy.toml

@@ -1,13 +1,22 @@
 type-complexity-threshold = 500
 disallowed-methods = [
-    # mutating environment variables in a multi-threaded context can
+    # Mutating environment variables in a multi-threaded context can
     # cause data races.
     # see https://github.com/rust-lang/rust/issues/90308 for details.
     "std::env::set_var",
     "std::env::remove_var",
 
-    # Avoid instances of https://github.com/rust-lang/rust/issues/86470
-    "std::time::Instant::duration_since",
-    "std::time::Instant::elapsed",
-    # Clippy doesn't let us ban std::time::Instant::sub, but it knows what it did.
+    # Avoid instances of https://github.com/rust-lang/rust/issues/86470 until tokio/tokio#4461 is
+    # available.
+    "tokio::time::Instant::duration_since",
+    "tokio::time::Instant::elapsed",
+    # Clippy doesn't let us ban tokio::time::Instant::sub, but it knows what it did.
+]
+disallowed-types = [
+    # Use parking_lot instead.
+    "std::sync::Mutex",
+    "std::sync::RwLock",
+
+    # Use tokio::time::Instant instead.
+    "std::time::Instant",
 ]

Cargo.lock

@@ -786,6 +786,7 @@ dependencies = [
  "linkerd-metrics",
  "linkerd-tracing",
  "linkerd2-proxy-api",
+ "parking_lot 0.12.0",
  "regex",
  "socket2 0.4.4",
  "tokio",
@@ -835,6 +836,7 @@ dependencies = [
  "linkerd-app-core",
  "linkerd-identity",
  "linkerd-io",
+ "parking_lot 0.12.0",
  "regex",
  "thiserror",
  "tokio",
@@ -982,6 +984,7 @@ dependencies = [
  "linkerd-stack",
  "parking_lot 0.12.0",
  "pin-project",
+ "tokio",
  "tower",
  "tracing",
 ]
@@ -1326,6 +1329,7 @@ dependencies = [
  "futures",
  "linkerd-error",
  "linkerd-tracing",
+ "parking_lot 0.12.0",
  "pin-project",
  "thiserror",
  "tokio",
@@ -1464,6 +1468,7 @@ dependencies = [
  "linkerd-stack",
  "parking_lot 0.12.0",
  "pin-project",
+ "tokio",
  "tracing",
 ]
 

hyper-balance/src/lib.rs

@@ -1,4 +1,9 @@
-#![deny(warnings, rust_2018_idioms, clippy::disallowed_method)]
+#![deny(
+    warnings,
+    rust_2018_idioms,
+    clippy::disallowed_method,
+    clippy::disallowed_type
+)]
 #![forbid(unsafe_code)]
 
 use hyper::body::HttpBody;

linkerd/addr/src/lib.rs

@@ -1,4 +1,9 @@
-#![deny(warnings, rust_2018_idioms, clippy::disallowed_method)]
+#![deny(
+    warnings,
+    rust_2018_idioms,
+    clippy::disallowed_method,
+    clippy::disallowed_type
+)]
 #![forbid(unsafe_code)]
 use linkerd_dns_name::Name;
 use std::{

linkerd/app/admin/src/lib.rs

@@ -1,4 +1,9 @@
-#![deny(warnings, rust_2018_idioms, clippy::disallowed_method)]
+#![deny(
+    warnings,
+    rust_2018_idioms,
+    clippy::disallowed_method,
+    clippy::disallowed_type
+)]
 #![forbid(unsafe_code)]
 
 mod server;

linkerd/app/core/src/lib.rs

@@ -7,7 +7,12 @@
 //! - Tap
 //! - Metric labeling
 
-#![deny(warnings, rust_2018_idioms, clippy::disallowed_method)]
+#![deny(
+    warnings,
+    rust_2018_idioms,
+    clippy::disallowed_method,
+    clippy::disallowed_type
+)]
 #![forbid(unsafe_code)]
 
 pub use drain;

linkerd/app/gateway/src/lib.rs

@@ -1,4 +1,9 @@
-#![deny(warnings, rust_2018_idioms, clippy::disallowed_method)]
+#![deny(
+    warnings,
+    rust_2018_idioms,
+    clippy::disallowed_method,
+    clippy::disallowed_type
+)]
 #![forbid(unsafe_code)]
 
 mod gateway;

linkerd/app/inbound/src/lib.rs

@@ -3,7 +3,12 @@
 //! The inbound proxy is responsible for terminating traffic from other network
 //! endpoints inbound to the local application.
 
-#![deny(warnings, rust_2018_idioms, clippy::disallowed_method)]
+#![deny(
+    warnings,
+    rust_2018_idioms,
+    clippy::disallowed_method,
+    clippy::disallowed_type
+)]
 #![forbid(unsafe_code)]
 
 mod accept;

linkerd/app/integration/Cargo.toml

@@ -29,6 +29,7 @@ linkerd-metrics = { path = "../../metrics", features = ["test_util"] }
 linkerd2-proxy-api = { version = "0.3", features = ["destination", "arbitrary"] }
 linkerd-app-test = { path = "../test" }
 linkerd-tracing = { path = "../../tracing" }
+parking_lot = "0.12"
 regex = "1"
 socket2 = "0.4"
 tokio = { version = "1", features = ["io-util", "net", "rt", "macros"] }

linkerd/app/integration/src/client.rs

@@ -1,7 +1,7 @@
 use super::*;
 use linkerd_app_core::proxy::http::trace;
-use std::io;
-use std::sync::{Arc, Mutex};
+use parking_lot::Mutex;
+use std::{io, sync::Arc};
 use tokio::net::TcpStream;
 use tokio::sync::{mpsc, oneshot};
 use tokio::task::JoinHandle;
@@ -315,7 +315,6 @@ impl tower::Service<hyper::Uri> for Conn {
         let running = self
             .running
             .lock()
-            .unwrap()
             .take()
             .expect("test client cannot connect twice");
         Box::pin(async move {

linkerd/app/integration/src/controller.rs

@@ -3,10 +3,11 @@ use super::*;
 use linkerd2_proxy_api::destination as pb;
 use linkerd2_proxy_api::net;
 use linkerd_app_core::proxy::http::trace;
+use parking_lot::Mutex;
 use std::collections::{HashMap, VecDeque};
 use std::net::IpAddr;
 use std::ops::{Bound, RangeBounds};
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::UnboundedReceiverStream;
 use tonic as grpc;
@@ -86,7 +87,6 @@ impl Controller {
         };
         self.expect_dst_calls
             .lock()
-            .unwrap()
             .push_back(Dst::Call(dst, Ok(rx)));
         DstSender(tx)
     }
@@ -108,12 +108,11 @@ impl Controller {
         };
         self.expect_dst_calls
             .lock()
-            .unwrap()
             .push_back(Dst::Call(dst, Err(status)));
     }
 
     pub fn no_more_destinations(&self) {
-        self.expect_dst_calls.lock().unwrap().push_back(Dst::Done);
+        self.expect_dst_calls.lock().push_back(Dst::Done);
     }
 
     pub async fn delay_listen<F>(self, f: F) -> Listening
@@ -148,10 +147,7 @@ impl Controller {
             path,
             ..Default::default()
         };
-        self.expect_profile_calls
-            .lock()
-            .unwrap()
-            .push_back((dst, rx));
+        self.expect_profile_calls.lock().push_back((dst, rx));
         ProfileSender(tx)
     }
 
@@ -232,52 +228,51 @@ impl pb::destination_server::Destination for Controller {
         let _e = span.enter();
         tracing::debug!(request = ?req.get_ref(), "received");
 
-        if let Ok(mut calls) = self.expect_dst_calls.lock() {
-            if self.unordered {
-                let mut calls_next: VecDeque<Dst> = VecDeque::new();
-                if calls.is_empty() {
-                    tracing::warn!("calls exhausted");
-                }
-                while let Some(call) = calls.pop_front() {
-                    if let Dst::Call(dst, updates) = call {
-                        tracing::debug!(?dst, "checking");
-                        if &dst == req.get_ref() {
-                            tracing::info!(?dst, ?updates, "found request");
-                            calls_next.extend(calls.drain(..));
-                            *calls = calls_next;
-                            return updates.map(grpc::Response::new);
-                        }
-
-                        calls_next.push_back(Dst::Call(dst, updates));
-                    }
-                }
-
-                tracing::warn!(remaining = calls_next.len(), "missed");
-                *calls = calls_next;
-                return Err(grpc_unexpected_request());
+        let mut calls = self.expect_dst_calls.lock();
+        if self.unordered {
+            let mut calls_next: VecDeque<Dst> = VecDeque::new();
+            if calls.is_empty() {
+                tracing::warn!("calls exhausted");
             }
-
-            match calls.pop_front() {
-                Some(Dst::Call(dst, updates)) => {
-                    tracing::debug!(?dst, "checking next call");
+            while let Some(call) = calls.pop_front() {
+                if let Dst::Call(dst, updates) = call {
+                    tracing::debug!(?dst, "checking");
                     if &dst == req.get_ref() {
                         tracing::info!(?dst, ?updates, "found request");
+                        calls_next.extend(calls.drain(..));
+                        *calls = calls_next;
                         return updates.map(grpc::Response::new);
                     }
 
-                    tracing::warn!(?dst, ?updates, "request does not match");
-                    let msg = format!(
-                        "expected get call for {:?} but got get call for {:?}",
-                        dst, req
-                    );
-                    calls.push_front(Dst::Call(dst, updates));
-                    return Err(grpc::Status::new(grpc::Code::Unavailable, msg));
+                    calls_next.push_back(Dst::Call(dst, updates));
                 }
-                Some(Dst::Done) => {
-                    panic!("unit test controller expects no more Destination.Get calls")
+            }
+
+            tracing::warn!(remaining = calls_next.len(), "missed");
+            *calls = calls_next;
+            return Err(grpc_unexpected_request());
+        }
+
+        match calls.pop_front() {
+            Some(Dst::Call(dst, updates)) => {
+                tracing::debug!(?dst, "checking next call");
+                if &dst == req.get_ref() {
+                    tracing::info!(?dst, ?updates, "found request");
+                    return updates.map(grpc::Response::new);
                 }
-                _ => {}
+
+                tracing::warn!(?dst, ?updates, "request does not match");
+                let msg = format!(
+                    "expected get call for {:?} but got get call for {:?}",
+                    dst, req
+                );
+                calls.push_front(Dst::Call(dst, updates));
+                return Err(grpc::Status::new(grpc::Code::Unavailable, msg));
             }
+            Some(Dst::Done) => {
+                panic!("unit test controller expects no more Destination.Get calls")
+            }
+            _ => {}
         }
 
         Err(grpc_no_results())
@@ -296,18 +291,17 @@ impl pb::destination_server::Destination for Controller {
         );
         let _e = span.enter();
         tracing::debug!(request = ?req.get_ref(), "received");
-        if let Ok(mut calls) = self.expect_profile_calls.lock() {
-            if let Some((dst, profile)) = calls.pop_front() {
-                tracing::debug!(?dst, "checking next call");
-                if &dst == req.get_ref() {
-                    tracing::info!(?dst, ?profile, "found request");
-                    return Ok(grpc::Response::new(Box::pin(profile)));
-                }
-
-                tracing::warn!(?dst, ?profile, "request does not match");
-                calls.push_front((dst, profile));
-                return Err(grpc_unexpected_request());
+        let mut calls = self.expect_profile_calls.lock();
+        if let Some((dst, profile)) = calls.pop_front() {
+            tracing::debug!(?dst, "checking next call");
+            if &dst == req.get_ref() {
+                tracing::info!(?dst, ?profile, "found request");
+                return Ok(grpc::Response::new(Box::pin(profile)));
             }
+
+            tracing::warn!(?dst, ?profile, "request does not match");
+            calls.push_front((dst, profile));
+            return Err(grpc_unexpected_request());
         }
 
         Err(grpc_no_results())

linkerd/app/integration/src/identity.rs

@@ -1,9 +1,10 @@
 use super::*;
+use parking_lot::Mutex;
 use std::{
     collections::VecDeque,
     fs, io,
     path::{Path, PathBuf},
-    sync::{Arc, Mutex},
+    sync::Arc,
     time::{Duration, SystemTime},
 };
 
@@ -198,7 +199,7 @@ impl Controller {
                 .map_err(|e| grpc::Status::new(grpc::Code::Internal, format!("{}", e)));
             Box::pin(fut)
         });
-        self.expect_calls.lock().unwrap().push_back(func);
+        self.expect_calls.lock().push_back(func);
         self
     }
 
@@ -222,7 +223,6 @@ impl pb::identity_server::Identity for Controller {
         let f = self
             .expect_calls
             .lock()
-            .unwrap()
             .pop_front()
             .map(|mut f| f(req.into_inner()));
         if let Some(f) = f {

linkerd/app/integration/src/lib.rs

@@ -1,6 +1,11 @@
 //! Shared infrastructure for integration tests
 
-#![deny(warnings, rust_2018_idioms, clippy::disallowed_method)]
+#![deny(
+    warnings,
+    rust_2018_idioms,
+    clippy::disallowed_method,
+    clippy::disallowed_type
+)]
 #![forbid(unsafe_code)]
 
 mod test_env;
@@ -66,8 +71,8 @@ macro_rules! assert_eventually {
     ($cond:expr, retries: $retries:expr, $($arg:tt)+) => {
         {
             use std::{env, u64};
-            use std::time::{Instant, Duration};
             use std::str::FromStr;
+            use tokio::time::{Instant, Duration};
             use tracing::Instrument as _;
             // TODO: don't do this *every* time eventually is called (lazy_static?)
             let patience = env::var($crate::ENV_TEST_PATIENCE_MS).ok()

linkerd/app/integration/src/metrics.rs

@@ -195,8 +195,8 @@ impl MetricMatch {
     #[track_caller]
     pub async fn assert_in(&self, client: &crate::client::Client) {
         use std::str::FromStr;
-        use std::time::{Duration, Instant};
         use std::{env, u64};
+        use tokio::time::{Duration, Instant};
         use tracing::Instrument as _;
         const MAX_RETRIES: usize = 5;
         // TODO: don't do this *every* time eventually is called (lazy_static?)