root
/
rabbitmq-server
mirror of https://github.com/rabbitmq/rabbitmq-server.git
1
0
Fork 0
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Remove erlang:port_command/2 hack 

as selective receives are efficient in OTP 26:
```
OTP-18431
Application(s):
compiler, stdlib
Related Id(s):
PR-6739
Improved the selective receive optimization, which can now be enabled for references returned from other functions.
This greatly improves the performance of gen_server:send_request/3, gen_server:wait_response/2, and similar functions.
```
This commit is contained in:
David Ansari 2023-06-13 15:32:07 +00:00
parent 61557d0747
commit cad067f5fa
7 changed files with 43 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
deps/rabbit_common/src/rabbit_net.erl vendored
View File

@ -11,7 +11,7 @@
-include_lib("kernel/include/net_address.hrl").
-export([is_ssl/1, ssl_info/1, controlling_process/2, getstat/2,
recv/1, sync_recv/2, async_recv/3, port_command/2, getopts/2,
recv/1, sync_recv/2, async_recv/3, getopts/2,
setopts/2, send/2, close/1, fast_close/1, sockname/1, peername/1,
peercert/1, connection_string/2, socket_ends/2, is_loopback/1,
tcp_host/1, unwrap_socket/1, maybe_get_proxy_socket/1,
@ -50,7 +50,6 @@
rabbit_types:error(any()).
-spec async_recv(socket(), integer(), timeout()) ->
rabbit_types:ok(any()).
-spec port_command(socket(), iolist()) -> 'true'.
-spec getopts
(socket(),
[atom() |
@ -58,7 +57,7 @@
non_neg_integer() | binary()}]) ->
ok_val_or_error(opts()).
-spec setopts(socket(), opts()) -> ok_or_any_error().
-spec send(socket(), binary() | iolist()) -> ok_or_any_error().
-spec send(socket(), iodata()) -> ok_or_any_error().
-spec close(socket()) -> ok_or_any_error().
-spec fast_close(socket()) -> ok_or_any_error().
-spec sockname(socket()) ->
@ -161,40 +160,6 @@ async_recv(Sock, Length, infinity) when is_port(Sock) ->
async_recv(Sock, Length, Timeout) when is_port(Sock) ->
prim_inet:async_recv(Sock, Length, Timeout).
port_command(Sock, Data) when ?IS_SSL(Sock) ->
case ssl:send(Sock, Data) of
ok -> self() ! {inet_reply, Sock, ok},
true;
{error, Reason} -> erlang:error(Reason)
end;
port_command(Sock, Data) when is_port(Sock) ->
Fun = case persistent_term:get(rabbit_net_tcp_send, undefined) of
undefined ->
Rel = list_to_integer(erlang:system_info(otp_release)),
%% gen_tcp:send/2 does a selective receive of
%% {inet_reply, Sock, Status[, CallerTag]}
F = if Rel >= 26 ->
%% Selective receive is optimised:
%% https://github.com/erlang/otp/issues/6455
fun gen_tcp_send/2;
Rel < 26 ->
%% Avoid costly selective receive.
fun erlang:port_command/2
end,
ok = persistent_term:put(rabbit_net_tcp_send, F),
F;
F ->
F
end,
Fun(Sock, Data).
gen_tcp_send(Sock, Data) ->
case gen_tcp:send(Sock, Data) of
ok -> self() ! {inet_reply, Sock, ok},
true;
{error, Reason} -> erlang:error(Reason)
end.
getopts(Sock, Options) when ?IS_SSL(Sock) ->
ssl:getopts(Sock, Options);
getopts(Sock, Options) when is_port(Sock) ->

40
deps/rabbit_common/src/rabbit_writer.erl vendored
View File

@ -268,10 +268,6 @@ handle_message({send_command_and_notify, QPid, ChPid, MethodRecord, Content},
handle_message({'DOWN', _MRef, process, QPid, _Reason}, State) ->
rabbit_amqqueue_common:notify_sent_queue_down(QPid),
State;
handle_message({inet_reply, _, ok}, State) ->
rabbit_event:ensure_stats_timer(State, #wstate.stats_timer, emit_stats);
handle_message({inet_reply, _, Status}, _State) ->
exit({writer, send_failed, Status});
handle_message(emit_stats, State = #wstate{reader = ReaderPid}) ->
ReaderPid ! ensure_stats,
rabbit_event:reset_stats_timer(State, #wstate.stats_timer);
@ -384,33 +380,15 @@ maybe_flush(State = #wstate{pending = Pending}) ->
internal_flush(State = #wstate{pending = []}) ->
State;
internal_flush(State = #wstate{sock = Sock, pending = Pending}) ->
ok = port_cmd(Sock, lists:reverse(Pending)),
State#wstate{pending = []}.
%% gen_tcp:send/2 does a selective receive of {inet_reply, Sock,
%% Status} to obtain the result. That is bad when it is called from
%% the writer since it requires scanning of the writers possibly quite
%% large message queue.
%%
%% So instead we lift the code from prim_inet:send/2, which is what
%% gen_tcp:send/2 calls, do the first half here and then just process
%% the result code in handle_message/2 as and when it arrives.
%%
%% This means we may end up happily sending data down a closed/broken
%% socket, but that's ok since a) data in the buffers will be lost in
%% any case (so qualitatively we are no worse off than if we used
%% gen_tcp:send/2), and b) we do detect the changed socket status
%% eventually, i.e. when we get round to handling the result code.
%%
%% Also note that the port has bounded buffers and port_command blocks
%% when these are full. So the fact that we process the result
%% asynchronously does not impact flow control.
port_cmd(Sock, Data) ->
true = try rabbit_net:port_command(Sock, Data)
catch error:Error -> exit({writer, send_failed, Error})
end,
ok.
internal_flush(State0 = #wstate{sock = Sock, pending = Pending}) ->
case rabbit_net:send(Sock, lists:reverse(Pending)) of
ok ->
ok;
{error, Reason} ->
exit({writer, send_failed, Reason})
end,
State = State0#wstate{pending = []},
rabbit_event:ensure_stats_timer(State, #wstate.stats_timer, emit_stats).
%% Some processes (channel, writer) can get huge amounts of binary
%% garbage when processing huge messages at high speed (since we only

40
deps/rabbitmq_amqp1_0/src/rabbit_amqp1_0_writer.erl vendored
View File

@ -142,10 +142,6 @@ handle_message({send_command_and_notify, QPid, ChPid, MethodRecord, Content},
handle_message({'DOWN', _MRef, process, QPid, _Reason}, State) ->
rabbit_amqqueue:notify_sent_queue_down(QPid),
State;
handle_message({inet_reply, _, ok}, State) ->
rabbit_event:ensure_stats_timer(State, #wstate.stats_timer, emit_stats);
handle_message({inet_reply, _, Status}, _State) ->
exit({writer, send_failed, Status});
handle_message(emit_stats, State = #wstate{reader = ReaderPid}) ->
ReaderPid ! ensure_stats,
rabbit_event:reset_stats_timer(State, #wstate.stats_timer);
@ -251,30 +247,12 @@ maybe_flush(State = #wstate{pending = Pending}) ->
flush(State = #wstate{pending = []}) ->
State;
flush(State = #wstate{sock = Sock, pending = Pending}) ->
ok = port_cmd(Sock, lists:reverse(Pending)),
State#wstate{pending = []}.
%% gen_tcp:send/2 does a selective receive of {inet_reply, Sock,
%% Status} to obtain the result. That is bad when it is called from
%% the writer since it requires scanning of the writers possibly quite
%% large message queue.
%%
%% So instead we lift the code from prim_inet:send/2, which is what
%% gen_tcp:send/2 calls, do the first half here and then just process
%% the result code in handle_message/2 as and when it arrives.
%%
%% This means we may end up happily sending data down a closed/broken
%% socket, but that's ok since a) data in the buffers will be lost in
%% any case (so qualitatively we are no worse off than if we used
%% gen_tcp:send/2), and b) we do detect the changed socket status
%% eventually, i.e. when we get round to handling the result code.
%%
%% Also note that the port has bounded buffers and port_command blocks
%% when these are full. So the fact that we process the result
%% asynchronously does not impact flow control.
port_cmd(Sock, Data) ->
true = try rabbit_net:port_command(Sock, Data)
catch error:Error -> exit({writer, send_failed, Error})
end,
ok.
flush(State0 = #wstate{sock = Sock, pending = Pending}) ->
case rabbit_net:send(Sock, lists:reverse(Pending)) of
ok ->
ok;
{error, Reason} ->
exit({writer, send_failed, Reason})
end,
State = State0#wstate{pending = []},
rabbit_event:ensure_stats_timer(State, #wstate.stats_timer, emit_stats).

21
deps/rabbitmq_mqtt/src/rabbit_mqtt_reader.erl vendored
View File

@ -193,12 +193,6 @@ handle_info({Tag, Sock, Reason}, State = #state{socket = Sock})
when Tag =:= tcp_error; Tag =:= ssl_error ->
network_error(Reason, State);
handle_info({inet_reply, Sock, ok}, State = #state{socket = Sock}) ->
{noreply, State, ?HIBERNATE_AFTER};
handle_info({inet_reply, Sock, {error, Reason}}, State = #state{socket = Sock}) ->
network_error(Reason, State);
handle_info({conserve_resources, Conserve}, State) ->
maybe_process_deferred_recv(
control_throttle(State #state{ conserve = Conserve }));
@ -335,13 +329,14 @@ process_received_bytes(Bytes, State = #state{socket = Socket,
case ProcState of
connect_packet_unprocessed ->
Send = fun(Data) ->
try rabbit_net:port_command(Socket, Data)
catch error:Reason ->
?LOG_ERROR("writing to MQTT socket ~p failed: ~p",
[Socket, Reason]),
exit({send_failed, Reason})
end,
ok
case rabbit_net:send(Socket, Data) of
ok ->
ok;
{error, Reason} ->
?LOG_ERROR("writing to MQTT socket ~p failed: ~p",
[Socket, Reason]),
exit({send_failed, Reason})
end
end,
try rabbit_mqtt_processor:init(Packet, Socket, ConnName, Send) of
{ok, ProcState1} ->

4
deps/rabbitmq_stomp/src/rabbit_stomp_processor.erl vendored
View File

@ -46,7 +46,7 @@ adapter_name(State) ->
#stomp_configuration{},
{SendFun, AdapterInfo, SSLLoginName, PeerAddr})
-> #proc_state{}
when SendFun :: fun((atom(), binary()) -> term()),
when SendFun :: fun((binary()) -> term()),
AdapterInfo :: #amqp_adapter_info{},
SSLLoginName :: atom() | binary(),
PeerAddr :: inet:ip_address().
@ -1174,7 +1174,7 @@ send_frame(Command, Headers, BodyFragments, State) ->
send_frame(Frame, State = #proc_state{send_fun = SendFun,
trailing_lf = TrailingLF}) ->
SendFun(async, rabbit_stomp_frame:serialize(Frame, TrailingLF)),
SendFun(rabbit_stomp_frame:serialize(Frame, TrailingLF)),
State.
send_error_frame(Message, ExtraHeaders, Format, Args, State) ->

29
deps/rabbitmq_stomp/src/rabbit_stomp_reader.erl vendored
View File

@ -140,12 +140,6 @@ handle_info({Tag, Sock}, State=#reader_state{socket=Sock})
handle_info({Tag, Sock, Reason}, State=#reader_state{socket=Sock})
when Tag =:= tcp_error; Tag =:= ssl_error ->
{stop, {inet_error, Reason}, State};
handle_info({inet_reply, _Sock, {error, closed}}, State) ->
{stop, normal, State};
handle_info({inet_reply, _, ok}, State) ->
{noreply, State, hibernate};
handle_info({inet_reply, _, Status}, State) ->
{stop, Status, State};
handle_info(emit_stats, State) ->
{noreply, emit_stats(State), hibernate};
handle_info({conserve_resources, Conserve}, State) ->
@ -259,7 +253,7 @@ process_received_bytes(Bytes,
log_reason({network_error, {frame_too_big, {FrameLength1, MaxFrameSize}}}, State),
{stop, normal, State};
false ->
case rabbit_stomp_processor:process_frame(Frame, ProcState) of
try rabbit_stomp_processor:process_frame(Frame, ProcState) of
{ok, NewProcState, Conn} ->
PS = rabbit_stomp_frame:initial_state(),
NextState = maybe_block(State, Frame),
@ -271,6 +265,10 @@ process_received_bytes(Bytes,
{stop, Reason, NewProcState} ->
{stop, Reason,
processor_state(NewProcState, State)}
catch exit:{send_failed, closed} ->
{stop, normal, State};
exit:{send_failed, Reason} ->
{stop, Reason, State}
end
end;
{error, Reason} ->
@ -404,16 +402,13 @@ log_tls_alert(Alert, ConnName) ->
processor_args(Configuration, Sock) ->
RealSocket = rabbit_net:unwrap_socket(Sock),
SendFun = fun (sync, IoData) ->
%% no messages emitted
catch rabbit_net:send(RealSocket, IoData);
(async, IoData) ->
%% {inet_reply, _, _} will appear soon
%% We ignore certain errors here, as we will be
%% receiving an asynchronous notification of the
%% same (or a related) fault shortly anyway. See
%% bug 21365.
catch rabbit_net:port_command(RealSocket, IoData)
SendFun = fun(IoData) ->
case rabbit_net:send(RealSocket, IoData) of
ok ->
ok;
{error, Reason} ->
exit({send_failed, Reason})
end
end,
{ok, {PeerAddr, _PeerPort}} = rabbit_net:sockname(RealSocket),
{SendFun, adapter_info(Sock),

2
deps/rabbitmq_web_stomp/src/rabbit_web_stomp_handler.erl vendored
View File

@ -127,7 +127,7 @@ close_connection(Pid, Reason) ->
init_processor_state(#state{socket=Sock, peername=PeerAddr, auth_hd=AuthHd}) ->
Self = self(),
SendFun = fun (_Sync, Data) ->
SendFun = fun(Data) ->
Self ! {send, Data},
ok
end,