Merge pull request #14238 from rabbitmq/optimise-amqp-parser

Make AMQP 1.0 faster

Author: ansd

deps/amqp10_client/src/amqp10_client_internal.hrl

@@ -7,7 +7,6 @@
 
 -define(AMQP_PROTOCOL_HEADER, <<"AMQP", 0, 1, 0, 0>>).
 -define(SASL_PROTOCOL_HEADER, <<"AMQP", 3, 1, 0, 0>>).
--define(FRAME_HEADER_SIZE, 8).
 
 -define(TIMEOUT, 5000).
 

deps/amqp10_client/src/amqp10_client_session.erl

@@ -625,8 +625,8 @@ send_transfer(Transfer0, Sections0, FooterOpt, MaxMessageSize,
                      channel = Channel,
                      connection_config = Config}) ->
     OutMaxFrameSize = maps:get(outgoing_max_frame_size, Config),
-    Transfer = amqp10_framing:encode_bin(Transfer0),
-    TransferSize = iolist_size(Transfer),
+    Transfer = Transfer0#'v1_0.transfer'{more = false},
+    TransferSize = iolist_size(amqp10_framing:encode_bin(Transfer)),
     Sections = encode_sections(Sections0, FooterOpt),
     SectionsBin = iolist_to_binary(Sections),
     if is_integer(MaxMessageSize) andalso
@@ -636,8 +636,8 @@ send_transfer(Transfer0, Sections0, FooterOpt, MaxMessageSize,
        true ->
            % TODO: this does not take the extended header into account
            % see: 2.3
-           MaxPayloadSize = OutMaxFrameSize - TransferSize - ?FRAME_HEADER_SIZE,
-           Frames = build_frames(Channel, Transfer0, SectionsBin, MaxPayloadSize, []),
+           MaxPayloadSize = OutMaxFrameSize - ?FRAME_HEADER_SIZE - TransferSize,
+           Frames = build_frames(Channel, Transfer, SectionsBin, MaxPayloadSize, []),
            ok = socket_send(Socket, Frames),
            {ok, length(Frames)}
     end.
@@ -722,7 +722,7 @@ set_flow_session_fields(Flow, #state{next_incoming_id = NID,
 
 build_frames(Channel, Trf, Bin, MaxPayloadSize, Acc)
   when byte_size(Bin) =< MaxPayloadSize ->
-    T = amqp10_framing:encode_bin(Trf#'v1_0.transfer'{more = false}),
+    T = amqp10_framing:encode_bin(Trf),
     Frame = amqp10_binary_generator:build_frame(Channel, [T, Bin]),
     lists:reverse([Frame | Acc]);
 build_frames(Channel, Trf, Payload, MaxPayloadSize, Acc) ->

deps/amqp10_common/include/amqp10_types.hrl

@@ -2,6 +2,10 @@
 
 % [1.6.5]
 -type uint() :: 0..?UINT_MAX.
+
+% [2.3.1]
+-define(FRAME_HEADER_SIZE, 8).
+
 % [2.8.4]
 -type link_handle() :: uint().
 % [2.8.8]

deps/amqp10_common/src/amqp10_binary_generator.erl

@@ -99,10 +99,10 @@ generate1({boolean, false}) -> [16#56, 16#00];
 %% bits set to zero and values < 256.
 generate1({ubyte,    V})                           -> [16#50, V];
 generate1({ushort,   V})                           -> <<16#60,V:16/unsigned>>;
-generate1({uint,     V}) when V =:= 0              -> 16#43;
+generate1({uint,     0})                           -> 16#43;
 generate1({uint,     V}) when V < 256              -> [16#52, V];
 generate1({uint,     V})                           -> <<16#70,V:32/unsigned>>;
-generate1({ulong,    V}) when V =:= 0              -> 16#44;
+generate1({ulong,    0})                           -> 16#44;
 generate1({ulong,    V}) when V < 256              -> [16#53, V];
 generate1({ulong,    V})                           -> <<16#80,V:64/unsigned>>;
 generate1({byte,     V})                           -> <<16#51,V:8/signed>>;
@@ -204,9 +204,9 @@ constructor(array) -> 16#f0; % use large array type for all nested arrays
 constructor({described, Descriptor, Primitive}) ->
     [16#00, generate1(Descriptor), constructor(Primitive)].
 
-generate2(symbol, {symbol, V}) -> [<<(size(V)):32>>, V];
-generate2(utf8, {utf8, V}) -> [<<(size(V)):32>>, V];
-generate2(binary, {binary, V}) -> [<<(size(V)):32>>, V];
+generate2(symbol, {symbol, V}) -> [<<(byte_size(V)):32>>, V];
+generate2(utf8, {utf8, V}) -> [<<(byte_size(V)):32>>, V];
+generate2(binary, {binary, V}) -> [<<(byte_size(V)):32>>, V];
 generate2(boolean, true) -> 16#01;
 generate2(boolean, false) -> 16#00;
 generate2(boolean, {boolean, true}) -> 16#01;