Skip to content

Commit c92df5b

Browse files
ansdmichaelklishin
ansd
authored and
michaelklishin
committed
Speed up AMQP 1.0 parser by generating less garbage
This commit results in great performance improvements for AMQP 1.0. Stream filtering via AMQP SQL Filters or AMQP Property Filters where the broker reads messages from the stream into memory and the filter returns false, i.e. messages are not sent to the client: Before this commit: ~400,000 msgs/s After this commit: ~500,000 msgs/s There is also a ~10% increase in end-to-end throughput for normal AMQP workloads, e.g. when sending to and receiving from a classic queue. Prior to this commit, a lot of garbage was created leading to many minor garbage collections. This commit reduces the garbage being generated. The new module amqp10_composite performs very well: * Single tuple update setting multiple fields at once, sometimes done even in-place without copying the tuple. * The list of fields is passed into X registers, no need for any new allocations. On the serialisation side, this commit also generates less garbage by: 1. Avoiding tuple_to_list/1 2. Omitting trailing elements of the list that are null This will also lead to fewer bytes per message sent on the wire and less resource usage for the clients as they need to parse fewer fields.
1 parent e96a9ff commit c92df5b

File tree

3 files changed

+488
-8
lines changed

3 files changed

+488
-8
lines changed

‎deps/amqp10_common/src/amqp10_composite.erl‎

