ra v3 wip

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -42,6 +42,7 @@
          %% ra_machine callbacks
          init/1,
          apply/3,
+         live_indexes/1,
          state_enter/2,
          tick/2,
          overview/1,
@@ -680,6 +681,15 @@ apply(_Meta, Cmd, State) ->
     ?LOG_DEBUG("rabbit_fifo: unhandled command ~W", [Cmd, 10]),
     {State, ok, []}.
 
+-spec live_indexes(state()) ->
+    [ra:index()].
+live_indexes(#?STATE{returns = Returns,
+                     messages = Messages,
+                     dlx = Dlx}) ->
+    DlxIndexes = rabbit_fifo_dlx:live_indexes(Dlx),
+    RtnIndexes = [I || ?MSG(I, _) <- lqueue:to_list(Returns)],
+    DlxIndexes ++ RtnIndexes ++ rabbit_fifo_q:indexes(Messages).
+
 convert_v3_to_v4(#{} = _Meta, StateV3) ->
     %% TODO: consider emitting release cursors as checkpoints
     Messages0 = rabbit_fifo_v3:get_field(messages, StateV3),
@@ -932,7 +942,7 @@ get_checked_out(CKey, From, To, #?STATE{consumers = Consumers}) ->
     end.
 
 -spec version() -> pos_integer().
-version() -> 7.
+version() -> 8.
 
 which_module(0) -> rabbit_fifo_v0;
 which_module(1) -> rabbit_fifo_v1;
@@ -941,7 +951,8 @@ which_module(3) -> rabbit_fifo_v3;
 which_module(4) -> ?MODULE;
 which_module(5) -> ?MODULE;
 which_module(6) -> ?MODULE;
-which_module(7) -> ?MODULE.
+which_module(7) -> ?MODULE;
+which_module(8) -> ?MODULE.
 
 -define(AUX, aux_v3).
 
@@ -951,6 +962,12 @@ which_module(7) -> ?MODULE.
                      messages_total :: non_neg_integer(),
                      indexes = ?CHECK_MIN_INDEXES :: non_neg_integer(),
                      bytes_in = 0 :: non_neg_integer()}).
+-record(snapshot, {index :: ra:index(),
+                   timestamp :: milliseconds(),
+                   % smallest_index :: undefined | ra:index(),
+                   messages_total :: non_neg_integer(),
+                   % indexes = ?CHECK_MIN_INDEXES :: non_neg_integer(),
+                   bytes_out = 0 :: non_neg_integer()}).
 -record(aux_gc, {last_raft_idx = 0 :: ra:index()}).
 -record(aux, {name :: atom(),
               capacity :: term(),
@@ -961,7 +978,7 @@ which_module(7) -> ?MODULE.
                gc = #aux_gc{} :: #aux_gc{},
                tick_pid :: undefined | pid(),
                cache = #{} :: map(),
-               last_checkpoint :: #checkpoint{},
+               last_checkpoint :: #checkpoint{} | #snapshot{},
                bytes_in = 0 :: non_neg_integer(),
                bytes_out = 0 :: non_neg_integer()}).
 
@@ -973,10 +990,10 @@ init_aux(Name) when is_atom(Name) ->
     Now = erlang:monotonic_time(microsecond),
     #?AUX{name = Name,
           capacity = {inactive, Now, 1, 1.0},
-          last_checkpoint = #checkpoint{index = 0,
-                                        timestamp = erlang:system_time(millisecond),
-                                        messages_total = 0,
-                                        bytes_in = 0}}.
+          last_checkpoint = #snapshot{index = 0,
+                                      timestamp = erlang:system_time(millisecond),
+                                      messages_total = 0,
+                                      bytes_out = 0}}.
 
 handle_aux(RaftState, Tag, Cmd, #aux{name = Name,
                                      capacity = Cap,
@@ -994,13 +1011,16 @@ handle_aux(RaftState, Tag, Cmd, AuxV2, RaAux)
 handle_aux(leader, cast, eval,
            #?AUX{last_decorators_state = LastDec,
                  bytes_in = BytesIn,
+                 bytes_out = BytesOut,
                  last_checkpoint = Check0} = Aux0,
            RaAux) ->
     #?STATE{cfg = #cfg{resource = QName}} = MacState =
         ra_aux:machine_state(RaAux),
 
     Ts = erlang:system_time(millisecond),
