aboutsummaryrefslogtreecommitdiffstats
path: root/src/cowboy_http2.erl
diff options
context:
space:
mode:
Diffstat (limited to 'src/cowboy_http2.erl')
-rw-r--r--src/cowboy_http2.erl215
1 files changed, 156 insertions, 59 deletions
diff --git a/src/cowboy_http2.erl b/src/cowboy_http2.erl
index e6e8a8b..85fdeef 100644
--- a/src/cowboy_http2.erl
+++ b/src/cowboy_http2.erl
@@ -36,11 +36,20 @@
-record(stream, {
id = undefined :: cowboy_stream:streamid(),
%% Stream handlers and their state.
- state = undefined :: {module(), any()},
+ state = undefined :: {module(), any()} | flush,
%% Whether we finished sending data.
- local = idle :: idle | upgrade | cowboy_stream:fin(),
+ local = idle :: idle | upgrade | cowboy_stream:fin() | flush,
+ %% Local flow control window (how much we can send).
+ local_window :: integer(),
+ %% Buffered data waiting for the flow control window to increase.
+ local_buffer = queue:new() :: queue:queue(
+ {cowboy_stream:fin(), non_neg_integer(), iolist()
+ | {sendfile, non_neg_integer(), pos_integer(), file:name_all()}}),
+ local_buffer_size = 0 :: non_neg_integer(),
%% Whether we finished receiving data.
remote = nofin :: cowboy_stream:fin(),
+ %% Remote flow control window (how much we accept to receive).
+ remote_window :: integer(),
%% Request body length.
body_length = 0 :: non_neg_integer()
}).
@@ -67,7 +76,7 @@
% header_table_size => 4096,
% enable_push => false, %% We are the server. Push is never enabled.
% max_concurrent_streams => infinity,
-% initial_window_size => 65535,
+ initial_window_size => 65535,
max_frame_size => 16384
% max_header_list_size => infinity
} :: map(),
@@ -75,7 +84,13 @@
%% We need to be careful there. It's well possible that we send
%% two SETTINGS frames before we receive a SETTINGS ack.
next_settings = #{} :: undefined | map(), %% @todo perhaps set to undefined by default
- remote_settings = #{} :: map(),
+ remote_settings = #{
+ initial_window_size => 65535
+ } :: map(),
+
+ %% Connection-wide flow control window.
+ local_window = 65535 :: integer(), %% How much we can send.
+ remote_window = 65535 :: integer(), %% How much we accept to receive.
%% Stream identifiers.
client_streamid = 0 :: non_neg_integer(),
@@ -269,17 +284,22 @@ parse_settings_preface(State, _, _, _) ->
%% and terminate the stream if this is the end of it.
%% DATA frame.
-frame(State=#state{streams=Streams}, {data, StreamID, IsFin0, Data}) ->
+frame(State=#state{remote_window=ConnWindow, streams=Streams},
+ {data, StreamID, IsFin0, Data}) ->
case lists:keyfind(StreamID, #stream.id, Streams) of
- Stream = #stream{state=StreamState0, remote=nofin, body_length=Len0} ->
- Len = Len0 + byte_size(Data),
+ Stream = #stream{state=StreamState0, remote=nofin,
+ remote_window=StreamWindow, body_length=Len0} ->
+ DataLen = byte_size(Data),
+ Len = Len0 + DataLen,
IsFin = case IsFin0 of
fin -> {fin, Len};
nofin -> nofin
end,
try cowboy_stream:data(StreamID, IsFin, Data, StreamState0) of
{Commands, StreamState} ->
- commands(State, Stream#stream{state=StreamState, body_length=Len}, Commands)
+ commands(State#state{remote_window=ConnWindow - DataLen},
+ Stream#stream{state=StreamState, remote_window=StreamWindow - DataLen,
+ body_length=Len}, Commands)
catch Class:Reason ->
error_logger:error_msg("Exception occurred in "
"cowboy_stream:data(~p, ~p, ~p, ~p) with reason ~p:~p.",
@@ -342,13 +362,22 @@ frame(State, {ping_ack, _Opaque}) ->
frame(State, Frame={goaway, _, _, _}) ->
terminate(State, {stop, Frame, 'Client is going away.'});
%% Connection-wide WINDOW_UPDATE frame.
-frame(State, {window_update, _Increment}) ->
- %% @todo control flow
- State;
+frame(State=#state{local_window=ConnWindow}, {window_update, Increment}) ->
+ send_data(State#state{local_window=ConnWindow + Increment});
%% Stream-specific WINDOW_UPDATE frame.
-frame(State, {window_update, _StreamID, _Increment}) ->
- %% @todo stream-specific control flow
- State;
+frame(State0=#state{streams=Streams0}, {window_update, StreamID, Increment}) ->
+ case lists:keyfind(StreamID, #stream.id, Streams0) of
+ Stream0 = #stream{local_window=StreamWindow} ->
+ {State, Stream} = send_data(State0,
+ Stream0#stream{local_window=StreamWindow + Increment}),
+ Streams = lists:keystore(StreamID, #stream.id, Streams0, Stream),
+ State#state{streams=Streams};
+ false ->
+ %% @todo Receiving this frame on a stream in the idle state is an error.
+ %% WINDOW_UPDATE frames may be received for a short period of time
+ %% after a stream is closed. They must be ignored.
+ State0
+ end;
%% Unexpected CONTINUATION frame.
frame(State, {continuation, _, _, _}) ->
terminate(State, {connection_error, protocol_error,
@@ -419,11 +448,8 @@ commands(State=#state{socket=Socket, transport=Transport, encode_state=EncodeSta
[{sendfile, fin, O, B, P}|Tail]);
_ ->
Transport:send(Socket, cow_http2:headers(StreamID, nofin, HeaderBlock)),
- %% @todo 16384 is the default SETTINGS_MAX_FRAME_SIZE.
- %% Use the length set by the server instead, if any.
- %% @todo Would be better if we didn't have to convert to binary.
- send_data(Socket, Transport, StreamID, fin, iolist_to_binary(Body), 16384),
- commands(State#state{encode_state=EncodeState}, Stream#stream{local=fin}, Tail)
+ {State1, Stream1} = send_data(State, Stream#stream{local=nofin}, fin, Body),
+ commands(State1#state{encode_state=EncodeState}, Stream1, Tail)
end;
%% @todo response when local!=idle
%% Send response headers and initiate chunked encoding.
@@ -445,10 +471,9 @@ commands(State=#state{socket=Socket, transport=Transport, encode_state=EncodeSta
%% split into multiple calls and flow control should be used to make
%% sure we only send as fast as the client can receive and don't block
%% anything.
-commands(State=#state{socket=Socket, transport=Transport}, Stream=#stream{id=StreamID, local=nofin},
- [{data, IsFin, Data}|Tail]) ->
- Transport:send(Socket, cow_http2:data(StreamID, IsFin, Data)),
- commands(State, Stream#stream{local=IsFin}, Tail);
+commands(State0, Stream0=#stream{local=nofin}, [{data, IsFin, Data}|Tail]) ->
+ {State, Stream} = send_data(State0, Stream0, IsFin, Data),
+ commands(State, Stream, Tail);
%% @todo data when local!=nofin
@@ -463,16 +488,10 @@ commands(State=#state{socket=Socket, transport=Transport}, Stream=#stream{id=Str
%% to ensure the file is sent in chunks (which would require a better
%% flow control at the stream handler level). One thing for sure, the
%% implementation necessarily varies between HTTP/1.1 and HTTP/2.
-commands(State=#state{socket=Socket, transport=Transport}, Stream=#stream{id=StreamID, local=nofin},
+commands(State0, Stream0=#stream{local=nofin},
[{sendfile, IsFin, Offset, Bytes, Path}|Tail]) ->
- %% @todo We currently have a naive implementation without a
- %% scheduler to prioritize frames that need to be sent.
- %% A future update will need to queue such data frames
- %% and only send them when there is nothing currently
- %% being sent. We would probably also benefit from doing
- %% asynchronous sends.
- sendfile(Socket, Transport, StreamID, IsFin, Offset, Bytes, Path, 16384),
- commands(State, Stream#stream{local=IsFin}, Tail);
+ {State, Stream} = send_data(State0, Stream0, IsFin, {sendfile, Offset, Bytes, Path}),
+ commands(State, Stream, Tail);
%% @todo sendfile when local!=nofin
%% Send a push promise.
%%
@@ -500,9 +519,15 @@ commands(State0=#state{socket=Socket, transport=Transport, server_streamid=Promi
State = stream_init(State0#state{server_streamid=PromisedStreamID + 2, encode_state=EncodeState},
PromisedStreamID, fin, iolist_to_binary(HeaderBlock)),
commands(State, Stream, Tail);
-%% @todo Update the flow control state.
-commands(State, Stream, [{flow, _Size}|Tail]) ->
- commands(State, Stream, Tail);
+commands(State=#state{socket=Socket, transport=Transport, remote_window=ConnWindow},
+ Stream=#stream{id=StreamID, remote_window=StreamWindow},
+ [{flow, Size}|Tail]) ->
+ Transport:send(Socket, [
+ cow_http2:window_update(Size),
+ cow_http2:window_update(StreamID, Size)
+ ]),
+ commands(State#state{remote_window=ConnWindow + Size},
+ Stream#stream{remote_window=StreamWindow + Size}, Tail);
%% Supervise a child process.
commands(State=#state{children=Children}, Stream=#stream{id=StreamID},
[{spawn, Pid, _Shutdown}|Tail]) -> %% @todo Shutdown
@@ -536,30 +561,89 @@ status(Status) when is_integer(Status) ->
status(<< H, T, U, _/bits >>) when H >= $1, H =< $9, T >= $0, T =< $9, U >= $0, U =< $9 ->
<< H, T, U >>.
-%% This same function is found in gun_http2.
-send_data(Socket, Transport, StreamID, IsFin, Data, Length) ->
- if
- Length < byte_size(Data) ->
- << Payload:Length/binary, Rest/bits >> = Data,
- Transport:send(Socket, cow_http2:data(StreamID, nofin, Payload)),
- send_data(Socket, Transport, StreamID, IsFin, Rest, Length);
- true ->
- Transport:send(Socket, cow_http2:data(StreamID, IsFin, Data))
- end.
-%% @todo This is currently awfully slow. But at least it's correct.
-sendfile(Socket, Transport, StreamID, IsFin, Offset, Bytes, Path, Length) ->
- if
- Length < Bytes ->
- Transport:send(Socket, cow_http2:data_header(StreamID, nofin, Length)),
- Transport:sendfile(Socket, Path, Offset, Length),
- sendfile(Socket, Transport, StreamID, IsFin,
- Offset + Length, Bytes - Length, Path, Length);
- true ->
+
+
+%% @todo Should we ever want to implement the PRIORITY mechanism,
+%% this would be the place to do it. Right now, we just go over
+%% all streams and send what we can until either everything is
+%% sent or we run out of space in the window.
+send_data(State=#state{streams=Streams}) ->
+ resume_streams(State, Streams, []).
+
+%% @todo When streams terminate we need to remove the stream.
+resume_streams(State, [], Acc) ->
+ State#state{streams=lists:reverse(Acc)};
+%% While technically we should never get < 0 here, let's be on the safe side.
+resume_streams(State=#state{local_window=ConnWindow}, Streams, Acc)
+ when ConnWindow =< 0 ->
+ State#state{streams=lists:reverse(Acc, Streams)};
+%% We rely on send_data/2 to do all the necessary checks about the stream.
+resume_streams(State0, [Stream0|Tail], Acc) ->
+ {State, Stream} = send_data(State0, Stream0),
+ resume_streams(State, Tail, [Stream|Acc]).
+
+%% @todo We might want to print an error if local=fin.
+%%
+%% @todo It's possible that the stream terminates. We must remove it.
+send_data(State=#state{local_window=ConnWindow},
+ Stream=#stream{local=IsFin, local_window=StreamWindow, local_buffer_size=BufferSize})
+ when ConnWindow =< 0; IsFin =:= fin; StreamWindow =< 0; BufferSize =:= 0 ->
+ {State, Stream};
+send_data(State0, Stream0=#stream{local_buffer=Q0, local_buffer_size=BufferSize}) ->
+ %% We know there is an item in the queue.
+ {{value, {IsFin, DataSize, Data}}, Q} = queue:out(Q0),
+ {State, Stream} = send_data(State0,
+ Stream0#stream{local_buffer=Q, local_buffer_size=BufferSize - DataSize},
+ IsFin, Data),
+ send_data(State, Stream).
+
+%% Send data immediately if we can, buffer otherwise.
+%% @todo We might want to print an error if local=fin.
+send_data(State=#state{local_window=ConnWindow},
+ Stream=#stream{local_window=StreamWindow}, IsFin, Data)
+ when ConnWindow =< 0; StreamWindow =< 0 ->
+ {State, queue_data(Stream, IsFin, Data)};
+send_data(State=#state{socket=Socket, transport=Transport, local_window=ConnWindow},
+ Stream=#stream{id=StreamID, local_window=StreamWindow}, IsFin, Data) ->
+ MaxFrameSize = 16384, %% @todo Use the real SETTINGS_MAX_FRAME_SIZE set by the client.
+ SendSize = min(min(ConnWindow, StreamWindow), MaxFrameSize),
+ case Data of
+ {sendfile, Offset, Bytes, Path} when Bytes =< SendSize ->
Transport:send(Socket, cow_http2:data_header(StreamID, IsFin, Bytes)),
- Transport:sendfile(Socket, Path, Offset, Bytes)
+ Transport:sendfile(Socket, Path, Offset, Bytes),
+ {State#state{local_window=ConnWindow - SendSize},
+ Stream#stream{local=IsFin, local_window=StreamWindow - SendSize}};
+ {sendfile, Offset, Bytes, Path} ->
+ Transport:send(Socket, cow_http2:data_header(StreamID, nofin, SendSize)),
+ Transport:sendfile(Socket, Path, Offset, SendSize),
+ send_data(State#state{local_window=ConnWindow - SendSize},
+ Stream#stream{local_window=StreamWindow - SendSize},
+ IsFin, {sendfile, Offset + SendSize, Bytes - SendSize, Path});
+ Iolist0 ->
+ IolistSize = iolist_size(Iolist0),
+ if
+ IolistSize =< SendSize ->
+ Transport:send(Socket, cow_http2:data(StreamID, IsFin, Iolist0)),
+ {State#state{local_window=ConnWindow - SendSize},
+ Stream#stream{local=IsFin, local_window=StreamWindow - SendSize}};
+ true ->
+ {Iolist, More} = cowboy_iolists:split(SendSize, Iolist0),
+ Transport:send(Socket, cow_http2:data(StreamID, nofin, Iolist)),
+ send_data(State#state{local_window=ConnWindow - SendSize},
+ Stream#stream{local_window=StreamWindow - SendSize},
+ IsFin, More)
+ end
end.
+queue_data(Stream=#stream{local_buffer=Q0, local_buffer_size=Size0}, IsFin, Data) ->
+ DataSize = case Data of
+ {sendfile, _, Bytes, _} -> Bytes;
+ Iolist -> iolist_size(Iolist)
+ end,
+ Q = queue:in({IsFin, DataSize, Data}, Q0),
+ Stream#stream{local_buffer=Q, local_buffer_size=Size0 + DataSize}.
+
-spec terminate(#state{}, _) -> no_return().
terminate(undefined, Reason) ->
exit({shutdown, Reason});
@@ -641,12 +725,17 @@ stream_init(State0=#state{ref=Ref, socket=Socket, transport=Transport, peer=Peer
'Error while trying to decode HPACK-encoded header block. (RFC7540 4.3)'})
end.
-stream_handler_init(State=#state{opts=Opts}, StreamID, RemoteIsFin, LocalIsFin, Req) ->
+stream_handler_init(State=#state{opts=Opts,
+ local_settings=#{initial_window_size := RemoteWindow},
+ remote_settings=#{initial_window_size := LocalWindow}},
+ StreamID, RemoteIsFin, LocalIsFin, Req) ->
try cowboy_stream:init(StreamID, Req, Opts) of
{Commands, StreamState} ->
commands(State#state{client_streamid=StreamID},
#stream{id=StreamID, state=StreamState,
- remote=RemoteIsFin, local=LocalIsFin}, Commands)
+ remote=RemoteIsFin, local=LocalIsFin,
+ local_window=LocalWindow, remote_window=RemoteWindow},
+ Commands)
catch Class:Reason ->
error_logger:error_msg("Exception occurred in "
"cowboy_stream:init(~p, ~p, ~p) with reason ~p:~p.",
@@ -676,12 +765,20 @@ stream_terminate(State=#state{socket=Socket, transport=Transport,
Children = stream_terminate_children(Children0, StreamID, []),
State1#state{streams=Streams, children=Children};
%% When a response was sent but not terminated, we need to close the stream.
- {value, #stream{state=StreamState, local=nofin}, Streams} when Reason =:= normal ->
+ {value, #stream{state=StreamState, local=nofin, local_buffer_size=0}, Streams}
+ when Reason =:= normal ->
Transport:send(Socket, cow_http2:data(StreamID, fin, <<>>)),
stream_call_terminate(StreamID, Reason, StreamState),
Children = stream_terminate_children(Children0, StreamID, []),
State#state{streams=Streams, children=Children};
- %% Otherwise we sent an RST_STREAM and the stream is already closed.
+ %% Unless there is still data in the buffer. We can however reset
+ %% a few fields and set a special local state to avoid confusion.
+ {value, Stream=#stream{state=StreamState, local=nofin}, Streams} ->
+ stream_call_terminate(StreamID, Reason, StreamState),
+ Children = stream_terminate_children(Children0, StreamID, []),
+ State#state{streams=[Stream#stream{state=flush, local=flush}|Streams],
+ children=Children};
+ %% Otherwise we sent an RST_STREAM and/or the stream is already closed.
{value, #stream{state=StreamState}, Streams} ->
stream_call_terminate(StreamID, Reason, StreamState),
Children = stream_terminate_children(Children0, StreamID, []),