Lines changed: 336 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,336 @@
1+
%% This Source Code Form is subject to the terms of the Mozilla Public
2+
%% License, v. 2.0. If a copy of the MPL was not distributed with this
3+
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
4+
%%
5+
%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
6+
%%
7+
8+
-module(amqp10_composite).
9+
10+
-include("amqp10_framing.hrl").
11+
12+
-export([flow/2,
13+
transfer/2,
14+
disposition/2,
15+
header/2,
16+
properties/2]).
17+
18+
-spec flow(#'v1_0.flow'{}, nonempty_list()) ->
19+
#'v1_0.flow'{}.
20+
flow(F, [F1, F2, F3, F4]) ->
21+
F#'v1_0.flow'{next_incoming_id = ntu(F1),
22+
incoming_window = ntu(F2),
23+
next_outgoing_id = ntu(F3),
24+
outgoing_window = ntu(F4)};
25+
flow(F, [F1, F2, F3, F4, F5]) ->
26+
F#'v1_0.flow'{next_incoming_id = ntu(F1),
27+
incoming_window = ntu(F2),
28+
next_outgoing_id = ntu(F3),
29+
outgoing_window = ntu(F4),
30+
handle = ntu(F5)};
31+
flow(F, [F1, F2, F3, F4, F5, F6]) ->
32+
F#'v1_0.flow'{next_incoming_id = ntu(F1),
33+
incoming_window = ntu(F2),
34+
next_outgoing_id = ntu(F3),
35+
outgoing_window = ntu(F4),
36+
handle = ntu(F5),
37+
delivery_count = ntu(F6)};
38+
flow(F, [F1, F2, F3, F4, F5, F6, F7]) ->
39+
F#'v1_0.flow'{next_incoming_id = ntu(F1),
40+
incoming_window = ntu(F2),
41+
next_outgoing_id = ntu(F3),
42+
outgoing_window = ntu(F4),
43+
handle = ntu(F5),
44+
delivery_count = ntu(F6),
45+
link_credit = ntu(F7)};
46+
flow(F, [F1, F2, F3, F4, F5, F6, F7, F8]) ->
47+
F#'v1_0.flow'{next_incoming_id = ntu(F1),
48+
incoming_window = ntu(F2),
49+
next_outgoing_id = ntu(F3),
50+
outgoing_window = ntu(F4),
51+
handle = ntu(F5),
52+
delivery_count = ntu(F6),
53+
link_credit = ntu(F7),
54+
available = ntu(F8)};
55+
flow(F, [F1, F2, F3, F4, F5, F6, F7, F8, F9]) ->
56+
F#'v1_0.flow'{next_incoming_id = ntu(F1),
57+
incoming_window = ntu(F2),
58+
next_outgoing_id = ntu(F3),
59+
outgoing_window = ntu(F4),
60+
handle = ntu(F5),
61+
delivery_count = ntu(F6),
62+
link_credit = ntu(F7),
63+
available = ntu(F8),
64+
drain = ntu(F9)};
65+
flow(F, [F1, F2, F3, F4, F5, F6, F7, F8, F9, F10]) ->
66+
F#'v1_0.flow'{next_incoming_id = ntu(F1),
67+
incoming_window = ntu(F2),
68+
next_outgoing_id = ntu(F3),
69+
outgoing_window = ntu(F4),
70+
handle = ntu(F5),
71+
delivery_count = ntu(F6),
72+
link_credit = ntu(F7),
73+
available = ntu(F8),
74+
drain = ntu(F9),
75+
echo = ntu(F10)};
76+
flow(F, [F1, F2, F3, F4, F5, F6, F7, F8, F9, F10, F11]) ->
77+
F#'v1_0.flow'{next_incoming_id = ntu(F1),
78+
incoming_window = ntu(F2),
79+
next_outgoing_id = ntu(F3),
80+
outgoing_window = ntu(F4),
81+
handle = ntu(F5),
82+
delivery_count = ntu(F6),
83+
link_credit = ntu(F7),
84+
available = ntu(F8),
85+
drain = ntu(F9),
86+
echo = ntu(F10),
87+
properties = amqp10_framing:decode(F11)}.
88+
89+
-spec transfer(#'v1_0.transfer'{}, nonempty_list()) ->
90+
#'v1_0.transfer'{}.
91+
transfer(T, [F1]) ->
92+
T#'v1_0.transfer'{handle = ntu(F1)};
93+
transfer(T, [F1, F2]) ->
94+
T#'v1_0.transfer'{handle = ntu(F1),
95+
delivery_id = ntu(F2)};
96+
transfer(T, [F1, F2, F3]) ->
97+
T#'v1_0.transfer'{handle = ntu(F1),
98+
delivery_id = ntu(F2),
99+
delivery_tag = ntu(F3)};
100+
transfer(T, [F1, F2, F3, F4]) ->
101+
T#'v1_0.transfer'{handle = ntu(F1),
102+
delivery_id = ntu(F2),
103+
delivery_tag = ntu(F3),
104+
message_format = ntu(F4)};
105+
transfer(T, [F1, F2, F3, F4, F5]) ->
106+
T#'v1_0.transfer'{handle = ntu(F1),
107+
delivery_id = ntu(F2),
108+
delivery_tag = ntu(F3),
109+
message_format = ntu(F4),
110+
settled = ntu(F5)};
111+
transfer(T, [F1, F2, F3, F4, F5, F6]) ->
112+
T#'v1_0.transfer'{handle = ntu(F1),
113+
delivery_id = ntu(F2),
114+
delivery_tag = ntu(F3),
115+
message_format = ntu(F4),
116+
settled = ntu(F5),
117+
more = ntu(F6)};
118+
transfer(T, [F1, F2, F3, F4, F5, F6, F7]) ->
119+
T#'v1_0.transfer'{handle = ntu(F1),
120+
delivery_id = ntu(F2),
121+
delivery_tag = ntu(F3),
122+
message_format = ntu(F4),
123+
settled = ntu(F5),
124+
more = ntu(F6),
125+
rcv_settle_mode = ntu(F7)};
126+
transfer(T, [F1, F2, F3, F4, F5, F6, F7, F8]) ->
127+
T#'v1_0.transfer'{handle = ntu(F1),
128+
delivery_id = ntu(F2),
129+
delivery_tag = ntu(F3),
130+
message_format = ntu(F4),
131+
settled = ntu(F5),
132+
more = ntu(F6),
133+
rcv_settle_mode = ntu(F7),
134+
state = amqp10_framing:decode(F8)};
135+
transfer(T, [F1, F2, F3, F4, F5, F6, F7, F8, F9]) ->
136+
T#'v1_0.transfer'{handle = ntu(F1),
137+
delivery_id = ntu(F2),
138+
delivery_tag = ntu(F3),
139+
message_format = ntu(F4),
140+
settled = ntu(F5),
141+
more = ntu(F6),
142+
rcv_settle_mode = ntu(F7),
143+
state = amqp10_framing:decode(F8),
144+
resume = ntu(F9)};
145+
transfer(T, [F1, F2, F3, F4, F5, F6, F7, F8, F9, F10]) ->
146+
T#'v1_0.transfer'{handle = ntu(F1),
147+
delivery_id = ntu(F2),
148+
delivery_tag = ntu(F3),
149+
message_format = ntu(F4),
150+
settled = ntu(F5),
151+
more = ntu(F6),
152+
rcv_settle_mode = ntu(F7),
153+
state = amqp10_framing:decode(F8),
154+
resume = ntu(F9),
155+
aborted = ntu(F10)};
156+
transfer(T, [F1, F2, F3, F4, F5, F6, F7, F8, F9, F10, F11]) ->
157+
T#'v1_0.transfer'{handle = ntu(F1),
158+
delivery_id = ntu(F2),
159+
delivery_tag = ntu(F3),
160+
message_format = ntu(F4),
161+
settled = ntu(F5),
162+
more = ntu(F6),
163+
rcv_settle_mode = ntu(F7),
164+
state = amqp10_framing:decode(F8),
165+
resume = ntu(F9),
166+
aborted = ntu(F10),
167+
batchable = ntu(F11)}.
168+
169+
-spec disposition(#'v1_0.disposition'{}, nonempty_list()) ->
170+
#'v1_0.disposition'{}.
171+
disposition(D, [F1, F2]) ->
172+
D#'v1_0.disposition'{role = ntu(F1),
173+
first = ntu(F2)};
174+
disposition(D, [F1, F2, F3]) ->
175+
D#'v1_0.disposition'{role = ntu(F1),
176+
first = ntu(F2),
177+
last = ntu(F3)};
178+
disposition(D, [F1, F2, F3, F4]) ->
179+
D#'v1_0.disposition'{role = ntu(F1),
180+
first = ntu(F2),
181+
last = ntu(F3),
182+
settled = ntu(F4)};
183+
disposition(D, [F1, F2, F3, F4, F5]) ->
184+
D#'v1_0.disposition'{role = ntu(F1),
185+
first = ntu(F2),
186+
last = ntu(F3),
187+
settled = ntu(F4),
188+
state = amqp10_framing:decode(F5)};
189+
disposition(D, [F1, F2, F3, F4, F5, F6]) ->
190+
D#'v1_0.disposition'{role = ntu(F1),
191+
first = ntu(F2),
192+
last = ntu(F3),
193+
settled = ntu(F4),
194+
state = amqp10_framing:decode(F5),
195+
batchable = ntu(F6)}.
196+
197+
-spec header(#'v1_0.header'{}, list()) ->
198+
#'v1_0.header'{}.
199+
header(H, []) ->
200+
H;
201+
header(H, [F1]) ->
202+
H#'v1_0.header'{durable = ntu(F1)};
203+
header(H, [F1, F2]) ->
204+
H#'v1_0.header'{durable = ntu(F1),
205+
priority = ntu(F2)};
206+
header(H, [F1, F2, F3]) ->
207+
H#'v1_0.header'{durable = ntu(F1),
208+
priority = ntu(F2),
209+
ttl = ntu(F3)};
210+
header(H, [F1, F2, F3, F4]) ->
211+
H#'v1_0.header'{durable = ntu(F1),
212+
priority = ntu(F2),
213+
ttl = ntu(F3),
214+
first_acquirer = ntu(F4)};
215+
header(H, [F1, F2, F3, F4, F5]) ->
216+
H#'v1_0.header'{durable = ntu(F1),
217+
priority = ntu(F2),
218+
ttl = ntu(F3),
219+
first_acquirer = ntu(F4),
220+
delivery_count = ntu(F5)}.
221+
222+
-spec properties(#'v1_0.properties'{}, list()) ->
223+
#'v1_0.properties'{}.
224+
properties(P, []) ->
225+
P;
226+
properties(P, [F1]) ->
227+
P#'v1_0.properties'{message_id = ntu(F1)};
228+
properties(P, [F1, F2]) ->
229+
P#'v1_0.properties'{message_id = ntu(F1),
230+
user_id = ntu(F2)};
231+
properties(P, [F1, F2, F3]) ->
232+
P#'v1_0.properties'{message_id = ntu(F1),
233+
user_id = ntu(F2),
234+
to = ntu(F3)};
235+
properties(P, [F1, F2, F3, F4]) ->
236+
P#'v1_0.properties'{message_id = ntu(F1),
237+
user_id = ntu(F2),
238+
to = ntu(F3),
239+
subject = ntu(F4)};
240+
properties(P, [F1, F2, F3, F4, F5]) ->
241+
P#'v1_0.properties'{message_id = ntu(F1),
242+
user_id = ntu(F2),
243+
to = ntu(F3),
244+
subject = ntu(F4),
245+
reply_to = ntu(F5)};
246+
properties(P, [F1, F2, F3, F4, F5, F6]) ->
247+
P#'v1_0.properties'{message_id = ntu(F1),
248+
user_id = ntu(F2),
249+
to = ntu(F3),
250+
subject = ntu(F4),
251+
reply_to = ntu(F5),
252+
correlation_id = ntu(F6)};
253+
properties(P, [F1, F2, F3, F4, F5, F6, F7]) ->
254+
P#'v1_0.properties'{message_id = ntu(F1),
255+
user_id = ntu(F2),
256+
to = ntu(F3),
257+
subject = ntu(F4),
258+
reply_to = ntu(F5),
259+
correlation_id = ntu(F6),
260+
content_type = ntu(F7)};
261+
properties(P, [F1, F2, F3, F4, F5, F6, F7, F8]) ->
262+
P#'v1_0.properties'{message_id = ntu(F1),
263+
user_id = ntu(F2),
264+
to = ntu(F3),
265+
subject = ntu(F4),
266+
reply_to = ntu(F5),
267+
correlation_id = ntu(F6),
268+
content_type = ntu(F7),
269+
content_encoding = ntu(F8)};
270+
properties(P, [F1, F2, F3, F4, F5, F6, F7, F8, F9]) ->
271+
P#'v1_0.properties'{message_id = ntu(F1),
272+
user_id = ntu(F2),
273+
to = ntu(F3),
274+
subject = ntu(F4),
275+
reply_to = ntu(F5),
276+
correlation_id = ntu(F6),
277+
content_type = ntu(F7),
278+
content_encoding = ntu(F8),
279+
absolute_expiry_time = ntu(F9)};
280+
properties(P, [F1, F2, F3, F4, F5, F6, F7, F8, F9, F10]) ->
281+
P#'v1_0.properties'{message_id = ntu(F1),
282+
user_id = ntu(F2),
283+
to = ntu(F3),
284+
subject = ntu(F4),
285+
reply_to = ntu(F5),
286+
correlation_id = ntu(F6),
287+
content_type = ntu(F7),
288+
content_encoding = ntu(F8),
289+
absolute_expiry_time = ntu(F9),
290+
creation_time = ntu(F10)};
291+
properties(P, [F1, F2, F3, F4, F5, F6, F7, F8, F9, F10, F11]) ->
292+
P#'v1_0.properties'{message_id = ntu(F1),
293+
user_id = ntu(F2),
294+
to = ntu(F3),
295+
subject = ntu(F4),
296+
reply_to = ntu(F5),
297+
correlation_id = ntu(F6),
298+
content_type = ntu(F7),
299+
content_encoding = ntu(F8),
300+
absolute_expiry_time = ntu(F9),
301+
creation_time = ntu(F10),
302+
group_id = ntu(F11)};
303+
properties(P, [F1, F2, F3, F4, F5, F6, F7, F8, F9, F10, F11, F12]) ->
304+
P#'v1_0.properties'{message_id = ntu(F1),
305+
user_id = ntu(F2),
306+
to = ntu(F3),
307+
subject = ntu(F4),
308+
reply_to = ntu(F5),
309+
correlation_id = ntu(F6),
310+
content_type = ntu(F7),
311+
content_encoding = ntu(F8),
312+
absolute_expiry_time = ntu(F9),
313+
creation_time = ntu(F10),
314+
group_id = ntu(F11),
315+
group_sequence = ntu(F12)};
316+
properties(P, [F1, F2, F3, F4, F5, F6, F7, F8, F9, F10, F11, F12, F13]) ->
317+
P#'v1_0.properties'{message_id = ntu(F1),
318+
user_id = ntu(F2),
319+
to = ntu(F3),
320+
subject = ntu(F4),
321+
reply_to = ntu(F5),
322+
correlation_id = ntu(F6),
323+
content_type = ntu(F7),
324+
content_encoding = ntu(F8),
325+
absolute_expiry_time = ntu(F9),
326+
creation_time = ntu(F10),
327+
group_id = ntu(F11),
328+
group_sequence = ntu(F12),
329+
reply_to_group_id = ntu(F13)}.
330+
331+
%% null to undefined
332+
-compile({inline, [ntu/1]}).
333+
ntu(null) ->
334+
undefined;
335+
ntu(Other) ->
336+
Other.

0 commit comments

Comments
 (0)