Skip to content

Commit

Permalink
test: requests over max stream starving capacity
Browse files Browse the repository at this point in the history
  • Loading branch information
dswij committed Jan 6, 2024
1 parent f1f99e0 commit abff268
Showing 1 changed file with 106 additions and 2 deletions.
108 changes: 106 additions & 2 deletions tests/h2-tests/tests/prioritization.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
use futures::future::join;
use futures::{FutureExt, StreamExt};
use futures::future::{join, select};
use futures::{pin_mut, FutureExt, StreamExt};

use h2_support::prelude::*;
use h2_support::DEFAULT_WINDOW_SIZE;
use std::task::Context;
Expand Down Expand Up @@ -408,3 +409,106 @@ async fn send_data_receive_window_update() {

join(mock, h2).await;
}

#[tokio::test]
async fn stream_count_over_max_stream_limit_does_not_starve_capacity() {
use tokio::sync::oneshot;

h2_support::trace_init!();

let (io, mut srv) = mock::new();

let (tx, rx) = oneshot::channel();

let srv = async move {
let _ = srv
.assert_client_handshake_with_settings(
frames::settings()
// super tiny server
.max_concurrent_streams(1),
)
.await;
srv.recv_frame(frames::headers(1).request("POST", "http://example.com/"))
.await;

srv.recv_frame(frames::data(1, vec![0; 16384])).await;
srv.recv_frame(frames::data(1, vec![0; 16384])).await;
srv.recv_frame(frames::data(1, vec![0; 16384])).await;
srv.recv_frame(frames::data(1, vec![0; 16383]).eos()).await;
srv.send_frame(frames::headers(1).response(200).eos()).await;

// All of these connection capacities should be assigned to stream 3
srv.send_frame(frames::window_update(0, 16384)).await;
srv.send_frame(frames::window_update(0, 16384)).await;
srv.send_frame(frames::window_update(0, 16384)).await;
srv.send_frame(frames::window_update(0, 16383)).await;

// StreamId(3) should be able to send all of its request with the conn capacity
srv.recv_frame(frames::headers(3).request("POST", "http://example.com/"))
.await;
srv.recv_frame(frames::data(3, vec![0; 16384])).await;
srv.send_frame(frames::window_update(0, 16384)).await;
srv.recv_frame(frames::data(3, vec![0; 16384])).await;
srv.send_frame(frames::window_update(0, 16384)).await;
srv.recv_frame(frames::data(3, vec![0; 16384])).await;
srv.send_frame(frames::window_update(0, 16384)).await;
srv.recv_frame(frames::data(3, vec![0; 16383]).eos()).await;
srv.send_frame(frames::window_update(0, 16384)).await;
srv.send_frame(frames::headers(3).response(200).eos()).await;

// Then all the future stream is guaranteed to be send-able by induction
tx.send(()).unwrap();
};

fn request() -> Request<()> {
Request::builder()
.method(Method::POST)
.uri("http://example.com/")
.body(())
.unwrap()
}

let client = async move {
let (mut client, mut conn) = client::Builder::new()
.handshake::<_, Bytes>(io)
.await
.expect("handshake");

let (req1, mut send1) = client.send_request(request(), false).unwrap();
let (req2, mut send2) = client.send_request(request(), false).unwrap();

// Use up the connection window.
send1.send_data(vec![0; 65535].into(), true).unwrap();
// Queue up for more connection window.
send2.send_data(vec![0; 65535].into(), true).unwrap();

// Queue up more pending open streams
let (_req3, mut send3) = client.send_request(request(), false).unwrap();
let (_req4, mut send4) = client.send_request(request(), false).unwrap();
let (_req5, mut send5) = client.send_request(request(), false).unwrap();
let (_req6, mut send6) = client.send_request(request(), false).unwrap();
let (_req7, mut send7) = client.send_request(request(), false).unwrap();

send3.send_data(vec![0; 65535].into(), true).unwrap();
send4.send_data(vec![0; 65535].into(), true).unwrap();
send5.send_data(vec![0; 65535].into(), true).unwrap();
send6.send_data(vec![0; 65535].into(), true).unwrap();
send7.send_data(vec![0; 65535].into(), true).unwrap();

let response = conn.drive(req1).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);

let response = conn.drive(req2).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);

let _ = rx.await;
};

let task = join(srv, client);
pin_mut!(task);

let t = tokio::time::sleep(Duration::from_secs(5)).map(|_| panic!("time out"));
pin_mut!(t);

select(task, t).await;
}

0 comments on commit abff268

Please sign in to comment.