Skip to content

webmachine_util: monotonic-time-aware now_diff helpers #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
samwar merged 2 commits into from
Jul 20, 2026
Merged

webmachine_util: monotonic-time-aware now_diff helpers #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
309abdf
Flush the request body guard cross-process-safe
samwar Jul 14, 2026
5bda3e2
webmachine_util: monotonic-time-aware now_diff helpers
samwar Jul 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 93 additions & 42 deletions src/webmachine_request.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,12 @@

-define(IDLE_TIMEOUT, infinity).

%% Finite timeout used when we're only draining leftover request bytes
%% (see maybe_flush_req_body/1). Kept intentionally short so that a
%% misbehaving or already-drained peer can never wedge the request
%% pipeline. Legitimate streaming reads still use ?IDLE_TIMEOUT.
-define(FLUSH_TIMEOUT, 5000).

-type t() :: #wm_reqstate{}.
-export_type([t/0]).

Expand Down Expand Up @@ -588,7 +594,10 @@ expects_continue(PassedState) ->
sent_continue() ->
get(webmachine_sent_continue) =:= true.

recv_stream_body(PassedState=#wm_reqstate{reqdata=RD}, MaxHunkSize) ->
recv_stream_body(PassedState, MaxHunkSize) ->
recv_stream_body(PassedState, MaxHunkSize, ?IDLE_TIMEOUT).

recv_stream_body(PassedState=#wm_reqstate{reqdata=RD}, MaxHunkSize, Timeout) ->
case expects_continue(PassedState) and (not sent_continue()) of
true ->
put(webmachine_sent_continue, true),
Expand All @@ -608,28 +617,29 @@ recv_stream_body(PassedState=#wm_reqstate{reqdata=RD}, MaxHunkSize) ->
{<<>>, done};
chunked ->
start_recv_chunked_body(PassedState#wm_reqstate.socket,
MaxHunkSize);
MaxHunkSize, Timeout);
Length ->
recv_unchunked_body(PassedState#wm_reqstate.socket,
MaxHunkSize, Length)
MaxHunkSize, Length, Timeout)
end.

recv_unchunked_body(Socket, MaxHunk, DataLeft) ->
recv_unchunked_body(Socket, MaxHunk, DataLeft, Timeout) ->
case MaxHunk >= DataLeft of
true ->
case mochiweb_socket:recv(Socket,DataLeft,?IDLE_TIMEOUT) of
{ok,Data1} ->
case mochiweb_socket:recv(Socket, DataLeft, Timeout) of
{ok, Data1} ->
record_stream_progress(done),
{Data1, done};
{error, Error} ->
throw({webmachine_recv_error, Error})
end;
false ->
case mochiweb_socket:recv(Socket,MaxHunk,?IDLE_TIMEOUT) of
{ok,Data2} ->
case mochiweb_socket:recv(Socket, MaxHunk, Timeout) of
{ok, Data2} ->
Next = fun() ->
recv_unchunked_body(Socket, MaxHunk,
DataLeft-MaxHunk)
DataLeft - MaxHunk,
Timeout)
end,
record_stream_progress(Next),
{Data2, Next};
Expand All @@ -638,33 +648,35 @@ recv_unchunked_body(Socket, MaxHunk, DataLeft) ->
end
end.

