Merge branch 'permessage-deflate' into feat/taosdata/permessage-deflate

* permessage-deflate: (78 commits)
  Fix clippy warnings and errors when disabling handshake feature
  Apply most suggestions by @erebe about permessage deflate
  Improve performance of processing continue frames
  add more details for utf8 errors for debugging
  Implement `From<Bytes>` for `Message`
  Don't copy data before decompression
  Apply suggestions from code review
  fix(Utf8Bytes): hash consistency for Borrow + Hash traits
  Bump version
  Implement `PartialOrd` and `Ord` for `Utf8Bytes`
  Derive `Hash` for `Utf8Bytes`
  Implement `Borrow<str>`, `AsRef<[u8]>`, `AsRef<str>` and `AsRef<Bytes>` for `Utf8Bytes
  Fix formatting of srv_accept_unmasked_frames example
  Decompress single-frame messages as well
  Add WebSocketConfig::read_buffer_size docs explaining performance/memory tradeoff (snapview#482)
  Update rand requirement from 0.8.0 to 0.9.0 (snapview#481)
  Remove byteorder, use bytes::Buf::get_uint instead (snapview#477)
  feat: add unsafe Utf8Bytes::from_bytes_unchecked (snapview#476)
  Keep the set-len refactor but use safe resize
  Revert "Revert "Use set-len when reading into buffer""
  ...

Author: zitsen

.github/workflows/ci.yml

@@ -7,12 +7,22 @@ jobs:
     name: Format
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@nightly
         with:
           components: rustfmt
       - run: cargo fmt --all --check
 
+  clippy:
+    name: Clippy
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          components: clippy
+      - run: cargo clippy --all --tests --all-features -- -D warnings
+
   test:
     name: Test
     runs-on: ubuntu-latest
@@ -24,7 +34,7 @@ jobs:
 
     steps:
       - name: Checkout sources
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Install toolchain
         uses: dtolnay/rust-toolchain@master
@@ -50,11 +60,11 @@ jobs:
     strategy:
       matrix:
         rust:
-          - 1.60.0
+          - 1.63.0
 
     steps:
       - name: Checkout sources
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Install toolchain
         uses: dtolnay/rust-toolchain@master
@@ -64,6 +74,9 @@ jobs:
       - name: Install dependencies
         run: sudo apt-get install libssl-dev
 
+      - name: Disable env_logger dev-dependency (cargo 1.63 workaround)
+        run: sed -i 's/env_logger/#env_logger/' Cargo.toml
+
       - name: Check
         run: cargo check
 
@@ -80,7 +93,7 @@ jobs:
 
     steps:
       - name: Checkout sources
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Install toolchain
         uses: dtolnay/rust-toolchain@master

CHANGELOG.md

@@ -1,3 +1,33 @@
+# 0.26.2
+- Add `WebSocketConfig::read_buffer_size` docs explaining performance/memory tradeoff.
+- Implement traits and add helper methods for the UTF8 payloads making them comparable and more ergonomic.
+
+# 0.26.1
+- Fix/revert unsoundness that could lead to UB with dodgy `Read` stream implementations.
+
+# 0.26.0
+- Simplify `Message` to use `Bytes` payload directly with simpler `Utf8Bytes` for text.
+- Change `CloseFrame` to use `Utf8Bytes` for `reason`.
+- Re-export `Bytes`.
+
+# 0.25.0
+
+- New `Payload` type for `Message` that allows sending messages with a payload that can be cheaply cloned (`Bytes`).
+  Long standing [issue](https://github.com/snapview/tungstenite-rs/issues/96) solved!
+- Add `WebSocketConfig::read_buffer_size` default 128 KiB. This improves high load read performance.
+  **Note: This default increases memory usage compared to previous versions particularly for users expecting a high number of connections. Configure 4-8 KiB to get a similar memory usage to 0.24**.
+- Make `WebSocketConfig` non-exhaustive & add builder style construction fns.
+- Remove deprecated `WebSocketConfig::max_send_queue`.
+- Trim spaces on `Sec-WebSocket-Protocol` header.
+- Eliminate data copies when reading complete messages & optimise read buffer. Improves performance.
+- Update `thiserror` to `2`.
+
+# 0.24.0
+
+- Raised MSRV to 1.63 to match `tokio-tungstenite`.
+- Connecting to WSS URL without TLS features specified results in a better error.
+- Handshake will now flush after completion to be safe (works better with buffered streams).
+
 # 0.23.0
 
 - Disable default features for `rustls` giving the user more flexibility.

Cargo.toml

@@ -7,11 +7,11 @@ authors = ["Alexey Galakhov", "Daniel Abramov"]
 license = "MIT OR Apache-2.0"
 readme = "README.md"
 homepage = "https://github.com/snapview/tungstenite-rs"
-documentation = "https://docs.rs/tungstenite/0.23.0"
+documentation = "https://docs.rs/tungstenite/0.26.2"
 repository = "https://github.com/snapview/tungstenite-rs"
-version = "0.23.0"
-edition = "2018"
-rust-version = "1.60"
+version = "0.26.2"
+edition = "2021"
+rust-version = "1.63"
 include = ["benches/**/*", "src/**/*", "examples/**/*", "LICENSE-*", "README.md", "CHANGELOG.md"]
 
 [package.metadata.docs.rs]
@@ -30,15 +30,14 @@ deflate = ["flate2"]
 
 [dependencies]
 data-encoding = { version = "2", optional = true }
-byteorder = "1.3.2"
-bytes = "1.0"
+bytes = "1.9.0"
 headers = { version = "0.4.0", optional = true }
 http = { version = "1.0", optional = true }
 httparse = { version = "1.3.4", optional = true }
 log = "0.4.8"
-rand = "0.8.0"
+rand = "0.9.0"
 sha1 = { version = "0.10", optional = true }
-thiserror = "1.0.23"
+thiserror = "2.0.7"
 url = { version = "2.1.0", optional = true }
 utf-8 = "0.7.5"
 
@@ -63,19 +62,22 @@ version = "1.0"
 
 [dependencies.rustls-native-certs]
 optional = true
-version = "0.7.0"
+version = "0.8.0"
 
 [dependencies.webpki-roots]
 optional = true
 version = "0.26"
 
 [dev-dependencies]
 criterion = "0.5.0"
-env_logger = "0.10.0"
+env_logger = "0.11"
 input_buffer = "0.5.0"
-rand = "0.8.4"
+rand = "0.9.0"
 socket2 = "0.5.5"
 
+[profile.bench]
+lto = "thin"
+
 [[bench]]
 name = "buffer"
 harness = false
@@ -84,6 +86,10 @@ harness = false
 name = "write"
 harness = false
 
+[[bench]]
+name = "read"
+harness = false
+
 [[example]]
 name = "client"
 required-features = ["handshake"]

benches/buffer.rs

@@ -77,7 +77,7 @@ impl<const CHUNK_SIZE: usize> Buf for StackReadBuffer<CHUNK_SIZE> {
     }
 
     fn advance(&mut self, cnt: usize) {
-        Buf::advance(self.as_cursor_mut(), cnt)
+        Buf::advance(self.as_cursor_mut(), cnt);
     }
 }
 
@@ -114,10 +114,10 @@ fn benchmark(c: &mut Criterion) {
     group.throughput(Throughput::Bytes(STREAM_SIZE as u64));
     group.bench_function("InputBuffer", |b| b.iter(|| input_buffer(black_box(stream.clone()))));
     group.bench_function("ReadBuffer (stack)", |b| {
-        b.iter(|| stack_read_buffer(black_box(stream.clone())))
+        b.iter(|| stack_read_buffer(black_box(stream.clone())));
     });
     group.bench_function("ReadBuffer (heap)", |b| {
-        b.iter(|| heap_read_buffer(black_box(stream.clone())))
+        b.iter(|| heap_read_buffer(black_box(stream.clone())));
     });
     group.finish();
 }

benches/read.rs

@@ -0,0 +1,92 @@
+//! Benchmarks for read performance.
+use criterion::{BatchSize, Criterion};
+use std::{
+    io::{self, Read, Write},
+    sync::{Arc, Mutex},
+};
+use tungstenite::{protocol::Role, Message, WebSocket};
+
+/// Mock stream with no artificial delays.
+#[derive(Default, Clone)]
+struct MockIo(Arc<Mutex<Vec<u8>>>);
+
+impl Read for MockIo {
+    fn read(&mut self, to: &mut [u8]) -> io::Result<usize> {
+        let mut data = self.0.lock().unwrap();
+        if data.is_empty() {
+            return Err(io::Error::new(io::ErrorKind::WouldBlock, "not ready"));
+        }
+        let len = data.len().min(to.len());
+        to[..len].copy_from_slice(data.drain(..len).as_slice());
+        Ok(len)
+    }
+}
+
+impl Write for MockIo {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.0.lock().unwrap().write(buf)
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+fn benchmark(c: &mut Criterion) {
+    /// Benchmark reading 100k mix of binary & text messages.
+    fn read_100k(role: Role, b: &mut criterion::Bencher<'_>) {
+        let io = MockIo::default();
+        let mut writer = WebSocket::from_raw_socket(
+            io.clone(),
+            match role {
+                Role::Client => Role::Server,
+                Role::Server => Role::Client,
+            },
+            None,
+        );
+        let mut ws = WebSocket::from_raw_socket(io, role, None);
+
+        b.iter_batched(
+            || {
+                let mut sum = 0;
+                for i in 0_u64..100_000 {
+                    writer
+                        .send(match i {
+                            _ if i % 3 == 0 => Message::binary(i.to_le_bytes().to_vec()),
+                            _ => Message::text(format!("{{\"id\":{i}}}")),
+                        })
+                        .unwrap();
+                    sum += i;
+                }
+                sum
+            },
+            |expected_sum| {
+                let mut sum = 0;
+                while sum != expected_sum {
+                    match ws.read().unwrap() {
+                        Message::Binary(v) => {
+                            let a: &[u8; 8] = v.as_ref().try_into().unwrap();
+                            sum += u64::from_le_bytes(*a);
+                        }
+                        Message::Text(msg) => {
+                            let i: u64 = msg.as_str()[6..msg.len() - 1].parse().unwrap();
+                            sum += i;
+                        }
+                        m => panic!("Unexpected {m}"),
+                    }
+                }
+            },
+            BatchSize::SmallInput,
+        );
+    }
+
+    c.bench_function("read+unmask 100k small messages (server)", |b| {
+        read_100k(Role::Server, b);
+    });
+
+    c.bench_function("read 100k small messages (client)", |b| {
+        read_100k(Role::Client, b);
+    });
+}
+
+criterion::criterion_group!(read_benches, benchmark);
+criterion::criterion_main!(read_benches);

benches/write.rs

@@ -1,11 +1,10 @@
 //! Benchmarks for write performance.
-use criterion::{BatchSize, Criterion};
+use criterion::Criterion;
 use std::{
-    hint,
-    io::{self, Read, Write},
+    hint, io,
     time::{Duration, Instant},
 };
-use tungstenite::{Message, WebSocket};
+use tungstenite::{protocol::Role, Message, WebSocket};
 
 const MOCK_WRITE_LEN: usize = 8 * 1024 * 1024;
 
@@ -16,12 +15,12 @@ const MOCK_WRITE_LEN: usize = 8 * 1024 * 1024;
 /// Each `flush` takes **~8µs** to simulate flush io.
 struct MockWrite(Vec<u8>);
 
-impl Read for MockWrite {
+impl io::Read for MockWrite {
     fn read(&mut self, _: &mut [u8]) -> io::Result<usize> {
         Err(io::Error::new(io::ErrorKind::WouldBlock, "reads not supported"))
     }
 }
-impl Write for MockWrite {
+impl io::Write for MockWrite {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         if self.0.len() + buf.len() > MOCK_WRITE_LEN {
             self.flush()?;
@@ -50,24 +49,28 @@ fn spin(duration: Duration) {
 }
 
 fn benchmark(c: &mut Criterion) {
-    // Writes 100k small json text messages then flushes
-    c.bench_function("write 100k small texts then flush", |b| {
-        let mut ws = WebSocket::from_raw_socket(
-            MockWrite(Vec::with_capacity(MOCK_WRITE_LEN)),
-            tungstenite::protocol::Role::Server,
-            None,
-        );
+    fn write_100k_then_flush(role: Role, b: &mut criterion::Bencher<'_>) {
+        let mut ws =
+            WebSocket::from_raw_socket(MockWrite(Vec::with_capacity(MOCK_WRITE_LEN)), role, None);
 
-        b.iter_batched(
-            || (0..100_000).map(|i| Message::Text(format!("{{\"id\":{i}}}"))),
-            |batch| {
-                for msg in batch {
-                    ws.write(msg).unwrap();
-                }
-                ws.flush().unwrap();
-            },
-            BatchSize::SmallInput,
-        )
+        b.iter(|| {
+            for i in 0_u64..100_000 {
+                let msg = match i {
+                    _ if i % 3 == 0 => Message::binary(i.to_le_bytes().to_vec()),
+                    _ => Message::text(format!("{{\"id\":{i}}}")),
+                };
+                ws.write(msg).unwrap();
+            }
+            ws.flush().unwrap();
+        });
+    }
+
+    c.bench_function("write 100k small messages then flush (server)", |b| {
+        write_100k_then_flush(Role::Server, b);
+    });
+
+    c.bench_function("write+mask 100k small messages then flush (client)", |b| {
+        write_100k_then_flush(Role::Client, b);
     });
 }
 

examples/autobahn-client.rs

@@ -11,18 +11,18 @@ fn get_case_count() -> Result<u32> {
     let (mut socket, _) = connect("ws://localhost:9001/getCaseCount")?;
     let msg = socket.read()?;
     socket.close(None)?;
-    Ok(msg.into_text()?.parse::<u32>().unwrap())
+    Ok(msg.into_text()?.as_str().parse::<u32>().unwrap())
 }
 
 fn update_reports() -> Result<()> {
-    let (mut socket, _) = connect(&format!("ws://localhost:9001/updateReports?agent={}", AGENT))?;
+    let (mut socket, _) = connect(format!("ws://localhost:9001/updateReports?agent={AGENT}"))?;
     socket.close(None)?;
     Ok(())
 }
 
 fn run_test(case: u32) -> Result<()> {
     info!("Running test case {}", case);
-    let case_url = &format!("ws://localhost:9001/runCase?case={}&agent={}", case, AGENT);
+    let case_url = format!("ws://localhost:9001/runCase?case={case}&agent={AGENT}");
 
     let mut config = WebSocketConfig::default();
     config.compression = Some(DeflateConfig::default());
@@ -46,7 +46,7 @@ fn main() {
     for case in 1..=total {
         if let Err(e) = run_test(case) {
             match e {
-                Error::ConnectionClosed | Error::Protocol(_) | Error::Utf8 => (),
+                Error::ConnectionClosed | Error::Protocol(_) | Error::Utf8(_) => (),
                 err => error!("test: {}", err),
             }
         }