New data delivery mechanism for HTTP/2+ Websocketdirect-data_delivery-for-h2-websocket

A new data_delivery mechanism called 'relay' has been added.
It bypasses stream handlers (and the buffering in cowboy_stream_h)
and sends the data directly to the process implementing
Websocket (and should work for other similar protocols
like HTTP/2 WebTransport).

Flow control in HTTP/2 is maintained in a simpler way,
via a configured flow value that is used to maintain
the window to a reasonable value when data is received.

The 'relay' data_delivery has been implemented for both
HTTP/2 and HTTP/3. It has not been implemented for HTTP/1.1
since switching protocol there overrides the connection process.

HTTP/2 Websocket is now better tested.

A bug was fixed with the 'stream_handlers' data_delivery
where active mode would not be reenabled if it was disabled
at some point.

The Websocket performance suite has been updated to
include tests that do not use Gun. Websocket modules
used by the performance suite use the 'relay' data_delivery
now. Performance is improved significantly with 'relay',
between 10% and 20% faster. HTTP/2 Websocket performance
is not on par with HTTP/1.1 still, but the remaining
difference is thought to be from the HTTP/2 overhead and
flow control.

9 files changed, 45 insertions, 24 deletions

diff --git a/test/handlers/ws_active_commands_h.erl b/test/handlers/ws_active_commands_h.erl
index 1c615e3..bbf8907 100644
--- a/test/handlers/ws_active_commands_h.erl
+++ b/test/handlers/ws_active_commands_h.erl
@@ -9,8 +9,8 @@
 -export([websocket_handle/2]).
 -export([websocket_info/2]).
 
-init(Req, RunOrHibernate) ->
-	{cowboy_websocket, Req, RunOrHibernate}.
+init(Req, Opts) ->
+	{cowboy_websocket, Req, maps:get(run_or_hibernate, Opts), Opts}.
 
 websocket_init(State=run) ->
 	erlang:send_after(1500, self(), active_true),
diff --git a/test/handlers/ws_deflate_commands_h.erl b/test/handlers/ws_deflate_commands_h.erl
index 14236bc..be3e16e 100644
--- a/test/handlers/ws_deflate_commands_h.erl
+++ b/test/handlers/ws_deflate_commands_h.erl
@@ -8,10 +8,11 @@
 -export([websocket_handle/2]).
 -export([websocket_info/2]).
 
