use mg-common macros

Author: jgallagher

Cargo.lock

@@ -5,6 +5,7 @@ edition = "2024"
 
 [dependencies]
 bfd.workspace = true
+mg-common.workspace = true
 mg-api-types.workspace = true
 rdb.workspace = true
 slog.workspace = true

bfd-async/Cargo.toml

@@ -5,6 +5,8 @@
 use crate::AddPeerError;
 use bfd::SessionCounters;
 use bfd::packet;
+use mg_common::read_lock;
+use mg_common::write_lock;
 use slog::Logger;
 use slog::warn;
 use slog_error_chain::InlineErrorChain;
@@ -228,7 +230,7 @@ impl Listener {
     }
 
     fn remove_peer(&self, peer: IpAddr) -> ListenerRemovePeerResult {
-        let mut sessions = self.sessions.write().unwrap();
+        let mut sessions = write_lock!(self.sessions);
         if sessions.remove(&peer).is_some() {
             if sessions.is_empty() {
                 ListenerRemovePeerResult::RemovedNowEmpty
@@ -245,7 +247,7 @@ impl Listener {
         peer: IpAddr,
         session: SessionHandle,
     ) -> Result<(), AddPeerError> {
-        let mut sessions = self.sessions.write().unwrap();
+        let mut sessions = write_lock!(self.sessions);
         match sessions.entry(peer) {
             hash_map::Entry::Occupied(_) => Err(AddPeerError::PeerExists(peer)),
             hash_map::Entry::Vacant(entry) => {
@@ -346,7 +348,7 @@ impl ListenerTask {
 
             // Do we expect traffic from this peer address?
             let Some(session) =
-                self.sessions.read().unwrap().get(&peer.ip()).cloned()
+                read_lock!(self.sessions).get(&peer.ip()).cloned()
             else {
                 warn!(
                     self.log, "unknown peer; dropping";

bfd-async/src/dispatcher.rs

@@ -23,6 +23,7 @@ use crate::wait_for_condition;
 use bfd::DEFAULT_DETECT_MULTIPLIER;
 use mg_api_types::bfd::BfdPeerState;
 use mg_api_types::bfd::SessionMode;
+use mg_common::lock;
 use slog::Logger;
 use std::net::IpAddr;
 use std::net::Ipv4Addr;
@@ -78,10 +79,7 @@ impl ListenerBackend for PreBoundSocketBackend {
         sessions: SharedSessions,
         log: Logger,
     ) -> Result<JoinHandle<()>, AddPeerError> {
-        let socket = self
-            .socket
-            .lock()
-            .unwrap()
+        let socket = lock!(self.socket)
             .take()
             .expect("PreBoundSocketBackend::spawn called more than once");
         let listen_task = ListenerTask::new(socket, sessions, log);

bfd-async/src/dispatcher/end_to_end.rs

@@ -13,6 +13,9 @@ use super::Dispatcher;
 use super::ListenerBackend;
 use super::SharedSessions;
 use crate::AddPeerError;
+use mg_common::parse;
+use mg_common::read_lock;
+use mg_common::sockaddr;
 use proptest::collection::vec;
 use proptest::prelude::*;
 use proptest::sample::select;
@@ -63,9 +66,9 @@ impl ListenerBackend for FakeBackend {
 // IPs/ports are arbitrary.
 fn arb_listen_addr() -> impl Strategy<Value = SocketAddr> {
     let addrs = vec![
-        "127.0.0.1:3784".parse().unwrap(),
-        "127.0.0.1:4784".parse().unwrap(),
-        "192.0.2.1:3784".parse().unwrap(),
+        sockaddr!("127.0.0.1:3784"),
+        sockaddr!("127.0.0.1:4784"),
+        sockaddr!("192.0.2.1:3784"),
     ];
     select(addrs)
 }
@@ -146,7 +149,7 @@ fn check_invariants(dispatcher: &Dispatcher, model: &Model) {
     // that address, and no listener is empty.
     for (addr, listener) in &dispatcher.listeners {
         let actual: BTreeSet<IpAddr> =
-            listener.sessions.read().unwrap().keys().copied().collect();
+            read_lock!(listener.sessions).keys().copied().collect();
         let expected: BTreeSet<IpAddr> = model
             .peer_to_addr
             .iter()

bfd-async/src/dispatcher/proptests.rs

@@ -10,6 +10,9 @@ use super::ListenerBackend;
 use super::ListenerTask;
 use bfd::SessionCounters;
 use bfd::packet::Control;
+use mg_common::ip;
+use mg_common::lock;
+use mg_common::parse;
 use slog::Discard;
 use slog::Logger;
 use slog::o;
@@ -51,10 +54,7 @@ impl TestBackend {
     }
 
     fn expect_bound_address(&self) -> SocketAddr {
-        self.bound_address
-            .lock()
-            .unwrap()
-            .expect("socket has been bound")
+        lock!(self.bound_address).expect("socket has been bound")
     }
 
     fn was_listener_task_dropped(&self) -> bool {
@@ -71,7 +71,7 @@ impl ListenerBackend for TestBackend {
     ) -> Result<tokio::task::JoinHandle<()>, crate::AddPeerError> {
         assert_eq!(listen_addr, LISTEN_ADDR);
 
-        let mut bound_address = self.bound_address.lock().unwrap();
+        let mut bound_address = lock!(self.bound_address);
         assert!(
             bound_address.is_none(),
             "TestBackend can only spawn one listening socket"
@@ -154,7 +154,7 @@ async fn drops_packet_from_unregistered_peer() {
 
     // Register a peer that is *not* the loopback source address our client
     // sends from, so the listener sees a mismatched source IP.
-    let peer: IpAddr = "192.0.2.1".parse().unwrap();
+    let peer = ip!("192.0.2.1");
 
     let mut rx = dispatcher
         .ensure(LISTEN_ADDR, peer, Arc::default(), &log)

bfd-async/src/dispatcher/tests.rs

@@ -3,6 +3,8 @@
 // file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
 use super::*;
+use mg_common::parse;
+use mg_common::sockaddr;
 use slog::Discard;
 use slog::o;
 use std::net::Ipv4Addr;
@@ -12,7 +14,7 @@ use tokio::time::timeout;
 
 #[tokio::test]
 async fn egress_socket_sets_ttl_v4() {
-    let sk = bind_egress_socket("127.0.0.1:0".parse().unwrap())
+    let sk = bind_egress_socket(sockaddr!("127.0.0.1:0"))
         .await
         .expect("created v4 egress socket");
     let sock = Socket::from(sk.into_std().expect("converted socket"));
@@ -21,7 +23,7 @@ async fn egress_socket_sets_ttl_v4() {
 
 #[tokio::test]
 async fn egress_socket_sets_hop_limit_v6() {
-    let sk = bind_egress_socket("[::1]:0".parse().unwrap())
+    let sk = bind_egress_socket(sockaddr!("[::1]:0"))
         .await
         .expect("created v6 egress socket");
     let sock = Socket::from(sk.into_std().expect("converted socket"));
@@ -145,7 +147,7 @@ async fn binds_ports_in_expected_range() {
 #[tokio::test(flavor = "multi_thread")]
 async fn exits_when_channel_closed() {
     let (tx, _counters, handle) =
-        spawn_egress("127.0.0.1:1".parse().unwrap(), Arc::default());
+        spawn_egress(sockaddr!("127.0.0.1:1"), Arc::default());
     drop(tx);
     timeout(Duration::from_secs(5), handle)
         .await

bfd-async/src/egress/tests.rs

@@ -100,6 +100,7 @@ impl RibTask {
 mod tests {
     use super::*;
     use crate::wait_for_condition;
+    use mg_common::lock;
     use std::net::Ipv4Addr;
     use std::sync::Arc;
     use std::sync::Mutex;
@@ -115,13 +116,13 @@ mod tests {
 
     impl NexthopSink for RecordingSink {
         fn set_nexthop_shutdown(&self, nexthop: IpAddr, shutdown: bool) {
-            self.0.lock().unwrap().push((nexthop, shutdown));
+            lock!(self.0).push((nexthop, shutdown));
         }
     }
 
     impl RecordingSink {
         fn writes(&self) -> Vec<(IpAddr, bool)> {
-            self.0.lock().unwrap().clone()
+            lock!(self.0).clone()
         }
     }
 

bfd-async/src/rib.rs

@@ -14,6 +14,8 @@ use super::*;
 use crate::wait_for_condition;
 use bfd::DEFAULT_DETECT_MULTIPLIER;
 use bfd::packet::Control;
+use mg_common::parse;
+use mg_common::sockaddr;
 use slog::Discard;
 use slog::o;
 use tokio::time::timeout;
@@ -49,7 +51,7 @@ fn spawn_driver(required_rx: Duration) -> Harness {
         DriverTask {
             sm,
             counters: Arc::clone(&counters),
-            remote_addr: "127.0.0.1:3784".parse().unwrap(),
+            remote_addr: sockaddr!("127.0.0.1:3784"),
             listener_rx,
             egress_tx,
             state_tx,