Being refactored: ECN additions

.github/workflows/qns.yml

@@ -2,7 +2,8 @@ name: QUIC Network Simulator
 
 on:
   schedule:
-    - cron: '42 3 * * *' # Runs at 03:42 UTC (m and h chosen arbitrarily) every day.
+    - cron: '42 3 * * 2,5' # Runs at 03:42 UTC (m and h chosen arbitrarily) twice a week.
+  workflow_dispatch:
   pull_request:
     branch: ["main"]
     paths:

neqo-client/Cargo.toml

@@ -11,7 +11,7 @@ license = "MIT OR Apache-2.0"
 [dependencies]
 neqo-crypto = { path = "./../neqo-crypto" }
 neqo-transport = { path = "./../neqo-transport" }
-neqo-common = { path="./../neqo-common" }
+neqo-common = { path="./../neqo-common", features = ["socket"] }
 neqo-http3 = { path = "./../neqo-http3" }
 neqo-qpack = { path = "./../neqo-qpack" }
 structopt = "0.3.7"

neqo-client/src/main.rs

@@ -11,7 +11,10 @@ use qlog::{events::EventImportance, streamer::QlogStreamer};
 
 use mio::{net::UdpSocket, Events, Poll, PollOpt, Ready, Token};
 
-use neqo_common::{self as common, event::Provider, hex, qlog::NeqoQlog, Datagram, Role};
+use neqo_common::{
+    self as common, bind, emit_datagram, event::Provider, hex, qlog::NeqoQlog, recv_datagram,
+    Datagram, Role,
+};
 use neqo_crypto::{
     constants::{TLS_AES_128_GCM_SHA256, TLS_AES_256_GCM_SHA384, TLS_CHACHA20_POLY1305_SHA256},
     init, AuthenticationStatus, Cipher, ResumptionToken,
@@ -345,14 +348,6 @@ impl QuicParameters {
     }
 }
 