-    {Check, Effects0} = do_checkpoints(Ts, Check0, RaAux, BytesIn, false),
+    EffMacVer = ra_aux:effective_machine_version(RaAux),
+    {Check, Effects0} = do_checkpoints(EffMacVer, Ts, Check0, RaAux,
+                                       BytesIn, BytesOut, false),
 
     %% this is called after each batch of commands have been applied
     %% set timer for message expire
@@ -1017,15 +1037,23 @@ handle_aux(leader, cast, eval,
     end;
 handle_aux(_RaftState, cast, eval,
            #?AUX{last_checkpoint = Check0,
-                 bytes_in = BytesIn} = Aux0,
+                 bytes_in = BytesIn,
+                 bytes_out = BytesOut
+                } = Aux0,
            RaAux) ->
     Ts = erlang:system_time(millisecond),
-    {Check, Effects} = do_checkpoints(Ts, Check0, RaAux, BytesIn, false),
+    EffMacVer = ra_aux:effective_machine_version(RaAux),
+    {Check, Effects} = do_checkpoints(EffMacVer, Ts, Check0, RaAux,
+                                      BytesIn, BytesOut, false),
     {no_reply, Aux0#?AUX{last_checkpoint = Check}, RaAux, Effects};
 handle_aux(_RaftState, cast, {bytes_in, {MetaSize, BodySize}},
            #?AUX{bytes_in = Bytes} = Aux0,
            RaAux) ->
     {no_reply, Aux0#?AUX{bytes_in = Bytes + MetaSize + BodySize}, RaAux, []};
+handle_aux(_RaftState, cast, {bytes_out, BodySize},
+           #?AUX{bytes_out = Bytes} = Aux0,
+           RaAux) ->
+    {no_reply, Aux0#?AUX{bytes_out = Bytes + BodySize}, RaAux, []};
 handle_aux(_RaftState, cast, {#return{msg_ids = MsgIds,
                                       consumer_key = Key} = Ret, Corr, Pid},
            Aux0, RaAux0) ->
@@ -1156,12 +1184,15 @@ handle_aux(_, _, garbage_collection, Aux, RaAux) ->
     {no_reply, force_eval_gc(RaAux, Aux), RaAux};
 handle_aux(_RaState, _, force_checkpoint,
            #?AUX{last_checkpoint  = Check0,
-                 bytes_in = BytesIn} = Aux, RaAux) ->
+                 bytes_in = BytesIn,
+                 bytes_out = BytesOut} = Aux, RaAux) ->
     Ts = erlang:system_time(millisecond),
     #?STATE{cfg = #cfg{resource = QR}} = ra_aux:machine_state(RaAux),
     ?LOG_DEBUG("~ts: rabbit_fifo: forcing checkpoint at ~b",
                      [rabbit_misc:rs(QR), ra_aux:last_applied(RaAux)]),
-    {Check, Effects} = do_checkpoints(Ts, Check0, RaAux, BytesIn, true),
+    EffMacVer = ra_aux:effective_machine_version(RaAux),
+    {Check, Effects} = do_checkpoints(EffMacVer, Ts, Check0, RaAux,
+                                      BytesIn, BytesOut, true),
     {no_reply, Aux#?AUX{last_checkpoint = Check}, RaAux, Effects};
 handle_aux(RaState, _, {dlx, _} = Cmd, Aux0, RaAux) ->
     #?STATE{dlx = DlxState,
@@ -1791,25 +1822,27 @@ complete(Meta, ConsumerKey, [MsgId],
          #consumer{checked_out = Checked0} = Con0,
          #?STATE{ra_indexes = Indexes0,
                  msg_bytes_checkout = BytesCheckout,
-                 messages_total = Tot} = State0) ->
+                 messages_total = Tot} = State0,
+        Effects) ->
     case maps:take(MsgId, Checked0) of
         {?MSG(Idx, Hdr), Checked} ->
             SettledSize = get_header(size, Hdr),
             Indexes = rabbit_fifo_index:delete(Idx, Indexes0),
             Con = Con0#consumer{checked_out = Checked,
                                 credit = increase_credit(Con0, 1)},
             State1 = update_or_remove_con(Meta, ConsumerKey, Con, State0),
