Use cow_http2_machine:ensure_window

Gun was very inefficient at receiving HTTP/2 bodies. Switching
to ensure_window and increasing the default window sizes brings
the response body reading performance at least on par with the
one for HTTP/1.1.

This has a small negative impact on message flow control because
we stop updating the window later than we did before, increasing
the number of extra messages we may send. The exact amount depends
on configuration and the exact moment flow control kicks in.

3 files changed, 34 insertions, 9 deletions

diff --git a/test/flow_SUITE.erl b/test/flow_SUITE.erl
index 937af26..5076bbc 100644
--- a/test/flow_SUITE.erl
+++ b/test/flow_SUITE.erl
@@ -71,6 +71,8 @@ default_flow_http2(_) ->
 				flow => 1,
 				%% We set the max frame size to the same as the initial
 				%% window size in order to reduce the number of data messages.
+				initial_connection_window_size => 65535,
+				initial_stream_window_size => 65535,
 				max_frame_size_received => 65535
 			},
 			protocols => [http2]
@@ -86,8 +88,11 @@ default_flow_http2(_) ->
 		%% Then we confirm that we can override it per request.
 		StreamRef2 = gun:get(ConnPid, "/", [], #{flow => 2}),
 		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef2),
-		%% We set the flow to 2 therefore we will receive *3* data messages
-		%% and then nothing because two windows have been fully consumed.
+		%% We set the flow to 2 but due to the ensure_window algorithm
+		%% we end up receiving *5* data messages before flow control kicks in,
+		%% equivalent to 3 SSE events.
+		{data, nofin, _} = gun:await(ConnPid, StreamRef2),
+		{data, nofin, _} = gun:await(ConnPid, StreamRef2),
 		{data, nofin, _} = gun:await(ConnPid, StreamRef2),
 		{data, nofin, _} = gun:await(ConnPid, StreamRef2),
 		{data, nofin, _} = gun:await(ConnPid, StreamRef2),
@@ -132,7 +137,11 @@ flow_http2(_) ->
 		{ok, ConnPid} = gun:open("localhost", Port, #{
 			%% We set the max frame size to the same as the initial
 			%% window size in order to reduce the number of data messages.
-			http2_opts => #{max_frame_size_received => 65535},
+			http2_opts => #{
+				initial_connection_window_size => 65535,
+				initial_stream_window_size => 65535,
+				max_frame_size_received => 65535
+			},
 			protocols => [http2]
 		}),
 		{ok, http2} = gun:await_up(ConnPid),
@@ -145,14 +154,16 @@ flow_http2(_) ->
 		%% We consumed all the window available.
 		65535 = byte_size(D1) + byte_size(D2),
 		{error, timeout} = gun:await(ConnPid, StreamRef, 3000),
-		%% We then update the flow and get *3* more data messages but no more.
+		%% We then update the flow and get *5* more data messages but no more.
 		gun:update_flow(ConnPid, StreamRef, 2),
 		{data, nofin, D3} = gun:await(ConnPid, StreamRef),
 		{data, nofin, D4} = gun:await(ConnPid, StreamRef),
 		{data, nofin, D5} = gun:await(ConnPid, StreamRef),
+		{data, nofin, D6} = gun:await(ConnPid, StreamRef),
+		{data, nofin, D7} = gun:await(ConnPid, StreamRef),
 		%% We consumed all the window available again.
-		%% D3 is the end of the truncated D2, D4 is full and D5 truncated.
-		65535 = byte_size(D3) + byte_size(D4) + byte_size(D5),
+		%% D3 is the end of the truncated D2, D4, D5 and D6 are full and D7 truncated.
+		131070 = byte_size(D3) + byte_size(D4) + byte_size(D5) + byte_size(D6) + byte_size(D7),
 		{error, timeout} = gun:await(ConnPid, StreamRef, 1000),
 		gun:close(ConnPid)
 	after
@@ -302,6 +313,8 @@ sse_flow_http2(_) ->
 			%% window size in order to reduce the number of data messages.
 			http2_opts => #{
 				content_handlers => [gun_sse_h, gun_data_h],
+				initial_connection_window_size => 65535,
+				initial_stream_window_size => 65535,
 				max_frame_size_received => 65535
 			},
 			protocols => [http2]
@@ -314,10 +327,12 @@ sse_flow_http2(_) ->
 		%% the second event was fully received.
 		{sse, _} = gun:await(ConnPid, StreamRef),
 		{error, timeout} = gun:await(ConnPid, StreamRef, 3000),
-		%% We then update the flow and get 2 more event messages but no more.
+		%% We then update the flow and get 3 more event messages but no more.
+		%% We get an extra message because of the ensure_window algorithm.
 		gun:update_flow(ConnPid, StreamRef, 2),
 		{sse, _} = gun:await(ConnPid, StreamRef),
 		{sse, _} = gun:await(ConnPid, StreamRef),
+		{sse, _} = gun:await(ConnPid, StreamRef),
 		{error, timeout} = gun:await(ConnPid, StreamRef, 1000),
 		gun:close(ConnPid)
 	after
diff --git a/test/gun_test.erl b/test/gun_test.erl
index e74fcd0..a2cbf6d 100644
--- a/test/gun_test.erl
+++ b/test/gun_test.erl
@@ -79,6 +79,8 @@ http2_handshake(Socket, Transport) ->
 	%% Receive the SETTINGS from the preface.
 	{ok, <<Len:24>>} = Transport:recv(Socket, 3, 5000),
 	{ok, <<4:8, 0:40, _:Len/binary>>} = Transport:recv(Socket, 6 + Len, 5000),
+	%% Receive the WINDOW_UPDATE sent with the preface.
+	{ok, <<4:24, 8:8, 0:40, _:32>>} = Transport:recv(Socket, 13, 5000),
 	%% Send the SETTINGS ack.
 	ok = Transport:send(Socket, cow_http2:settings_ack()),
 	%% Receive the SETTINGS ack.
diff --git a/test/rfc7540_SUITE.erl b/test/rfc7540_SUITE.erl
index 0db9dd0..14e9c0c 100644
--- a/test/rfc7540_SUITE.erl
+++ b/test/rfc7540_SUITE.erl
@@ -98,9 +98,17 @@ lingering_data_counts_toward_connection_window(_) ->
 		%% Skip RST_STREAM.
 		{ok, << 4:24, 3:8, 1:40, _:32 >>} = gen_tcp:recv(Socket, 13, 1000),
 		%% Received a WINDOW_UPDATE frame after we got RST_STREAM.
-		{ok, << 4:24, 8:8, 0:40, 1000:32 >>} = gen_tcp:recv(Socket, 13, 1000)
+		{ok, << 4:24, 8:8, 0:40, Increment:32 >>} = gen_tcp:recv(Socket, 13, 1000),
+		true = Increment > 0
 	end),
-	{ok, ConnPid} = gun:open("localhost", Port, #{protocols => [http2]}),
+	{ok, ConnPid} = gun:open("localhost", Port, #{
+		protocols => [http2],
+		http2_opts => #{
+			%% We don't set 65535 because we still want to have an initial WINDOW_UPDATE.
+			initial_connection_window_size => 65536,
+			initial_stream_window_size => 65535
+		}
+	}),
 	{ok, http2} = gun:await_up(ConnPid),
 	timer:sleep(100), %% Give enough time for the handshake to fully complete.
 	%% Step 1.