4.x: change AMQP on disk message format & speed up the AMQP parser

[ansd]

This PR comes with the following benefits

1. Immutable bare message

According to the AMQP spec:

The bare message is immutable within the AMQP network. That is, none of the sections can be changed by any node acting as an AMQP intermediary. If a section of the bare message is omitted, one MUST NOT be inserted by an intermediary. The exact encoding of sections of the bare message MUST NOT be modified. This preserves message hashes, HMACs and signatures based on the binary encoding of the bare message.

Prior to this commit, this requirement was violated by RabbitMQ.
RabbitMQ used to decode the bare message from the sending client and encode the bare message before sending to the receiving client. RabbitMQ violated the immutability of the bare message for example:

1. by not respecting the 1 vs 2 byte AMQP boolean encoding, and
2. by always providing the optional null value in the trailing fields to indicate absence of a value

This PR ensures that the bare message is never modified if the message is sent and consumed by AMQP clients.

2. End-to-end checksumming

Thanks to 1. Immutable bare message, this PR adds a new feature to the AMQP Erlang client: end-to-end checksumming over the bare message (i.e. body + application defined headers).

This PR allows an app to configure the AMQP client:

• for a sending link to automatically compute CRC-32 or Adler-32
  checksums over each bare message including the computed checksum
  as a footer annotation, and
• for a receiving link to automatically lookup the expected CRC-32
  or Adler-32 checksum in the footer annotation and, if present, check
  the received checksum against the actually computed checksum.

The feature comes with the following advantages:

1. Transparent end-to-end checksumming. Although checksumming is
   performed by TCP and RabbitMQ queues using the disk, end-to-end
   checksumming is a level higher up and can therefore detect bit flips
   within RabbitMQ nodes or load balancers and other bit flips that
   went unnoticed.
2. Not only is the body checksummed, but also the properties and
   application-properties sections. This is an advantage over AMQP 0.9.1
   because the AMQP protocol disallows modification of the bare message.
3. This feature is currently used for testing the RabbitMQ AMQP
   implementation, but it shows the feasibility of how apps could also
   get integrity guarantees of the whole bare message using HMACs or
   signatures.

3. Avoid parsing the AMQP body

Do not parse entire AMQP body.

Prior to this PR the entire amqp-value or amqp-sequence sections were parsed when receiving a message from a client.
Parsing the entire amqp-value or amqp-sequence sections can generate a huge amount of garbage depending on how large these sections are.

Not only will this PR not parse the amqp-value or amqp-sequence sections when receiving a message from a client, but also it won't parse the sections when converting from mc_amqp to another protocol, such as mc_amqpl or mc_mqtt. Given that other protocols cannot make use of amqp-value and amqp-sequence sections anyway, leave them AMQP encoded when converting from mc_amqp.

In fact prior to this commit, the entire body section was parsed generating huge amounts of garbage just to subsequently encode it again in mc_amqpl or mc_mqtt.

The new conversion interface from mc_amqp to other mc_* modules will either output amqp-data sections or the encoded amqp-value / amqp-sequence sections.

4. Faster AMQP parser

The AMQP parser got optimised.

When compiling file amqp10_binary_parser.erl with ERL_COMPILER_OPTIONS=bin_opt_info, all code now outputs OPTIMIZED: match context reused instead of BINARY CREATED or NOT OPTIMIZED. The only exception are arrays since arrays aren't used in the hot path.

Benchmarks

1. To remove artificial bottlenecks, increase link credit RabbitMQ grants to the sending clients and the session window as follows:

diff --git a/deps/rabbit/src/rabbit_amqp_session.erl b/deps/rabbit/src/rabbit_amqp_session.erl
index f26802be46..5c3ef1151e 100644
--- a/deps/rabbit/src/rabbit_amqp_session.erl
+++ b/deps/rabbit/src/rabbit_amqp_session.erl
@@ -29,7 +29,7 @@
 -define(HIBERNATE_AFTER, 6_000).
 -define(CREDIT_REPLY_TIMEOUT, 30_000).
 -define(UINT_OUTGOING_WINDOW, {uint, ?UINT_MAX}).