-            State1#?STATE{ra_indexes = Indexes,
+            {State1#?STATE{ra_indexes = Indexes,
                           msg_bytes_checkout = BytesCheckout - SettledSize,
-                          messages_total = Tot - 1};
+                          messages_total = Tot - 1},
+             [{aux, {bytes_out, SettledSize}}, Effects]};
         error ->
-            State0
+            {State0, Effects}
     end;
 complete(Meta, ConsumerKey, MsgIds,
          #consumer{checked_out = Checked0} = Con0,
          #?STATE{ra_indexes = Indexes0,
                  msg_bytes_checkout = BytesCheckout,
-                 messages_total = Tot} = State0) ->
+                 messages_total = Tot} = State0, Effects) ->
     {SettledSize, Checked, Indexes}
         = lists:foldl(
             fun (MsgId, {S0, Ch0, Idxs}) ->
@@ -1825,9 +1858,10 @@ complete(Meta, ConsumerKey, MsgIds,
     Con = Con0#consumer{checked_out = Checked,
                         credit = increase_credit(Con0, Len)},
     State1 = update_or_remove_con(Meta, ConsumerKey, Con, State0),
-    State1#?STATE{ra_indexes = Indexes,
+    {State1#?STATE{ra_indexes = Indexes,
                   msg_bytes_checkout = BytesCheckout - SettledSize,
-                  messages_total = Tot - Len}.
+                  messages_total = Tot - Len},
+     [{aux, {bytes_out, SettledSize}}, Effects]}.
 
 increase_credit(#consumer{cfg = #consumer_cfg{lifetime = once},
                           credit = Credit}, _) ->
