perf(gateway): size replay buffer entries once per dispatch (#2210)

fluxer_gateway/src/limited_deque.erl

@@ -7,6 +7,7 @@
 -export([
     new/2,
     push/2,
+    push/3,
     push_front/2,
     pop/1,
     pop_front/1,
@@ -17,11 +18,14 @@
     is_empty/1,
     filter/2,
     drop_while_front/2,
-    recompute_bytes/2
+    recompute_bytes/2,
+    entry_bytes/1
 ]).
 
 -export_type([deque/0]).
 
+-define(WORD_SIZE_KEY, {?MODULE, word_size}).
+
 -opaque deque() :: #{
     front := [term()],
     rear := [term()],
@@ -43,8 +47,11 @@ new(MaxCount, MaxBytes) ->
     }.
 
 -spec push(term(), deque()) -> deque().
-push(Item, #{rear := Rear, count := Count, bytes := Bytes} = D) ->
-    ItemBytes = entry_bytes(Item),
+push(Item, D) ->
+    push(Item, entry_bytes(Item), D).
+
+-spec push(term(), non_neg_integer(), deque()) -> deque().
+push(Item, ItemBytes, #{rear := Rear, count := Count, bytes := Bytes} = D) ->
     D1 = D#{rear := [Item | Rear], count := Count + 1, bytes := Bytes + ItemBytes},
     trim_front(D1).
 
@@ -176,7 +183,21 @@ trim_rear(D) ->
 
 -spec entry_bytes(term()) -> non_neg_integer().
 entry_bytes(Term) ->
-    erts_debug:flat_size(Term) * erlang:system_info(wordsize).
+    erts_debug:flat_size(Term) * word_size().
+
+-spec word_size() -> 4 | 8.
+word_size() ->
+    case persistent_term:get(?WORD_SIZE_KEY, undefined) of
+        4 -> 4;
+        8 -> 8;
+        _Other -> cache_word_size()
+    end.
+
+-spec cache_word_size() -> 4 | 8.
+cache_word_size() ->
+    WordSize = erlang:system_info(wordsize),
+    persistent_term:put(?WORD_SIZE_KEY, WordSize),
+    WordSize.
 
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
@@ -239,4 +260,41 @@ size_is_o1_test() ->
     D1 = lists:foldl(fun push/2, D0, lists:seq(1, 1000)),
     ?assertEqual(1000, size(D1)).
 
+push_with_precomputed_bytes_matches_push_test() ->
+    Item = #{event => presence_update, data => #{<<"status">> => <<"online">>}, seq => 7},
+    D0 = new(10, 1048576),
+    ?assertEqual(push(Item, D0), push(Item, entry_bytes(Item), D0)),
+    ?assertEqual(entry_bytes(Item), bytes(push(Item, D0))).
+
+push_with_precomputed_bytes_skips_recomputation_test() ->
+    Item = #{event => presence_update, data => #{<<"status">> => <<"online">>}, seq => 7},
+    D = push(Item, 1234, new(10, 1048576)),
+    ?assertEqual(1, size(D)),
+    ?assertEqual(1234, bytes(D)).
+
+push_with_precomputed_bytes_matches_push_at_byte_bound_test() ->
+    Item = lists:seq(1, 256),
+    ItemBytes = entry_bytes(Item),
+    D0 = new(10, ItemBytes * 2),
+    Items = [Item, Item, Item],
+    D1 = lists:foldl(fun push/2, D0, Items),
+    D2 = lists:foldl(fun(I, D) -> push(I, entry_bytes(I), D) end, D0, Items),
+    ?assertEqual(2, size(D1)),
+    ?assertEqual(ItemBytes * 2, bytes(D1)),
+    ?assertEqual(D1, D2).
+
+push_with_precomputed_bytes_matches_push_at_count_bound_test() ->
+    D0 = new(3, 0),
+    Items = [a, b, c, d],
+    D1 = lists:foldl(fun push/2, D0, Items),
+    D2 = lists:foldl(fun(I, D) -> push(I, entry_bytes(I), D) end, D0, Items),
+    ?assertEqual([b, c, d], to_list(D1)),
+    ?assertEqual(D1, D2).
+
+entry_bytes_uses_word_size_test() ->
+    ?assertEqual(
+        erts_debug:flat_size({a, b, c}) * erlang:system_info(wordsize),
+        entry_bytes({a, b, c})
+    ).
+
 -endif.

fluxer_gateway/src/session/session_dispatch.erl

@@ -139,16 +139,19 @@ do_handle_dispatch(Event, Data, State) ->
     {noreply, session_state()}.
 dispatch_replayable_event(Event, Data, NewSeq, State) ->
     Request = #{event => Event, data => Data, seq => NewSeq},
-    case is_oversized_event(Request) of
+    RequestBytes = buffer_entry_bytes(Request),
+    case is_oversized_event(RequestBytes) of
         true ->
             dispatch_without_replay(Event, Data, NewSeq, State);
         false ->
-            dispatch_with_replay(Event, Data, NewSeq, Request, State)
+            dispatch_with_replay(Event, Data, NewSeq, Request, RequestBytes, State)
     end.
 
--spec dispatch_with_replay(event(), map(), non_neg_integer(), map(), session_state()) ->
+-spec dispatch_with_replay(
+    event(), map(), non_neg_integer(), map(), non_neg_integer(), session_state()
+) ->
     {noreply, session_state()}.
-dispatch_with_replay(Event, Data, NewSeq, Request, State) ->
+dispatch_with_replay(Event, Data, NewSeq, Request, RequestBytes, State) ->
     Buffer = maps:get(buffer, State),
     NewBuffer =
         case is_list(Buffer) of
@@ -156,9 +159,9 @@ dispatch_with_replay(Event, Data, NewSeq, Request, State) ->
                 D = limited_deque:from_list(
                     Buffer, ?MAX_EVENT_BUFFER_SIZE, ?MAX_TOTAL_BUFFER_BYTES
                 ),
-                limited_deque:push(Request, D);
+                limited_deque:push(Request, RequestBytes, D);
             false ->
-                limited_deque:push(Request, Buffer)
+                limited_deque:push(Request, RequestBytes, Buffer)
         end,
     send_to_socket(maps:get(socket_pid, State, undefined), Event, Data, NewSeq),
     StateAfterMain = apply_state_updates(Event, Data, State, #{
@@ -225,9 +228,9 @@ needs_state_update(relationship_update) -> true;
 needs_state_update(relationship_remove) -> true;
 needs_state_update(_) -> false.
 
--spec is_oversized_event(map()) -> boolean().
-is_oversized_event(Request) ->
-    buffer_entry_bytes(Request) > ?MAX_SINGLE_EVENT_BUFFER_BYTES.
+-spec is_oversized_event(non_neg_integer()) -> boolean().
+is_oversized_event(RequestBytes) ->
+    RequestBytes > ?MAX_SINGLE_EVENT_BUFFER_BYTES.
 
 -spec should_skip_replay_buffer(event()) -> boolean().
 should_skip_replay_buffer(Event) ->
@@ -235,7 +238,7 @@ should_skip_replay_buffer(Event) ->
 
 -spec buffer_entry_bytes(term()) -> non_neg_integer().
 buffer_entry_bytes(Request) ->
-    erts_debug:flat_size(Request) * erlang:system_info(wordsize).
+    limited_deque:entry_bytes(Request).
 
 -spec send_to_socket(pid() | undefined, event(), term(), non_neg_integer()) -> ok.
 send_to_socket(undefined, _Event, _Data, _Seq) ->
@@ -314,12 +317,12 @@ base_state(Opts) ->
 
 is_oversized_event_small_event_test() ->
     Req = #{event => message_create, data => #{<<"content">> => <<"hello">>}, seq => 1},
-    ?assertEqual(false, is_oversized_event(Req)).
+    ?assertEqual(false, is_oversized_event(buffer_entry_bytes(Req))).
 
 is_oversized_event_large_event_test() ->
     LargeData = make_large_data(),
     Req = #{event => guild_create, data => LargeData, seq => 1},
-    ?assertEqual(true, is_oversized_event(Req)).
+    ?assertEqual(true, is_oversized_event(buffer_entry_bytes(Req))).
 
 ignored_message_create_dispatches_when_mentioned_test() ->
     State = base_state(#{