perf: no more channels for UDP send/recv (#1579)

## Description

Do not use channels for the hot path of sending and receiving QUIC UDP
packets.

* Put `PeerMap` into a `Mutex` so that we can access it for the address
remapping directly in the `AsyncUdpSocket` `poll_send/poll_recv` methods
* `poll_recv` and `poll_send` the underlying UDP socket directly in
`MagicSock as AsyncUdpSocket` and only forward Derp, Disco and Stun
packages over channels

## Notes & open questions


## Change checklist

- [ ] Self-review.
- [ ] Documentation updates if relevant.
- [ ] Tests if relevant.

---------

Co-authored-by: dignifiedquire <me@dignifiedquire.com>
Co-authored-by: Diva M <divma@protonmail.com>

Cargo.toml

@@ -7,6 +7,7 @@ members = [
   "iroh-net",
   "iroh-sync",
   "iroh-test",
+  "iroh-net/bench"
 ]
 resolver = "2"
 

iroh-gossip/examples/chat.rs

@@ -117,6 +117,9 @@ async fn main() -> anyhow::Result<()> {
             let gossip_cell = gossip_cell.clone();
             let notify = notify.clone();
             Box::new(move |endpoints| {
+                if endpoints.is_empty() {
+                    return;
+                }
                 // send our updated endpoints to the gossip protocol to be sent as PeerAddr to peers
                 if let Some(gossip) = gossip_cell.get() {
                     gossip.update_endpoints(endpoints).ok();
@@ -129,16 +132,15 @@ async fn main() -> anyhow::Result<()> {
         .await?;
     println!("> our peer id: {}", endpoint.peer_id());
 
+    // wait for a first endpoint update so that we know about our endpoint addresses
+    notify.notified().await;
+
+    let my_addr = endpoint.my_addr().await?;
     // create the gossip protocol
-    let gossip = Gossip::from_endpoint(endpoint.clone(), Default::default());
+    let gossip = Gossip::from_endpoint(endpoint.clone(), Default::default(), &my_addr.info);
     // insert the gossip handle into the gossip cell to be used in the endpoint callbacks above
     gossip_cell.set(gossip.clone()).unwrap();
 
-    // wait for a first endpoint update so that we know about our endpoint addresses
-    notify.notified().await;
-    // forward our initial endpoints to the gossip protocol
-    gossip.update_endpoints(&endpoint.local_endpoints().await?)?;
-
     // print a ticket that includes our own peer id and endpoint addresses
     let ticket = {
         let me = endpoint.my_addr().await?;

iroh-gossip/src/net.rs

@@ -74,13 +74,16 @@ pub struct Gossip {
 
 impl Gossip {
     /// Spawn a gossip actor and get a handle for it
-    pub fn from_endpoint(endpoint: MagicEndpoint, config: proto::Config) -> Self {
+    pub fn from_endpoint(
+        endpoint: MagicEndpoint,
+        config: proto::Config,
+        my_addr: &AddrInfo,
+    ) -> Self {
         let peer_id = endpoint.peer_id();
         let dialer = Dialer::new(endpoint.clone());
-        let peer_data = Default::default();
         let state = proto::State::new(
             peer_id,
-            peer_data,
+            encode_peer_data(my_addr).unwrap(),
             config,
             rand::rngs::StdRng::from_entropy(),
         );
@@ -631,11 +634,17 @@ async fn connection_loop(
 }
 
 fn encode_peer_data(info: &AddrInfo) -> anyhow::Result<PeerData> {
-    Ok(PeerData::new(postcard::to_stdvec(info)?))
+    let bytes = postcard::to_stdvec(info)?;
+    anyhow::ensure!(!bytes.is_empty(), "encoding empty peer data: {:?}", info);
+    Ok(PeerData::new(bytes))
 }
 
 fn decode_peer_data(peer_data: &PeerData) -> anyhow::Result<AddrInfo> {
-    let info = postcard::from_bytes(peer_data.as_bytes())?;
+    let bytes = peer_data.as_bytes();
+    if bytes.is_empty() {
+        return Ok(AddrInfo::default());
+    }
+    let info = postcard::from_bytes(bytes)?;
     Ok(info)
 }
 
@@ -691,10 +700,22 @@ mod test {
         let ep1 = create_endpoint(derp_map.clone()).await.unwrap();
         let ep2 = create_endpoint(derp_map.clone()).await.unwrap();
         let ep3 = create_endpoint(derp_map.clone()).await.unwrap();
+        let addr1 = AddrInfo {
+            derp_region: Some(derp_region),
+            direct_addresses: Default::default(),
+        };
+        let addr2 = AddrInfo {
+            derp_region: Some(derp_region),
+            direct_addresses: Default::default(),
+        };
+        let addr3 = AddrInfo {
+            derp_region: Some(derp_region),
+            direct_addresses: Default::default(),
+        };
 
-        let go1 = Gossip::from_endpoint(ep1.clone(), Default::default());
-        let go2 = Gossip::from_endpoint(ep2.clone(), Default::default());
-        let go3 = Gossip::from_endpoint(ep3.clone(), Default::default());
+        let go1 = Gossip::from_endpoint(ep1.clone(), Default::default(), &addr1);
+        let go2 = Gossip::from_endpoint(ep2.clone(), Default::default(), &addr2);
+        let go3 = Gossip::from_endpoint(ep3.clone(), Default::default(), &addr3);
         debug!("peer1 {:?}", ep1.peer_id());
         debug!("peer2 {:?}", ep2.peer_id());
         debug!("peer3 {:?}", ep3.peer_id());
@@ -707,11 +728,14 @@ mod test {
             spawn(endpoint_loop(ep3.clone(), go2.clone(), cancel.clone())),
         ];
 
+        debug!("----- adding peers  ----- ");
         let topic: TopicId = blake3::hash(b"foobar").into();
         // share info that pi1 is on the same derp_region
         let addr1 = PeerAddr::new(pi1).with_derp_region(derp_region);
         ep2.add_peer_addr(addr1.clone()).await.unwrap();
         ep3.add_peer_addr(addr1).await.unwrap();
+
+        debug!("----- joining  ----- ");
         // join the topics and wait for the connection to succeed
         go1.join(topic, vec![]).await.unwrap();
         go2.join(topic, vec![pi1]).await.unwrap().await.unwrap();

iroh-net/Cargo.toml

@@ -35,6 +35,7 @@ libc = "0.2.139"
 num_enum = "0.7"
 once_cell = "1.18.0"
 os_info = "3.6.0"
+parking_lot = "0.12.1"
 postcard = { version = "1", default-features = false, features = ["alloc", "use-std", "experimental-derive"] }
 rand = "0.8"
 rand_core = "0.6.4"

iroh-net/bench/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "iroh-net-bench"
+version = "0.1.0"
+edition = "2021"
+license = "MIT OR Apache-2.0"
+publish = false
+
+[dependencies]
+anyhow = "1.0.22"
+bytes = "1"
+hdrhistogram = { version = "7.2", default-features = false }
+iroh-net = { path = ".." }
+quinn = "0.10"
+rcgen = "0.11.1"
+rustls = { version = "0.21.0", default-features = false, features = ["quic"] }
+clap = { version = "4", features = ["derive"] }
+tokio = { version = "1.0.1", features = ["rt", "sync"] }
+tracing = "0.1.10"
+tracing-subscriber = { version = "0.3.0", default-features = false, features = ["env-filter", "fmt", "ansi", "time", "local-time"] }