@@ -1854,11 +1888,12 @@ increase_credit(#consumer{credit = Current}, Credit) ->
 complete_and_checkout(#{} = Meta, MsgIds, ConsumerKey,
                       #consumer{} = Con0,
                       Effects0, State0) ->
-    State1 = complete(Meta, ConsumerKey, MsgIds, Con0, State0),
+    {State1, Effects1} = complete(Meta, ConsumerKey, MsgIds,
+                                  Con0, State0, Effects0),
     %% a completion could have removed the active/quiescing consumer
-    Effects1 = add_active_effect(Con0, State1, Effects0),
-    {State2, Effects2} = activate_next_consumer(State1, Effects1),
-    checkout(Meta, State0, State2, Effects2).
+    Effects2 = add_active_effect(Con0, State1, Effects1),
+    {State2, Effects} = activate_next_consumer(State1, Effects2),
+    checkout(Meta, State0, State2, Effects).
 
 add_active_effect(#consumer{status = quiescing} = Consumer,
                   #?STATE{cfg = #cfg{consumer_strategy = single_active},
@@ -1950,8 +1985,9 @@ return_one(Meta, MsgId, ?MSG(_, _) = Msg0, DelivFailed, Anns,
             {DlxState, DlxEffects} =
                 rabbit_fifo_dlx:discard([Msg], delivery_limit, DLH, DlxState0),
             State1 = State0#?STATE{dlx = DlxState},
-            State = complete(Meta, ConsumerKey, [MsgId], Con0, State1),
-            {State, DlxEffects ++ Effects0};
+            {State, Effects} = complete(Meta, ConsumerKey, [MsgId],
+                                         Con0, State1, Effects0),
+            {State, DlxEffects ++ Effects};
         _ ->
             Checked = maps:remove(MsgId, Checked0),
             Con = Con0#consumer{checked_out = Checked,
@@ -2817,7 +2853,10 @@ convert(Meta, 5, To, State) ->
     convert(Meta, 6, To, State);
 convert(Meta, 6, To, State) ->
     %% no conversion needed, this version only includes a logic change
-    convert(Meta, 7, To, State).
+    convert(Meta, 7, To, State);
+convert(Meta, 7, To, State) ->
+    %% no conversion needed, this version only includes a logic change
+    convert(Meta, 8, To, State).
 
 smallest_raft_index(#?STATE{messages = Messages,
                             ra_indexes = Indexes,
@@ -2987,12 +3026,43 @@ priority_tag(Msg) ->
     end.
 
 
-do_checkpoints(Ts, #checkpoint{index = ChIdx,
-                               timestamp = ChTime,
-                               smallest_index = LastSmallest,
-                               bytes_in = LastBytesIn,
-                               indexes = MinIndexes} = Check0,
-               RaAux, BytesIn, Force) ->
+do_checkpoints(MacVer, Ts, #checkpoint{index = _ChIdx,
+                                    timestamp = _SnapTime},
+               RaAux, BytesIn, BytesOut, Force) when MacVer >= 8 ->
+    do_checkpoints(MacVer, Ts, #snapshot{}, RaAux, BytesIn, BytesOut, Force);
+do_checkpoints(MacVer, Ts, #snapshot{index = _ChIdx,
+                                    timestamp = SnapTime,
+                                    bytes_out = LastBytesOut} = Snap0,
+               RaAux, _BytesIn, BytesOut, _Force) when MacVer >= 8 ->
+    LastAppliedIdx = ra_aux:last_applied(RaAux),
+    #?STATE{} = MacState = ra_aux:machine_state(RaAux),
+    TimeSince = Ts - SnapTime,
+    MsgsTot = messages_total(MacState),
+    ra_aux:overview(RaAux),
+    % MaxBytesFactor = max(1, MsgsTot / CheckMaxIndexes),
+    EnoughDataRemoved = BytesOut - LastBytesOut > ?SNAP_OUT_BYTES,
+    {CheckMinInterval, _CheckMinIndexes, _CheckMaxIndexes} =
+        persistent_term:get(quorum_queue_checkpoint_config,
+                            {?CHECK_MIN_INTERVAL_MS, ?CHECK_MIN_INDEXES,
+                             ?CHECK_MAX_INDEXES}),
+    EnoughTimeHasPassed = TimeSince > CheckMinInterval,
+    case (EnoughTimeHasPassed andalso
+          EnoughDataRemoved) of
+        true ->
+            {#snapshot{index = LastAppliedIdx,
+                       timestamp = Ts,
+                       messages_total = MsgsTot,
+                       bytes_out = BytesOut},
+             [{release_cursor, LastAppliedIdx, MacState}]};
+        false ->
+            {Snap0, []}
+    end;
+do_checkpoints(MacVer,Ts, #checkpoint{index = ChIdx,
+                                      timestamp = ChTime,
+                                      smallest_index = LastSmallest,
+                                      bytes_in = LastBytesIn,
+                                      indexes = MinIndexes} = Check0,
+               RaAux, BytesIn, _BytesOut, Force) when MacVer < 8 ->
     LastAppliedIdx = ra_aux:last_applied(RaAux),
     IndexesSince = LastAppliedIdx - ChIdx,
     #?STATE{} = MacState = ra_aux:machine_state(RaAux),
@@ -3022,7 +3092,7 @@ do_checkpoints(Ts, #checkpoint{index = ChIdx,
            %% condition 1: enough indexes have been committed since the last
            %% checkpoint
            (IndexesSince > MinIndexes) orelse
-           %% condition 2: the queue is empty and _some_ commands
+          %% condition 2: the queue is empty and _some_ commands
            %% have been applied since the last checkpoint
            (MsgsTot == 0 andalso IndexesSince > 32)
           )

deps/rabbit/src/rabbit_fifo.hrl

@@ -105,6 +105,7 @@
 %% once these many bytes have been written since the last checkpoint
 %% we request a checkpoint irrespectively
 -define(CHECK_MAX_BYTES, 128_000_000).
+-define(SNAP_OUT_BYTES, 64_000_000).
 
 -define(USE_AVG_HALF_LIFE, 10000.0).
 %% an average QQ without any message uses about 100KB so setting this limit

deps/rabbit/src/rabbit_fifo_client.erl