start_recv_chunked_body(Socket, MaxHunk) ->
case read_chunk_length(Socket, false) of
start_recv_chunked_body(Socket, MaxHunk, Timeout) ->
case read_chunk_length(Socket, false, Timeout) of
0 ->
record_stream_progress(done),
{<<>>, done};
ChunkLength ->
recv_chunked_body(Socket,MaxHunk, ChunkLength)
recv_chunked_body(Socket, MaxHunk, ChunkLength, Timeout)
end.
recv_chunked_body(Socket, MaxHunk, LeftInChunk) ->
recv_chunked_body(Socket, MaxHunk, LeftInChunk, Timeout) ->
case MaxHunk >= LeftInChunk of
true ->
case mochiweb_socket:recv(Socket,LeftInChunk,?IDLE_TIMEOUT) of
{ok,Data1} ->
case mochiweb_socket:recv(Socket, LeftInChunk, Timeout) of
{ok, Data1} ->
Next = fun() ->
start_recv_chunked_body(Socket, MaxHunk)
start_recv_chunked_body(Socket, MaxHunk,
Timeout)
end,
record_stream_progress(Next),
{Data1, Next};
{error, Error} ->
throw({webmachine_recv_error, Error})
end;
false ->
case mochiweb_socket:recv(Socket,MaxHunk,?IDLE_TIMEOUT) of
{ok,Data2} ->
case mochiweb_socket:recv(Socket, MaxHunk, Timeout) of
{ok, Data2} ->
Next = fun() ->
recv_chunked_body(Socket, MaxHunk,
LeftInChunk-MaxHunk)
LeftInChunk - MaxHunk,
Timeout)
end,
record_stream_progress(Next),
{Data2, Next};
Expand All @@ -673,9 +685,9 @@ recv_chunked_body(Socket, MaxHunk, LeftInChunk) ->
end
end.

read_chunk_length(Socket, MaybeLastChunk) ->
read_chunk_length(Socket, MaybeLastChunk, Timeout) ->
mochiweb_socket:setopts(Socket, [{packet, line}]),
case mochiweb_socket:recv(Socket, 0, ?IDLE_TIMEOUT) of
case mochiweb_socket:recv(Socket, 0, Timeout) of
{ok, Header} ->
mochiweb_socket:setopts(Socket, [{packet, raw}]),
Splitter = fun (C) ->
Expand All @@ -689,7 +701,7 @@ read_chunk_length(Socket, MaybeLastChunk) ->
%% allow [badly formed] last chunk header to be
%% empty instead of '0' explicitly
if MaybeLastChunk -> 0;
true -> read_chunk_length(Socket, true)
true -> read_chunk_length(Socket, true, Timeout)
end;
_ ->
erlang:list_to_integer(Hex, 16)
Expand All @@ -713,28 +725,34 @@ maybe_flush_req_body(Req) ->
put(mochiweb_request_recv, true),
true;
undefined ->
case expects_continue(Req) and (not sent_continue()) of
case get(mochiweb_request_recv) of
true ->
%% If the request expected continue, but we didn't
%% send it, then tell mochiweb to ignore what the
%% content length or transfer encoding says about
%% a body.
put(mochiweb_request_recv, true),
%% The request body was already consumed via
%% wrq:req_body/1 (eager read). mochiweb's
%% request_recv flag is set but our own
%% webmachine_stream_progress was never touched
%% because the eager path bypasses
%% recv_stream_body/2. There is nothing left on
%% the socket to drain; attempting to recv here
%% would block on an already-empty socket.
true;
false ->
MaxFlush = max_flush_bytes(),
case MaxFlush of
0 ->
%% this server has been configured to
%% close connections on clients who send
%% requests whose bodies are ignored
false;
_ ->
%% There might be a body sitting out there we
%% haven't read - give it a try.
ReadSize = erlang:min(65535, MaxFlush),
flush_req_body(
catch recv_stream_body(Req, ReadSize), MaxFlush)
_ ->
%% Cross-process-safe fallback: some
%% applications dispatch body reads to a
%% separate worker process, so the
%% mochiweb_request_recv pdict flag never lands
%% on the mochiweb loop process even though the
%% body was consumed. The bodyfetch field is
%% set to `standard` or `stream` inside
%% call({req_body, MaxRecvBody}, ReqState) when
%% the body is read, and travels with the
%% ReqState by value across process boundaries
%% (unlike the pdict). Consult it before
%% falling into the drain path.
case Req#wm_reqstate.bodyfetch of
standard -> true;
stream -> true;
_ -> maybe_flush_undefined_drain(Req)
end
end;
Next ->
Expand All @@ -749,6 +767,39 @@ maybe_flush_req_body(Req) ->
end
end.