-init(Req, RunOrHibernate) ->
+init(Req, Opts) ->
+	DataDelivery = maps:get(data_delivery, Opts, stream_handlers),
 	{cowboy_websocket, Req,
-		#{deflate => true, hibernate => RunOrHibernate},
-		#{compress => true}}.
+		#{deflate => true, hibernate => maps:get(run_or_hibernate, Opts)},
+		#{compress => true, data_delivery => DataDelivery}}.
 
 websocket_handle(Frame, State=#{deflate := Deflate0, hibernate := run}) ->
 	Deflate = not Deflate0,
diff --git a/test/handlers/ws_handle_commands_h.erl b/test/handlers/ws_handle_commands_h.erl
index da3ffad..0a8513b 100644
--- a/test/handlers/ws_handle_commands_h.erl
+++ b/test/handlers/ws_handle_commands_h.erl
@@ -9,14 +9,20 @@
 -export([websocket_handle/2]).
 -export([websocket_info/2]).
 
-init(Req=#{pid := Pid}, RunOrHibernate) ->
+init(Req, Opts) ->
 	Commands0 = cowboy_req:header(<<"x-commands">>, Req),
 	Commands = binary_to_term(base64:decode(Commands0)),
 	case Commands of
-		bad -> ct_helper_error_h:ignore(Pid, cowboy_websocket, handler_call, 6);
-		_ -> ok
+		bad ->
+			Pid = case Req of
+				#{version := 'HTTP/2'} -> self();
+				#{pid := Pid0} -> Pid0
+			end,
+			ct_helper_error_h:ignore(Pid, cowboy_websocket, handler_call, 6);
+		_ ->
+			ok
 	end,
-	{cowboy_websocket, Req, {Commands, RunOrHibernate}}.
+	{cowboy_websocket, Req, {Commands, maps:get(run_or_hibernate, Opts)}, Opts}.
 
 websocket_init(State) ->
 	{[], State}.
diff --git a/test/handlers/ws_ignore.erl b/test/handlers/ws_ignore.erl
index 9fe3322..37595aa 100644
--- a/test/handlers/ws_ignore.erl
+++ b/test/handlers/ws_ignore.erl
@@ -8,6 +8,7 @@
 
 init(Req, _) ->
 	{cowboy_websocket, Req, undefined, #{
+		data_delivery => relay,
 		compress => true
 	}}.
 
diff --git a/test/handlers/ws_info_commands_h.erl b/test/handlers/ws_info_commands_h.erl
index d596473..2115727 100644
--- a/test/handlers/ws_info_commands_h.erl
+++ b/test/handlers/ws_info_commands_h.erl
@@ -10,14 +10,20 @@
 -export([websocket_handle/2]).
 -export([websocket_info/2]).
 
-init(Req=#{pid := Pid}, RunOrHibernate) ->
+init(Req, Opts) ->
 	Commands0 = cowboy_req:header(<<"x-commands">>, Req),
 	Commands = binary_to_term(base64:decode(Commands0)),
 	case Commands of
-		bad -> ct_helper_error_h:ignore(Pid, cowboy_websocket, handler_call, 6);
-		_ -> ok
+		bad ->
+			Pid = case Req of
+				#{version := 'HTTP/2'} -> self();
+				#{pid := Pid0} -> Pid0
+			end,
+			ct_helper_error_h:ignore(Pid, cowboy_websocket, handler_call, 6);
+		_ ->
+			ok
 	end,
-	{cowboy_websocket, Req, {Commands, RunOrHibernate}}.
+	{cowboy_websocket, Req, {Commands, maps:get(run_or_hibernate, Opts)}, Opts}.
 
 websocket_init(State) ->
 	self() ! shoot,
diff --git a/test/handlers/ws_init_commands_h.erl b/test/handlers/ws_init_commands_h.erl
index 8bae352..d821b18 100644
--- a/test/handlers/ws_init_commands_h.erl
+++ b/test/handlers/ws_init_commands_h.erl
@@ -9,14 +9,20 @@
 -export([websocket_handle/2]).
 -export([websocket_info/2]).
 
-init(Req=#{pid := Pid}, RunOrHibernate) ->
+init(Req, Opts) ->
 	Commands0 = cowboy_req:header(<<"x-commands">>, Req),
 	Commands = binary_to_term(base64:decode(Commands0)),
 	case Commands of
-		bad -> ct_helper_error_h:ignore(Pid, cowboy_websocket, handler_call, 6);
-		_ -> ok
+		bad ->
+			Pid = case Req of
+				#{version := 'HTTP/2'} -> self();
+				#{pid := Pid0} -> Pid0
+			end,
+			ct_helper_error_h:ignore(Pid, cowboy_websocket, handler_call, 6);
+		_ ->
+			ok
 	end,
-	{cowboy_websocket, Req, {Commands, RunOrHibernate}}.
+	{cowboy_websocket, Req, {Commands, maps:get(run_or_hibernate, Opts)}, Opts}.
 
 websocket_init(State={Commands, run}) ->
 	{Commands, State};
diff --git a/test/handlers/ws_init_h.erl b/test/handlers/ws_init_h.erl
index bbe9ef9..c44986f 100644
--- a/test/handlers/ws_init_h.erl
+++ b/test/handlers/ws_init_h.erl
@@ -8,9 +8,9 @@
 -export([websocket_handle/2]).
 -export([websocket_info/2]).
 
-init(Req, _) ->
+init(Req, Opts) ->
 	State = binary_to_atom(cowboy_req:qs(Req), latin1),
-	{cowboy_websocket, Req, State}.
+	{cowboy_websocket, Req, State, Opts}.
 
 %% Sleep to make sure the HTTP response was sent.
 websocket_init(State) ->
diff --git a/test/handlers/ws_set_options_commands_h.erl b/test/handlers/ws_set_options_commands_h.erl
index 1ab0af4..af768d6 100644
--- a/test/handlers/ws_set_options_commands_h.erl
+++ b/test/handlers/ws_set_options_commands_h.erl
@@ -7,9 +7,10 @@
 -export([websocket_handle/2]).
 -export([websocket_info/2]).
 
-init(Req, RunOrHibernate) ->
-	{cowboy_websocket, Req, RunOrHibernate,
-		#{idle_timeout => infinity}}.
+init(Req, Opts) ->
+	DataDelivery = maps:get(data_delivery, Opts, stream_handlers),
+	{cowboy_websocket, Req, maps:get(run_or_hibernate, Opts),
+		#{idle_timeout => infinity, data_delivery => DataDelivery}}.
 
 %% Set the idle_timeout option dynamically.
 websocket_handle({text, <<"idle_timeout_short">>}, State=run) ->
diff --git a/test/handlers/ws_shutdown_reason_commands_h.erl b/test/handlers/ws_shutdown_reason_commands_h.erl
index 90b435c..878b70a 100644
--- a/test/handlers/ws_shutdown_reason_commands_h.erl
+++ b/test/handlers/ws_shutdown_reason_commands_h.erl
@@ -10,9 +10,9 @@
 -export([websocket_handle/2]).
 -export([websocket_info/2]).
 
-init(Req, RunOrHibernate) ->
+init(Req, Opts) ->
 	TestPid = list_to_pid(binary_to_list(cowboy_req:header(<<"x-test-pid">>, Req))),
-	{cowboy_websocket, Req, {TestPid, RunOrHibernate}}.
+	{cowboy_websocket, Req, {TestPid, maps:get(run_or_hibernate, Opts)}, Opts}.
 
 websocket_init(State={TestPid, RunOrHibernate}) ->
 	TestPid ! {ws_pid, self()},