@@ -1031,13 +1031,14 @@ send_command(Server, Correlation, Command, Priority,
              #state{pending = Pending,
                     next_seq = Seq,
                     cfg = #cfg{soft_limit = SftLmt}} = State) ->
-    ok = case rabbit_fifo:is_return(Command) of
-             true ->
-                 %% returns are sent to the aux machine for pre-evaluation
-                 ra:cast_aux_command(Server, {Command, Seq, self()});
-             _ ->
-                 ra:pipeline_command(Server, Command, Seq, Priority)
-         end,
+    % ok = case rabbit_fifo:is_return(Command) of
+    %          true ->
+    %              %% returns are sent to the aux machine for pre-evaluation
+    %              ra:cast_aux_command(Server, {Command, Seq, self()});
+    %          _ ->
+    %              ra:pipeline_command(Server, Command, Seq, Priority)
+    %      end,
+    ok = ra:pipeline_command(Server, Command, Seq, Priority),
     State#state{pending = Pending#{Seq => {Correlation, Command}},
                 next_seq = Seq + 1,
                 slow = map_size(Pending) >= SftLmt}.

deps/rabbit/src/rabbit_fifo_dlx.erl

@@ -26,7 +26,8 @@
          dehydrate/1,
          stat/1,
          update_config/4,
-         smallest_raft_index/1
+         smallest_raft_index/1,
+         live_indexes/1
         ]).
 
 -record(checkout, {consumer :: pid(),
@@ -365,5 +366,9 @@ dehydrate(State) ->
 smallest_raft_index(#?MODULE{ra_indexes = Indexes}) ->
     rabbit_fifo_index:smallest(Indexes).
 
+-spec live_indexes(state()) -> [ra:index()].
+live_indexes(#?MODULE{ra_indexes = Indexes}) ->
+    rabbit_fifo_index:indexes(Indexes).
+
 annotate_msg(H, Msg) ->
     rabbit_fifo:annotate_msg(H, Msg).

deps/rabbit/src/rabbit_fifo_index.erl

@@ -13,6 +13,7 @@
          delete/2,
          size/1,
          smallest/1,
+         indexes/1,
          map/2,
          to_list/1
         ]).
@@ -90,6 +91,10 @@ size(#?MODULE{data = Data}) ->
 smallest(#?MODULE{smallest = Smallest}) ->
     Smallest.
 
+-spec indexes(state()) -> [ra:index()].
+indexes(#?MODULE{data = Data}) ->
+    maps:keys(Data).
+
 -spec map(fun(), state()) -> state().
 map(F, #?MODULE{data = Data} = State) ->
     State#?MODULE{data = maps:map(F, Data)}.

deps/rabbit/src/rabbit_fifo_q.erl

@@ -8,6 +8,7 @@
          get/1,
          len/1,
          from_lqueue/1,
+         indexes/1,
          get_lowest_index/1,
          overview/1
         ]).
@@ -81,6 +82,13 @@ from_lqueue(LQ) ->
                         in(no, Item, Acc)
                 end, new(), LQ).
 
+indexes(#?MODULE{hi = {Hi1, Hi2},
+                 no = {No1, No2}}) ->
+    A = lists:map(fun (?MSG(I, _)) -> I end, Hi1),
+    B = lists:foldl(fun (?MSG(I, _), Acc) -> [I | Acc] end, A, Hi2),
+    C = lists:foldl(fun (?MSG(I, _), Acc) -> [I | Acc] end, B, No1),
+    lists:foldl(fun (?MSG(I, _), Acc) -> [I | Acc] end, C, No2).
+
 -spec get_lowest_index(state()) -> undefined | ra:index().
 get_lowest_index(#?MODULE{len = 0}) ->
     undefined;

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -2008,6 +2008,7 @@ make_ra_conf(Q, ServerId, TickTimeout,
     FName = rabbit_misc:rs(QName),
     Formatter = {?MODULE, format_ra_event, [QName]},
     LogCfg = #{uid => UId,
+               min_snapshot_interval => 0,
                snapshot_interval => SnapshotInterval,
                min_checkpoint_interval => CheckpointInterval,
                max_checkpoints => 3},