%% Handles the "we have no idea whether a body is on the wire" branch
%% of maybe_flush_req_body/1. Only reached when
%% get(mochiweb_request_recv) is not `true` (i.e. no eager read has
%% claimed the body), so an actual attempt to recv is warranted. The
%% recv is bounded by ?FLUSH_TIMEOUT rather than ?IDLE_TIMEOUT so that
%% a bogus content-length or a peer that closed without shutting down
%% write cleanly cannot wedge the request pipeline.
maybe_flush_undefined_drain(Req) ->
case expects_continue(Req) and (not sent_continue()) of
true ->
%% If the request expected continue, but we didn't send
%% it, then tell mochiweb to ignore what the content
%% length or transfer encoding says about a body.
put(mochiweb_request_recv, true),
true;
false ->
MaxFlush = max_flush_bytes(),
case MaxFlush of
0 ->
%% this server has been configured to close
%% connections on clients who send requests whose
%% bodies are ignored
false;
_ ->
%% There might be a body sitting out there we
%% haven't read - give it a bounded try.
ReadSize = erlang:min(65535, MaxFlush),
flush_req_body(
catch recv_stream_body(Req, ReadSize, ?FLUSH_TIMEOUT),
MaxFlush)
end
end.

max_flush_bytes() ->
application:get_env(webmachine, max_flush_bytes, 67108864).

Expand Down
104 changes: 82 additions & 22 deletions src/webmachine_util.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
-export([choose_media_type/2, format_content_type/1]).
-export([choose_charset/2]).
-export([choose_encoding/2]).
-export([now_diff_milliseconds/2]).
-export([now_diff_milliseconds/2, now_diff_microseconds/2]).
-export([media_type_to_detail/1,
quoted_string/1,
split_quoted_strings/1]).
Expand All @@ -36,6 +36,10 @@
-include_lib("eqc/include/eqc.hrl").
-endif.
-include_lib("eunit/include/eunit.hrl").
%% Test-only include for the record-shape contract test — pulls the
%% wm_log_data record definition into test scope so we can exercise
%% the record + helper contract directly.
-include("webmachine_logger.hrl").
-export([accept_header_to_media_types/1]).
-endif.

Expand Down Expand Up @@ -369,22 +373,27 @@ unescape_quoted_string([Char | Rest], Acc) ->
unescape_quoted_string(Rest, [Char | Acc]).


%% @type now() = {MegaSecs, Secs, MicroSecs}

%% This is faster than timer:now_diff() because it does not use bignums.
%% But it returns *milliseconds* (timer:now_diff returns microseconds.)
%% From http://www.erlang.org/ml-archive/erlang-questions/200205/msg00027.html

%% @doc Compute the difference between two now() tuples, in milliseconds.
%% @spec now_diff_milliseconds(now(), now()) -> integer()
now_diff_milliseconds(undefined, undefined) ->
0;
now_diff_milliseconds(undefined, T2) ->
now_diff_milliseconds(os:timestamp(), T2);
now_diff_milliseconds({M,S,U}, {M,S1,U1}) ->
((S-S1) * 1000) + ((U-U1) div 1000);
now_diff_milliseconds({M,S,U}, {M1,S1,U1}) ->
((M-M1)*1000000+(S-S1))*1000 + ((U-U1) div 1000).
%% @doc Compute the difference between two erlang:monotonic_time/0
%% values (native units, per the wm_log_data record) in milliseconds.
%% If the "later" time is undefined, uses erlang:monotonic_time/0 as
%% a stand-in — this matches the semantic that end_time on the record
%% may not have been set yet when the diff is requested. If both are
%% undefined the diff is zero.
-spec now_diff_milliseconds(undefined | integer(), integer()) -> integer().
now_diff_milliseconds(T2, T1) ->
erlang:convert_time_unit(diff_native(T2, T1), native, millisecond).