--define(MAX_INCOMING_WINDOW, 400).
+-define(MAX_INCOMING_WINDOW, 5000).
 %% "The next-outgoing-id MAY be initialized to an arbitrary value" [2.5.6]
 -define(INITIAL_OUTGOING_TRANSFER_ID, ?UINT_MAX - 3).
 %% "Note that, despite its name, the delivery-count is not a count but a
@@ -55,7 +55,7 @@
 %% on target queue type. For the time being just use a static value that's something in between.
 %% An even better approach in future would be to dynamically grow (or shrink) the link credit
 %% we grant depending on how fast target queue(s) actually confirm messages.
--define(LINK_CREDIT_RCV, 128).
+-define(LINK_CREDIT_RCV, 5000).
 -define(MANAGEMENT_LINK_CREDIT_RCV, 8).
 -define(MANAGEMENT_NODE_ADDRESS, <<"/management">>).
 -define(DEFAULT_EXCHANGE_NAME, <<>>).

2. Start RabbitMQ and create 2 classic queues:

make run-broker PLUGINS="rabbitmq_management" FULL=1 RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+S 1 +JPperf true"
deps/rabbitmq_management/bin/rabbitmqadmin declare queue name=cq1 queue_type=classic durable=false
deps/rabbitmq_management/bin/rabbitmqadmin declare queue name=cq2 queue_type=classic durable=true
deps/rabbitmq_management/bin/rabbitmqadmin declare queue name=qq queue_type=quorum durable=true

3. Start quiver:

docker run -it --rm --add-host host.docker.internal:host-gateway ssorj/quiver:latest
bash-5.1# quiver --version
quiver 0.4.0-SNAPSHOT

4. Benchmark 1: classic queue with transient messages:

bash-5.1# quiver //host.docker.internal//queue/cq1 --duration 60s --body-size 12 --credit 10000

Results with PR:

Count ............................................. 3,699,634 messages
Duration ............................................... 60.0 seconds
Sender rate .......................................... 61,675 messages/s
Receiver rate ........................................ 61,666 messages/s
End-to-end rate ...................................... 61,665 messages/s

Results prior to PR:

Count ............................................. 3,460,189 messages
Duration ............................................... 60.0 seconds
Sender rate .......................................... 57,713 messages/s
Receiver rate ........................................ 57,675 messages/s
End-to-end rate ...................................... 57,673 messages/s

5. Benchmark 2: classic queue with persistent messages:

bash-5.1# quiver //host.docker.internal//queue/cq2 --duration 60s --body-size 12 --credit 10000 --durable

Results with PR:

Count ............................................. 2,450,989 messages
Duration ............................................... 60.0 seconds
Sender rate .......................................... 40,973 messages/s
Receiver rate ........................................ 40,856 messages/s
End-to-end rate ...................................... 40,851 messages/s

Results prior to PR:

Count ............................................. 1,990,530 messages
Duration ............................................... 60.0 seconds
Sender rate .......................................... 33,301 messages/s
Receiver rate ........................................ 33,182 messages/s
End-to-end rate ...................................... 33,177 messages/s

Using persistent messages and classic queues, the end to end throughput increase is 23% with this PR.

For the last benchmark, the CPU flame graphs show that this PR spends only 2.6% instead of 5.5% in function rabbit_amqp_reader:parse_frame_body/2 and that this PR spends 18.9% instead of 25.6% in garbage_collect.

6. Benchmark 3: quorum queue with persistent messages:

bash-5.1# quiver //host.docker.internal//queue/qq --duration 60s --body-size 12 --credit 10000 --durable

Results with PR:

Count ............................................. 3,103,960 messages
Duration ............................................... 60.0 seconds
Sender rate .......................................... 51,875 messages/s
Receiver rate ........................................ 51,774 messages/s
End-to-end rate ...................................... 51,740 messages/s

Results prior to PR:

Count ............................................. 2,487,133 messages
Duration ............................................... 60.0 seconds
Sender rate .......................................... 41,603 messages/s
Receiver rate ........................................ 41,486 messages/s
End-to-end rate ...................................... 41,459 messages/s

Using quorum queues, the end to end throughput increase is 24% with this PR.

[kjnilsson]

Looks good to me - a couple of optional comments only.
I can't say I have evaluated every single line of code fully but the approach is good.

Still not 100% sure that concatenating multiple data sections is the right way but it should be extremely rare I hope.