Direct data delivery for HTTP/2 Websocket

The current implementation moves the Websocket data through
stream handlers before the Websocket process receives it and
parses it. The new mechanism relays the data directly as
an Erlang message without any intermediary. It also manages
HTTP/2 flow control in a simpler way (it maintains it at a
configured value).

A new switch_protocol option (coming from the options returned
from init/2) called `data_delivery` decides which mechanism
to use: `stream_handlers` is the old behavior; `relay` the new.

WIP: Need to test the relay mechanism in other test suites

WIP: the benchmark test suite is greatly modified because I
     wanted to make sure Gun's poor performance was not negatively
	 impacting results (this bit me earlier). Some tests use it,
	 some don't, we should keep both for good measure.

WIP: Other minor todos.

Author: essen

src/cowboy_http2.erl

@@ -82,8 +82,14 @@
 -export_type([opts/0]).
 
 -record(stream, {
-	%% Whether the stream is currently stopping.
-	status = running :: running | stopping,
+	%% Whether the stream is currently in a special state.
+	%%
+	%% - The running state is the normal state of a stream.
+	%% - The relaying state is used by extended CONNECT protocols to
+	%%   use a 'relay' data_delivery method.
+	%% - The stopping state indicates the stream used the 'stop' command.
+	%% @todo Do we need to buffer data before the response is sent? Probably.
+	status = running :: running | {relaying, pid()} | stopping,
 
 	%% Flow requested for this stream.
 	flow = 0 :: non_neg_integer(),
@@ -327,6 +333,8 @@ loop(State=#state{parent=Parent, socket=Socket, transport=Transport,
 		%% Messages pertaining to a stream.
 		{{Pid, StreamID}, Msg} when Pid =:= self() ->
 			before_loop(info(State, StreamID, Msg), Buffer);
+		{'$cowboy_stream_data', {Pid, StreamID}, IsFin, Data} when Pid =:= self() ->
+			before_loop(commands(State, StreamID, [{data, IsFin, Data}]), Buffer);
 		%% Exit signal from children.
 		Msg = {'EXIT', Pid, _} ->
 			before_loop(down(State, Pid, Msg), Buffer);
@@ -520,6 +528,14 @@ data_frame(State0=#state{opts=Opts, flow=Flow0, streams=Streams}, StreamID, IsFi
 				reset_stream(State0, StreamID, {internal_error, {Class, Exception},
 					'Unhandled exception in cowboy_stream:data/4.'})
 			end;
+		%% Stream handlers are not used for the data when relaying.
+		#{StreamID := #stream{status={relaying, RelayPid}}} ->
+			RelayPid ! {'$cowboy_stream_data', {self(), StreamID}, IsFin, Data},
+			%% We keep a steady flow using the configured flow value.
+			%% Because we do not change the 'flow' value the update_window/2
+			%% function will always maintain this value (of course with
+			%% thresholds applying).
+			update_window(State0, StreamID);
 		%% We ignore DATA frames for streams that are stopping.
 		#{} ->
 			State0
@@ -866,6 +882,19 @@ commands(State=#state{socket=Socket, transport=Transport, http2_status=upgrade},
 	commands(State, StreamID, Tail);
 %% Use a different protocol within the stream (CONNECT :protocol).
 %% @todo Make sure we error out when the feature is disabled.
+%% There are two data_delivery: stream_handlers and relay.
+%% The former just has the data go through stream handlers
+%% like normal requests. The latter relays data directly.
+commands(State0=#state{flow=Flow, streams=Streams}, StreamID,
+		[{switch_protocol, Headers, _Mod, ModState=#{data_delivery := relay}}|Tail]) ->
+	State1 = info(State0, StreamID, {headers, 200, Headers}),
+	#{StreamID := Stream} = Streams,
+	#{data_delivery_pid := RelayPid} = ModState,
+	RelayFlow = maps:get(data_delivery_flow, ModState, 131072),
+	State = State1#state{flow=Flow + RelayFlow, streams=Streams#{StreamID => Stream#stream{
+		status={relaying, RelayPid},
+		flow=RelayFlow}}},
+	commands(State, StreamID, Tail);
 commands(State0, StreamID, [{switch_protocol, Headers, _Mod, _ModState}|Tail]) ->
 	State = info(State0, StreamID, {headers, 200, Headers}),
 	commands(State, StreamID, Tail);

src/cowboy_websocket.erl

@@ -91,7 +91,7 @@
 	parent :: undefined | pid(),
 	ref :: ranch:ref(),
 	socket = undefined :: inet:socket() | {pid(), cowboy_stream:streamid()} | undefined,
-	transport = undefined :: module() | undefined,
+	transport :: module() | {data_delivery, stream_handlers | relay},
 	opts = #{} :: opts(),
 	active = true :: boolean(),
 	handler :: module(),
@@ -149,7 +149,7 @@ upgrade(Req, Env, Handler, HandlerState) ->
 %% @todo Immediately crash if a response has already been sent.
 upgrade(Req0=#{version := Version}, Env, Handler, HandlerState, Opts) ->
 	FilteredReq = case maps:get(req_filter, Opts, undefined) of
-		undefined -> maps:with([method, version, scheme, host, port, path, qs, peer], Req0);
+		undefined -> maps:with([method, version, scheme, host, port, path, qs, peer, streamid], Req0);
 		FilterFun -> FilterFun(Req0)
 	end,
 	Utf8State = case maps:get(validate_utf8, Opts, true) of
@@ -273,12 +273,28 @@ websocket_handshake(State=#state{key=Key},
 %% For HTTP/2 we do not let the process die, we instead keep it
 %% for the Websocket stream. This is because in HTTP/2 we only
 %% have a stream, it doesn't take over the whole connection.
-websocket_handshake(State, Req=#{ref := Ref, pid := Pid, streamid := StreamID},
+%%
+%% There are two methods of delivering data to the Websocket session:
+%% - 'stream_handlers' is the default and makes the data go
+%%   through stream handlers just like when reading a request body;
+%% - 'relay' is a new method where data is sent as a message as
+%%   soon as it is received from the socket in a DATA frame.
+websocket_handshake(State=#state{opts=Opts},
+		Req=#{ref := Ref, pid := Pid, streamid := StreamID},
 		HandlerState, _Env) ->
 	%% @todo We don't want date and server headers.
 	Headers = cowboy_req:response_headers(#{}, Req),
-	Pid ! {{Pid, StreamID}, {switch_protocol, Headers, ?MODULE, {State, HandlerState}}},
-	takeover(Pid, Ref, {Pid, StreamID}, undefined, #{}, <<>>,
+	DataDelivery = maps:get(data_delivery, Opts, stream_handlers),
+	ModState = #{
+		data_delivery => DataDelivery,
+		%% For relay data_delivery. The flow is a hint and may
+		%% not be used by the underlying protocol.
+		data_delivery_pid => self(),
+		data_delivery_flow => maps:get(data_delivery_flow, Opts, 131072)
+	},
+	Pid ! {{Pid, StreamID}, {switch_protocol, Headers, ?MODULE, ModState}},
+	%% @todo We can't call the normal takeover because it tries to parse.
+	takeover(Pid, Ref, {Pid, StreamID}, {data_delivery, DataDelivery}, #{}, <<>>,
 		{State, HandlerState}).
 
 %% Connection process.
@@ -301,7 +317,7 @@ websocket_handshake(State, Req=#{ref := Ref, pid := Pid, streamid := StreamID},
 -type parse_state() :: #ps_header{} | #ps_payload{}.
 
 -spec takeover(pid(), ranch:ref(), inet:socket() | {pid(), cowboy_stream:streamid()},
-	module() | undefined, any(), binary(),
+	module() | {data_delivery, stream_handlers | relay}, any(), binary(),
 	{#state{}, any()}) -> no_return().
 takeover(Parent, Ref, Socket, Transport, Opts, Buffer,
 		{State0=#state{opts=WsOpts, handler=Handler, req=Req}, HandlerState}) ->
@@ -311,7 +327,7 @@ takeover(Parent, Ref, Socket, Transport, Opts, Buffer,
 		_ -> ranch:remove_connection(Ref)
 	end,
 	Messages = case Transport of
-		undefined -> undefined;
+		{data_delivery, _} -> undefined;
 		_ -> Transport:messages()
 	end,
 	State = set_idle_timeout(State0#state{parent=Parent,
@@ -355,13 +371,14 @@ after_init(State, HandlerState, ParseState) ->
 %% immediately but there might still be data to be processed in
 %% the message queue.
 
-setopts_active(#state{transport=undefined}) ->
+setopts_active(#state{transport={data_delivery, _}}) ->
 	ok;
 setopts_active(#state{socket=Socket, transport=Transport, opts=Opts}) ->
 	N = maps:get(active_n, Opts, 1),
 	Transport:setopts(Socket, [{active, N}]).
 
-maybe_read_body(#state{socket=Stream={Pid, _}, transport=undefined, active=true}) ->
+maybe_read_body(#state{transport={data_delivery, stream_handlers},
+		socket=Stream={Pid, _}, active=true}) ->
 	%% @todo Keep Ref around.
 	ReadBodyRef = make_ref(),
 	Pid ! {Stream, {read_body, self(), ReadBodyRef, auto, infinity}},
@@ -374,10 +391,11 @@ active(State) ->
 	maybe_read_body(State),
 	State#state{active=true}.
 
-passive(State=#state{transport=undefined}) ->
+passive(State=#state{transport={data_delivery, _}}) ->
 	%% Unfortunately we cannot currently cancel read_body.
 	%% But that's OK, we will just stop reading the body
 	%% after the next message.
+	%% @todo We can't stop relay data_delivery currently.
 	State#state{active=false};
 passive(State=#state{socket=Socket, transport=Transport, messages=Messages}) ->
 	Transport:setopts(Socket, [{active, false}]),
@@ -454,6 +472,10 @@ loop(State=#state{parent=Parent, socket=Socket, messages=Messages,
 		{request_body, _Ref, fin, _, Data} ->
 			maybe_read_body(State),
 			parse(?reset_idle_timeout(State), HandlerState, ParseState, Data);
+		%% @todo It would be better to check StreamID.
+		%% @todo We must ensure that IsFin=fin is handled like a socket close?
+		{'$cowboy_stream_data', {Pid, _StreamID}, _IsFin, Data} when Pid =:= Parent ->
+			parse(?reset_idle_timeout(State), HandlerState, ParseState, Data);
 		%% Timeouts.
 		{timeout, TRef, ?MODULE} ->
 			tick_idle_timeout(State, HandlerState, ParseState);
@@ -662,9 +684,14 @@ commands([Frame|Tail], State, Data0) ->
 			commands(Tail, State, Data)
 	end.
 
-transport_send(#state{socket=Stream={Pid, _}, transport=undefined}, IsFin, Data) ->
+transport_send(#state{transport={data_delivery, stream_handlers},
+		socket=Stream={Pid, _}}, IsFin, Data) ->
 	Pid ! {Stream, {data, IsFin, Data}},
 	ok;
+transport_send(#state{transport={data_delivery, relay},
+		socket=Stream={Pid, _}}, IsFin, Data) ->
+	Pid ! {'$cowboy_stream_data', Stream, IsFin, Data},
+	ok;
 transport_send(#state{socket=Socket, transport=Transport}, _, Data) ->
 	Transport:send(Socket, Data).
 

src/cowboy_webtransport.erl

@@ -277,6 +277,7 @@ handler_terminate(#state{handler=Handler, req=Req}, HandlerState, Reason) ->
 %% callback is not implemented.
 %%
 %% @todo Better type than map() for the cowboy_stream state.
+%% @todo Is this really useful?
 
 -spec info(cowboy_stream:streamid(), any(), State)
 	-> {cowboy_stream:commands(), State} when State::map().

test/handlers/ws_ignore.erl

@@ -8,6 +8,7 @@
 
 init(Req, _) ->
 	{cowboy_websocket, Req, undefined, #{
+		data_delivery => relay,
 		compress => true
 	}}.
 

test/ws_SUITE_data/ws_echo.erl

@@ -8,6 +8,7 @@
 
 init(Req, _) ->
 	{cowboy_websocket, Req, undefined, #{
+		data_delivery => relay,
 		compress => true
 	}}.