-fn emit_datagram(socket: &mio::net::UdpSocket, d: Datagram) -> io::Result<()> {
-    let sent = socket.send_to(&d[..], &d.destination())?;
-    if sent != d.len() {
-        eprintln!("Unable to send all {} bytes of datagram", d.len());
-    }
-    Ok(())
-}
-
 fn get_output_file(
     url: &Url,
     output_dir: &Option<PathBuf>,
@@ -413,7 +408,9 @@ fn process_loop(
 
         let mut datagrams: Vec<Datagram> = Vec::new();
         'read: loop {
-            match socket.recv_from(&mut buf[..]) {
+            let mut tos = 0;
+            let mut ttl = 0;
+            match recv_datagram(socket, &mut buf[..], &mut tos, &mut ttl) {
                 Err(ref err)
                     if err.kind() == ErrorKind::WouldBlock
                         || err.kind() == ErrorKind::Interrupted =>
@@ -430,7 +427,7 @@ fn process_loop(
                         break 'read;
                     }
                     if sz > 0 {
-                        let d = Datagram::new(remote, *local_addr, &buf[..sz]);
+                        let d = Datagram::new(remote, *local_addr, tos, ttl, &buf[..sz]);
                         datagrams.push(d);
                     }
                 }
@@ -450,7 +447,7 @@ fn process_loop(
         'write: loop {
             match client.process_output(Instant::now()) {
                 Output::Datagram(dgram) => {
-                    if let Err(e) = emit_datagram(socket, dgram) {
+                    if let Err(e) = emit_datagram(socket, &dgram) {
                         eprintln!("UDP write error: {e}");
                         client.close(Instant::now(), 0, e.to_string());
                         exiting = true;
@@ -1046,12 +1043,12 @@ fn main() -> Res<()> {
             SocketAddr::V6(..) => SocketAddr::new(IpAddr::V6(Ipv6Addr::from([0; 16])), 0),
         };
 
-        let socket = match UdpSocket::bind(&local_addr) {
+        let socket = match bind(&local_addr) {
             Err(e) => {
                 eprintln!("Unable to bind UDP socket: {e}");
                 exit(1)
             }
-            Ok(s) => s,
+            Ok(s) => UdpSocket::from_socket(s)?,
         };
 
         let poll = Poll::new()?;
@@ -1140,7 +1137,7 @@ mod old {
 
     use super::{qlog_new, KeyUpdateState, Res};
     use mio::{Events, Poll};
-    use neqo_common::{event::Provider, Datagram};
+    use neqo_common::{event::Provider, recv_datagram, Datagram};
     use neqo_crypto::{AuthenticationStatus, ResumptionToken};
     use neqo_transport::{
         Connection, ConnectionEvent, EmptyConnectionIdGenerator, Error, Output, State, StreamId,
@@ -1348,7 +1345,9 @@ mod old {
             )?;
 
             'read: loop {
-                match socket.recv_from(&mut buf[..]) {
+                let mut tos = 0;
+                let mut ttl = 0;
+                match recv_datagram(socket, &mut buf[..], &mut tos, &mut ttl) {
                     Err(ref err)
                         if err.kind() == ErrorKind::WouldBlock
                             || err.kind() == ErrorKind::Interrupted =>
@@ -1365,7 +1364,7 @@ mod old {
                             break 'read;
                         }
                         if sz > 0 {
-                            let d = Datagram::new(remote, *local_addr, &buf[..sz]);
+                            let d = Datagram::new(remote, *local_addr, tos, ttl, &buf[..sz]);
                             client.process_input(&d, Instant::now());
                             handler.maybe_key_update(client)?;
                         }
@@ -1382,7 +1381,7 @@ mod old {
             'write: loop {
                 match client.process_output(Instant::now()) {
                     Output::Datagram(dgram) => {
-                        if let Err(e) = emit_datagram(socket, dgram) {
+                        if let Err(e) = emit_datagram(socket, &dgram) {
                             eprintln!("UDP write error: {e}");
                             client.close(Instant::now(), 0, e.to_string());
                             exiting = true;

neqo-common/Cargo.toml

@@ -7,16 +7,30 @@ rust-version = "1.70.0"
 license = "MIT OR Apache-2.0"
 build = "build.rs"
 
+[build-dependencies]
+cfg_aliases = "0.2.0"
+
 [dependencies]
-log = {version = "0.4.0", default-features = false}
-env_logger = {version = "0.10", default-features = false}
+log = { version = "0.4.0", default-features = false }
+enum-map = "2.7.3"
+env_logger = { version = "0.10", default-features = false }
 lazy_static = "1.3.0"
+mio = { version = "0.6.17", optional = true }
 qlog = "0.11.0"
-time = {version = "0.3", features = ["formatting"]}
+time = { version = "0.3", features = ["formatting"] }
+nix = { git = "https://github.com/larseggert/nix.git", branch = "feat-tos", features = [
+        "socket",
+        "net",
+        "uio",
+], optional = true }
+
+[dev-dependencies]
+test-fixture = { path = "../test-fixture" }
 
 [features]
 deny-warnings = []
 ci = []
+socket = ["dep:nix", "dep:mio"]
 
 [target."cfg(windows)".dependencies.winapi]
 version = "0.3"

neqo-common/build.rs

@@ -1,8 +1,13 @@
+use cfg_aliases::cfg_aliases;
 use std::env;
 
 fn main() {
     let target = env::var("TARGET").unwrap();
     if target.contains("windows") {
         println!("cargo:rustc-link-lib=winmm");
     }
+
+    cfg_aliases! {
+        posix_socket: { any(target_os = "macos", target_os = "linux", target_os = "android") },
+    }
 }

neqo-common/src/datagram.rs

@@ -7,20 +7,97 @@
 use std::net::SocketAddr;
 use std::ops::Deref;
 
+use enum_map::Enum;
+
 use crate::hex_with_len;
 
+// ECN (Explicit Congestion Notification) codepoints mapped to the
+// lower 2 bits of the TOS field.
+// https://www.iana.org/assignments/dscp-registry/dscp-registry.xhtml
+#[derive(Copy, Clone, PartialEq, Eq, Enum)]
+#[repr(u8)]
+pub enum IpTosEcn {
+    NotEct = 0b00, // Not-ECT (Not ECN-Capable Transport) [RFC3168]
+    Ect1 = 0b01,   // ECT(1) (ECN-Capable Transport(1))[1] [RFC8311][RFC Errata 5399][RFC9331]
+    Ect0 = 0b10,   // ECT(0) (ECN-Capable Transport(0)) [RFC3168]
+    Ce = 0b11,     // CE (Congestion Experienced) [RFC3168]
+}
+
+impl From<u8> for IpTosEcn {
+    fn from(v: u8) -> Self {
+        match v & 0b11 {
+            0b00 => IpTosEcn::NotEct,
+            0b01 => IpTosEcn::Ect1,
+            0b10 => IpTosEcn::Ect0,
+            0b11 => IpTosEcn::Ce,
+            _ => unreachable!(),
+        }
+    }
+}
+
+impl From<IpTosEcn> for u8 {
+    fn from(val: IpTosEcn) -> Self {
+        val as u8
+    }
+}
+
+impl std::fmt::Debug for IpTosEcn {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        match self {
+            IpTosEcn::NotEct => f.write_str("Not-ECT"),
+            IpTosEcn::Ect1 => f.write_str("ECT(1)"),
+            IpTosEcn::Ect0 => f.write_str("ECT(0)"),
+            IpTosEcn::Ce => f.write_str("CE"),
+        }
+    }
+}
+
+// DiffServ Codepoints, mapped to the upper six bits of the TOS field.
+// https://www.iana.org/assignments/dscp-registry/dscp-registry.xhtml
+#[derive(Copy, Clone, PartialEq, Eq)]
+#[repr(u8)]
+pub enum IpTosDscp {
+    Cs0 = 0b0000_0000,        // [RFC2474]
+    Cs1 = 0b0010_0000,        // [RFC2474]
+    Cs2 = 0b0100_0000,        // [RFC2474]
+    Cs3 = 0b0110_0000,        // [RFC2474]
+    Cs4 = 0b1000_0000,        // [RFC2474]
+    Cs5 = 0b1010_0000,        // [RFC2474]
+    Cs6 = 0b1100_0000,        // [RFC2474]
+    Cs7 = 0b1110_0000,        // [RFC2474]
+    Af11 = 0b0010_1000,       // [RFC2597]
+    Af12 = 0b0011_0000,       // [RFC2597]
+    Af13 = 0b0011_1000,       // [RFC2597]
+    Af21 = 0b0100_1000,       // [RFC2597]
+    Af22 = 0b0101_0000,       // [RFC2597]
+    Af23 = 0b0101_1000,       // [RFC2597]
+    Af31 = 0b0110_1000,       // [RFC2597]
+    Af32 = 0b0111_0000,       // [RFC2597]
+    Af33 = 0b0111_1000,       // [RFC2597]
+    Af41 = 0b1000_1000,       // [RFC2597]
+    Af42 = 0b1001_0000,       // [RFC2597]
+    Af43 = 0b1001_1000,       // [RFC2597]
+    Ef = 0b1011_1000,         // [RFC3246]
+    VoiceAdmit = 0b1011_0000, // [RFC5865]
+    Le = 0b0000_0100,         // [RFC8622]
+}
+
 #[derive(PartialEq, Eq, Clone)]
 pub struct Datagram {
     src: SocketAddr,
     dst: SocketAddr,
+    tos: u8,
+    ttl: u8,
     d: Vec<u8>,
 }
 
 impl Datagram {
-    pub fn new<V: Into<Vec<u8>>>(src: SocketAddr, dst: SocketAddr, d: V) -> Self {
+    pub fn new<V: Into<Vec<u8>>>(src: SocketAddr, dst: SocketAddr, tos: u8, ttl: u8, d: V) -> Self {
         Self {
             src,
             dst,
+            tos,
+            ttl,
             d: d.into(),
         }
     }
@@ -34,6 +111,16 @@ impl Datagram {
     pub fn destination(&self) -> SocketAddr {
         self.dst
     }
+
+    #[must_use]
+    pub fn tos(&self) -> u8 {
+        self.tos
+    }
+
+    #[must_use]
+    pub fn ttl(&self) -> u8 {
+        self.ttl
+    }
 }
 
 impl Deref for Datagram {

neqo-common/src/lib.rs

@@ -15,14 +15,18 @@ pub mod hrtime;
 mod incrdecoder;
 pub mod log;
 pub mod qlog;
+#[cfg(feature = "socket")]
+pub mod socket;
 pub mod timer;
 
 pub use self::codec::{Decoder, Encoder};
-pub use self::datagram::Datagram;
+pub use self::datagram::{Datagram, IpTosDscp, IpTosEcn};
 pub use self::header::Header;
 pub use self::incrdecoder::{
     IncrementalDecoderBuffer, IncrementalDecoderIgnore, IncrementalDecoderUint,
 };
+#[cfg(feature = "socket")]
+pub use self::socket::{bind, emit_datagram, recv_datagram};
 
 use std::fmt::Write;
 

neqo-common/src/qlog.rs

@@ -48,6 +48,11 @@ impl NeqoQlog {
         })
     }
 
+    #[must_use]
+    pub fn inner(&self) -> Rc<RefCell<Option<NeqoQlogShared>>> {
+        Rc::clone(&self.inner)
+    }
+
     /// Create a disabled `NeqoQlog` configuration.
     #[must_use]
     pub fn disabled() -> Self {
@@ -144,3 +149,39 @@ pub fn new_trace(role: Role) -> qlog::TraceSeq {
         }),
     }
 }
+
+#[cfg(test)]
+mod test {
+    use qlog::events::Event;
+    use test_fixture::EXPECTED_LOG_HEADER;
+
+    const EV_DATA: qlog::events::EventData =
+        qlog::events::EventData::SpinBitUpdated(qlog::events::connectivity::SpinBitUpdated {
+            state: true,
+        });
+
+    const EXPECTED_LOG_EVENT: &str = concat!(
+        "\u{1e}",
+        r#"{"time":0.0,"name":"connectivity:spin_bit_updated","data":{"state":true}}"#,
+        "\n"
+    );
+
+    #[test]
+    fn new_neqo_qlog() {
+        let (_log, contents) = test_fixture::new_neqo_qlog();
+        assert_eq!(contents.to_string(), EXPECTED_LOG_HEADER);
+    }
+
+    #[test]
+    fn add_event() {
+        let (mut log, contents) = test_fixture::new_neqo_qlog();
+        log.add_event(|| Some(Event::with_time(1.1, EV_DATA)));
+        assert_eq!(
+            contents.to_string(),
+            format!(
+                "{EXPECTED_LOG_HEADER}{e}",
+                e = EXPECTED_LOG_EVENT.replace("\"time\":0.0,", "\"time\":1.1,")
+            )
+        );
+    }
+}