Unbounded stream receive window

Depend on connection window only.
libp2p · Nov 23, 2023 · ca14cec · ca14cec
1 parent e6200e7
commit ca14cec
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 10 deletions.
diff --git a/test-harness/src/lib.rs b/test-harness/src/lib.rs
@@ -454,7 +454,11 @@ impl Arbitrary for TestConfig {
             WindowUpdateMode::OnReceive
         });
         c.set_read_after_close(Arbitrary::arbitrary(g));
-        c.set_receive_window(256 * 1024 + usize::arbitrary(g) % (768 * 1024));
+        if bool::arbitrary(g) {
+            c.set_receive_window(Some(256 * 1024 + usize::arbitrary(g) % (768 * 1024)));
+        } else {
+            c.set_receive_window(None);
+        }
         TestConfig(c)
     }
 }
diff --git a/yamux/src/connection/stream.rs b/yamux/src/connection/stream.rs
@@ -25,7 +25,6 @@ use futures::{
     ready, SinkExt,
 };
 use parking_lot::{Mutex, MutexGuard};
-use std::convert::TryInto;
 use std::time::Instant;
 use std::{
     fmt, io,
@@ -548,8 +547,13 @@ impl Shared {
         match self.config.window_update_mode {
             WindowUpdateMode::OnReceive => {
                 // TODO: the whole mode should be removed.
-                debug_assert!(self.config.receive_window >= self.window);
-                return Some(self.config.receive_window.saturating_sub(self.window));
+                debug_assert!(self.config.receive_window.unwrap_or(usize::MAX) >= self.window);
+                return Some(
+                    self.config
+                        .receive_window
+                        .unwrap_or(u32::MAX as usize)
+                        .saturating_sub(self.window),
+                );
             }
             WindowUpdateMode::OnRead => {
                 debug_assert!(
@@ -584,7 +588,7 @@ impl Shared {
                         self.window_max = std::cmp::min(
                             std::cmp::min(
                                 self.window_max.saturating_mul(2),
-                                self.config.receive_window,
+                                self.config.receive_window.unwrap_or(usize::MAX),
                             ),
                             self.window_max
                                 + ((self.config.connection_window

diff --git a/yamux/src/lib.rs b/yamux/src/lib.rs
@@ -103,7 +103,7 @@ pub enum WindowUpdateMode {
 #[derive(Debug, Clone)]
 pub struct Config {
     // TODO: Rename to max_stream_receive_window
-    receive_window: usize,
+    receive_window: Option<usize>,
     // TODO: Rename to max_connection_receive_window
     connection_window: usize,
     max_buffer_size: usize,
@@ -116,7 +116,8 @@ pub struct Config {
 impl Default for Config {
     fn default() -> Self {
         Config {
-            receive_window: 16 * 1024 * 1024,
+            // TODO: Add rational: given that we have a connection window, ...
+            receive_window: None,
             // TODO: reevaluate default.
             // TODO: Add setter.
             connection_window: 1 * 1024 * 1024 * 1024,
@@ -136,12 +137,12 @@ impl Config {
     /// # Panics
     ///
     /// If the given receive window is < 256 KiB.
-    pub fn set_receive_window(&mut self, n: usize) -> &mut Self {
+    pub fn set_receive_window(&mut self, n: Option<usize>) -> &mut Self {
         self.receive_window = n;
         self.check();
         self
     }
-    
+
     pub fn set_connection_window(&mut self, n: usize) -> &mut Self {
         self.connection_window = n;
         self.check();
@@ -185,8 +186,9 @@ impl Config {
         self
     }
 
+    // TODO: Consider doing the check on creation, not on each builder method call.
     fn check(&self) {
-        assert!(self.receive_window >= DEFAULT_CREDIT);
+        assert!(self.receive_window.unwrap_or(usize::MAX) >= DEFAULT_CREDIT);
         assert!(self.connection_window >= self.max_num_streams * DEFAULT_CREDIT);
     }
 }