Add gun:ping/2,3 for user-initiated ping for HTTP/2

Signed-off-by: Viktor Söderqvist <[email protected]>

doc/src/manual/gun.ping.asciidoc

@@ -0,0 +1,68 @@
+= gun:ping(3)
+
+== Name
+
+gun:ping - Check the health or the round-trip time of a connection
+without sending a request.
+
+== Description
+
+[source,erlang]
+----
+ping(ConnPid)
+    -> ping(ConnPid, #{})
+
+ping(ConnPid, ReqOpts)
+    -> PingRef
+
+ConnPid   :: pid()
+ReqOpts   :: gun:req_opts()
+PingRef   :: gun:stream_ref()
+----
+
+Send a ping.
+
+A ping can be sent to check the health or to measure the
+round-trip time of a connection, without sending a request.
+
+The function `ping/1,2` sends a ping immediately, if the
+protocol supports pings. The server responds with a ping ack.
+A call to `gun:await/2,3` returns `ping_ack` when the ping
+ack has been received from the server.
+
+Currently, explicit ping is supported only for HTTP/2.
+
+== Arguments
+
+ConnPid::
+
+The pid of the Gun connection process.
+
+ReqOpts::
+
+Request options. Only the `reply_to` and `tunnel` options
+can be used.
+
+== Return value
+
+A reference that identifies the ping is returned. This
+reference must be passed in subsequent calls and will be
+received in messages related to this ping.
+
+== Changelog
+
+* *2.x*: Function introduced.
+
+== Examples
+
+.Perform a request
+[source,erlang]
+----
+PingRef = gun:ping(ConnPid).
+ping_ack = gun:await(ConnPid, PingRef).
+----
+
+== See also
+
+link:man:gun(3)[gun(3)],
+link:man:gun:await(3)[gun:await(3)],

src/gun.erl

@@ -60,6 +60,10 @@
 %% Streaming data.
 -export([data/4]).
 
+%% User pings.
+-export([ping/1]).
+-export([ping/2]).
+
 %% Tunneling.
 -export([connect/2]).
 -export([connect/3]).
@@ -721,6 +725,20 @@ data(ServerPid, StreamRef, IsFin, Data) ->
 			gen_statem:cast(ServerPid, {data, self(), StreamRef, IsFin, Data})
 	end.
 
+%% User pings.
+
+-spec ping(pid()) -> stream_ref().
+ping(ServerPid) ->
+	ping(ServerPid, #{}).
+
+-spec ping(pid(), req_opts()) -> stream_ref().
+ping(ServerPid, ReqOpts) ->
+	Tunnel = get_tunnel(ReqOpts),
+	StreamRef = make_stream_ref(Tunnel),
+	ReplyTo = maps:get(reply_to, ReqOpts, self()),
+	gen_statem:cast(ServerPid, {ping, ReplyTo, StreamRef}),
+	StreamRef.
+
 %% Tunneling.
 
 -spec connect(pid(), connect_destination()) -> stream_ref().
@@ -782,6 +800,8 @@ await(ServerPid, StreamRef, Timeout, MRef) ->
 			{up, Protocol};
 		{gun_notify, ServerPid, Type, Info} ->
 			{notify, Type, Info};
+		{gun_ping_ack, ServerPid, StreamRef} ->
+			ping_ack;
 		{gun_error, ServerPid, StreamRef, Reason} ->
 			{error, {stream_error, Reason}};
 		{gun_error, ServerPid, Reason} ->
@@ -1367,6 +1387,15 @@ connected_ws_only(Type, Event, State) ->
 %%
 %% @todo It might be better, internally, to pass around a URIMap
 %% containing the target URI, instead of separate Host/Port/PathWithQs.
+connected(cast, {ping, ReplyTo, StreamRef},
+		State=#state{protocol=Protocol, protocol_state=ProtoState}) ->
+	case erlang:function_exported(Protocol, ping, 3) of
+		true ->
+			Commands = Protocol:ping(ProtoState, dereference_stream_ref(StreamRef, State), ReplyTo),
+			commands(Commands, State);
+		false ->
+			ReplyTo ! {gun_error, self(), StreamRef, not_supported_for_protocol}
+	end;
 connected(cast, {headers, ReplyTo, StreamRef, Method, Path, Headers, InitialFlow},
 		State=#state{origin_host=Host, origin_port=Port,
 			protocol=Protocol, protocol_state=ProtoState, cookie_store=CookieStore0,

src/gun_http2.erl

@@ -27,6 +27,7 @@
 -export([closing/4]).
 -export([close/4]).
 -export([keepalive/3]).
+-export([ping/3]).
 -export([headers/12]).
 -export([request/13]).
 -export([data/7]).
@@ -82,6 +83,12 @@
 	tunnel :: undefined | #tunnel{}
 }).
 
+-record(user_ping, {
+	ref :: reference(),
+	reply_to :: pid(),
+	payload :: integer()
+}).
+
 -record(http2_state, {
 	reply_to :: pid(),
 	socket :: inet:socket() | ssl:sslsocket(),
@@ -115,6 +122,9 @@
 	streams = #{} :: #{cow_http2:streamid() => #stream{}},
 	stream_refs = #{} :: #{reference() => cow_http2:streamid()},
 
+	%% User-initiated pings that have not yet been acknowledged.
+	user_pings = [] :: [#user_ping{}],
+
 	%% Number of pings that have been sent but not yet acknowledged.
 	%% Used to determine whether the connection should be closed when
 	%% the keepalive_tolerance option is set.
@@ -351,7 +361,7 @@ frame(State=#http2_state{http2_machine=HTTP2Machine0}, Frame, CookieStore, EvHan
 
 maybe_ack_or_notify(State=#http2_state{reply_to=ReplyTo, socket=Socket,
 		transport=Transport, opts=Opts, http2_machine=HTTP2Machine,
-		pings_unack=PingsUnack}, Frame) ->
+		pings_unack=PingsUnack, user_pings=UserPings0}, Frame) ->
 	case Frame of
 		{settings, _} ->
 			%% We notify remote settings changes only if the user requested it.
@@ -371,8 +381,21 @@ maybe_ack_or_notify(State=#http2_state{reply_to=ReplyTo, socket=Socket,
 				ok -> {state, State};
 				Error={error, _} -> Error
 			end;
-		{ping_ack, _Opaque} ->
+		{ping_ack, 0} ->
+			%% Internal ping payload used for keepalive.
 			{state, State#http2_state{pings_unack=PingsUnack - 1}};
+		{ping_ack, Payload} ->
+			%% User ping.
+			case lists:keytake(Payload, #user_ping.payload, UserPings0) of
+				{value, #user_ping{ref=StreamRef, reply_to=PingReplyTo}, UserPings} ->
+					RealStreamRef = stream_ref(State, StreamRef),
+					PingReplyTo ! {gun_ping_ack, self(), RealStreamRef},
+					{state, State#http2_state{user_pings=UserPings}};
+				false ->
+					%% Ignore unexpected ping ack. RFC 7540
+					%% doesn't explicitly forbid it.
+					{state, State}
+			end;
 		_ ->
 			{state, State}
 	end.
@@ -934,6 +957,20 @@ keepalive(State=#http2_state{socket=Socket, transport=Transport, pings_unack=Pin
 			{Error, EvHandlerState}
 	end.
 
+ping(State=#http2_state{socket=Socket, transport=Transport, user_pings=UserPings}, StreamRef, ReplyTo) ->
+	%% Use non-zero 64-bit payload for user pings. 0 is reserved for keepalive.
+	Payload = case erlang:monotonic_time(microsecond) band 16#ffffffffffffffff of
+		0 -> 1;
+		Payload0 -> Payload0
+	end,
+	case Transport:send(Socket, cow_http2:ping(Payload)) of
+		ok ->
+			UserPing = #user_ping{ref = StreamRef, reply_to = ReplyTo, payload = Payload},
+			{state, State#http2_state{user_pings = UserPings ++ [UserPing]}};
+		Error={error, _} ->
+			Error
+	end.
+
 headers(State=#http2_state{socket=Socket, transport=Transport, opts=Opts,
 		http2_machine=HTTP2Machine0}, StreamRef, ReplyTo, Method, Host, Port,
 		Path, Headers0, InitialFlow0, CookieStore0, EvHandler, EvHandlerState0)

test/rfc7540_SUITE.erl

@@ -471,6 +471,30 @@ keepalive_tolerance_ping_ack_timeout(_) ->
 		error(timeout)
 	end.
 
+user_initiated_ping(_) ->
+	doc("The PING frame allows a client to safely test whether a connection is"
+            " still active without sending a request. (RFC7540 8.1.4)"),
+	{ok, OriginPid, OriginPort} = init_origin(tcp, http2, fun (_, _, Socket, Transport) ->
+		{ok, Data} = Transport:recv(Socket, 9, infinity),
+		<<Len:24, 6:8, %% PING
+			0:8, %% Flags
+			0:1, 0:31>> = Data,
+		{ok, Payload} = Transport:recv(Socket, Len, 1000),
+		8 = Len = byte_size(Payload),
+		Ack = <<8:24, 6:8, %% PING
+			1:8, %% Ack flag
+			0:1, 0:31, Payload/binary>>,
+		ok = Transport:send(Socket, Ack)
+	end),
+	{ok, Pid} = gun:open("localhost", OriginPort, #{
+		protocols => [http2]
+	}),
+	{ok, http2} = gun:await_up(Pid),
+	handshake_completed = receive_from(OriginPid),
+	PingRef = gun:ping(Pid),
+	ping_ack = gun:await(Pid, PingRef, 1000),
+	gun:close(Pid).
+
 do_ping_ack_loop_fun() ->
 	%% Receive ping, sync with parent, send ping ack, loop.
 	fun Loop(Parent, ListenSocket, Socket, Transport) ->