%% @doc As now_diff_milliseconds/2 but returns microseconds. Provided
%% so downstream access-log handlers that emit response times at
%% microsecond granularity don't need to duplicate the conversion.
-spec now_diff_microseconds(undefined | integer(), integer()) -> integer().
now_diff_microseconds(T2, T1) ->
erlang:convert_time_unit(diff_native(T2, T1), native, microsecond).

-spec diff_native(undefined | integer(), undefined | integer()) -> integer().
diff_native(undefined, undefined) -> 0;
diff_native(undefined, T1) -> diff_native(erlang:monotonic_time(), T1);
diff_native(T2, T1) -> T2 - T1.

-spec parse_range(RawRange::string(), ResourceLength::non_neg_integer()) ->
[{Start::non_neg_integer(), End::non_neg_integer()}].
Expand Down Expand Up @@ -523,11 +532,62 @@ guess_mime_test() ->


now_diff_milliseconds_test() ->
Late = {10, 10, 10},
Early1 = {10, 9, 9},
Early2 = {9, 9, 9},
?assertEqual(1000, now_diff_milliseconds(Late, Early1)),
?assertEqual(1000001000, now_diff_milliseconds(Late, Early2)).
NativeMs = erlang:convert_time_unit(1, millisecond, native),
T1 = 1000000000,
T2 = T1 + 1000 * NativeMs,
?assertEqual(1000, now_diff_milliseconds(T2, T1)).

now_diff_milliseconds_undefined_test() ->
?assertEqual(0, now_diff_milliseconds(undefined, undefined)),
Now = erlang:monotonic_time(),
Diff = now_diff_milliseconds(undefined, Now),
%% Undefined "later" time becomes erlang:monotonic_time(), which
%% is always >= Now on any single-node call sequence.
?assert(Diff >= 0).

now_diff_microseconds_test() ->
NativeUs = erlang:convert_time_unit(1, microsecond, native),
T1 = 1000000000,
T2 = T1 + 500 * NativeUs,
?assertEqual(500, now_diff_microseconds(T2, T1)).

now_diff_microseconds_undefined_test() ->
?assertEqual(0, now_diff_microseconds(undefined, undefined)),
Now = erlang:monotonic_time(),
Diff = now_diff_microseconds(undefined, Now),
?assert(Diff >= 0).

%% Record-shape contract test. Constructs a #wm_log_data{} with real
%% erlang:monotonic_time() values and passes the fields into
%% now_diff_{milli,micro}seconds/2 the way a downstream access-log
%% handler would. If the record's field types diverge from what these
%% helpers accept (as happened when the record migrated from
%% os:timestamp() tuples to native-unit integers), this test breaks at
%% compile time (record construction) or runtime (pattern match /
%% function clause), whichever comes first. That contract is what any
%% consumer reading start_time/end_time off the record relies on.
now_diff_wm_log_data_shape_test() ->
Start = erlang:monotonic_time(),
End = Start + erlang:convert_time_unit(1, millisecond, native),
LogData = #wm_log_data{start_time=Start, end_time=End},
#wm_log_data{start_time=StartTime, end_time=EndTime} = LogData,
?assertEqual(1, now_diff_milliseconds(EndTime, StartTime)),
?assertEqual(1000, now_diff_microseconds(EndTime, StartTime)).

%% Round-trip test: sleep a small interval, confirm the diff tracks
%% wall-clock. Catches unit-conversion bugs where a native → wrong-unit
%% conversion would silently return a nonsensical number (e.g.
%% mis-typed `nanosecond` or `second` for the target unit).
now_diff_milliseconds_realtime_test() ->
Start = erlang:monotonic_time(),
timer:sleep(15),
End = erlang:monotonic_time(),
DiffMs = now_diff_milliseconds(End, Start),
?assert(DiffMs >= 15),
%% Loose upper bound — a 15ms sleep should not take a second on
%% any realistic CI host. If this bound trips we have bigger
%% problems than unit conversion.
?assert(DiffMs < 1000).

parse_range_test() ->
ValidRange = "bytes=1-2",
Expand Down
Loading