Improve memory and disk usage when upgrading to new index

Also add progress output in the logs so we know things are
happening.

Author: Loïc Hoguin

deps/rabbit/src/rabbit_classic_queue_index_v2.erl

@@ -302,25 +302,35 @@ recover_segment(State, ContainsCheckFun, CountersRef, Fd,
                             Unacked - 1, LocBytes0)
     end.
 
-recover_index_v1_clean(State = #mqistate{ queue_name = Name }, Terms, IsMsgStoreClean,
+recover_index_v1_clean(State0 = #mqistate{ queue_name = Name }, Terms, IsMsgStoreClean,
                        ContainsCheckFun, OnSyncFun, OnSyncMsgFun) ->
+    #resource{virtual_host = VHost, name = QName} = Name,
+    logger:info("Converting clean queue ~s on vhost ~s to the new index format", [QName, VHost]),
     {_, _, V1State} = rabbit_queue_index:recover(Name, Terms, IsMsgStoreClean,
                                                  ContainsCheckFun, OnSyncFun, OnSyncMsgFun),
     %% We will ignore the counter results because on clean shutdown
     %% we do not need to calculate the values again. This lets us
     %% share code with dirty recovery.
     DummyCountersRef = counters:new(2, []),
-    recover_index_v1_common(State, Terms, V1State, DummyCountersRef).
+    State = recover_index_v1_common(State0, Terms, V1State, DummyCountersRef),
+    logger:info("Queue ~s on vhost ~s converted ~b total messages to the new index format",
+                [QName, VHost, counters:get(DummyCountersRef, ?RECOVER_COUNT)]),
+    State.
 
-recover_index_v1_dirty(State = #mqistate{ queue_name = Name }, Terms, IsMsgStoreClean,
+recover_index_v1_dirty(State0 = #mqistate{ queue_name = Name }, Terms, IsMsgStoreClean,
                        ContainsCheckFun, OnSyncFun, OnSyncMsgFun,
                        CountersRef) ->
+    #resource{virtual_host = VHost, name = QName} = Name,
+    logger:info("Converting dirty queue ~s on vhost ~s to the new index format", [QName, VHost]),
     %% We ignore the count and bytes returned here because we cannot trust
     %% rabbit_queue_index: it has a bug that may lead to more bytes being
     %% returned than it really has.
     {_, _, V1State} = rabbit_queue_index:recover(Name, Terms, IsMsgStoreClean,
                                                  ContainsCheckFun, OnSyncFun, OnSyncMsgFun),
-    recover_index_v1_common(State, Terms, V1State, CountersRef).
+    State = recover_index_v1_common(State0, Terms, V1State, CountersRef),
+    logger:info("Queue ~s on vhost ~s converted ~b total messages to the new index format",
+                [QName, VHost, counters:get(CountersRef, ?RECOVER_COUNT)]),
+    State.
 
 recover_index_v1_common(State0 = #mqistate{ queue_name = #resource{ virtual_host = VHost },
                                             dir = Dir }, Terms, V1State, CountersRef) ->
@@ -359,16 +369,22 @@ recover_index_v1_common(State0 = #mqistate{ queue_name = #resource{ virtual_host
 
 recover_index_v1_loop(State, _, _, _, HiSeqId, HiSeqId) ->
     State;
-recover_index_v1_loop(State0, MSClient, V1State0, CountersRef, LoSeqId, HiSeqId) ->
+recover_index_v1_loop(State0 = #mqistate{ queue_name = Name },
+                      MSClient, V1State0, CountersRef, LoSeqId, HiSeqId) ->
     UpSeqId = lists:min([rabbit_queue_index:next_segment_boundary(LoSeqId),
                          HiSeqId]),
     {Messages, V1State} = rabbit_queue_index:read(LoSeqId, UpSeqId, V1State0),
-    counters:add(CountersRef, ?RECOVER_COUNT, length(Messages)),
-    State = lists:foldl(fun({MsgOrId, SeqId, Props, IsPersistent, IsDelivered}, State1) ->
+    %% We do a garbage collect immediately after the old index read
+    %% and ack because they may have created a lot of garbage.
+    garbage_collect(),
+    MessagesCount = length(Messages),
+    counters:add(CountersRef, ?RECOVER_COUNT, MessagesCount),
+    State2 = lists:foldl(fun({MsgOrId, SeqId, Props, IsPersistent, IsDelivered}, State1) ->
         %% We must move embedded messages to the message store.
         MsgId = case MsgOrId of
             Msg = #basic_message{ id = MsgId0 } ->
-                rabbit_msg_store:write(MsgId0, Msg, MSClient),
+                %% We must do a synchronous write to avoid overloading the message store.
+                rabbit_msg_store:sync_write(MsgId0, Msg, MSClient),
                 MsgId0;
             MsgId0 ->
                 MsgId0
@@ -378,6 +394,15 @@ recover_index_v1_loop(State0, MSClient, V1State0, CountersRef, LoSeqId, HiSeqId)
         counters:add(CountersRef, ?RECOVER_BYTES, Props#message_properties.size),
         publish(MsgId, SeqId, Props, IsPersistent, IsDelivered, infinity, State1)
     end, State0, Messages),
+    State = flush(State2),
+    %% We have written everything to disk. We can delete the old segment file
+    %% to free up much needed space, to avoid doubling disk usage during the upgrade.
+    rabbit_queue_index:delete_segment_file_for_seq_id(LoSeqId, V1State),
+    %% Log some progress to keep the user aware of what's going on, as moving
+    %% embedded messages can take quite some time.
+    #resource{virtual_host = VHost, name = QName} = Name,
+    logger:info("Queue ~s on vhost ~s converted ~b more messages to the new index format",
+                [QName, VHost, MessagesCount]),
     recover_index_v1_loop(State, MSClient, V1State, CountersRef, UpSeqId, HiSeqId).
 
 -spec terminate(rabbit_types:vhost(), [any()], State) -> State when State::mqistate().

deps/rabbit/src/rabbit_msg_store.erl

@@ -20,6 +20,7 @@
 -export([scan_file_for_valid_messages/1]). %% salvage tool
 
 -export([transform_dir/3, force_recovery/2]). %% upgrade
+-export([sync_write/3]). %% Used when upgrading to the modern index.
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
          code_change/3, prioritise_call/4, prioritise_cast/3,
@@ -515,6 +516,17 @@ write_flow(MsgId, Msg,
 
 write(MsgId, Msg, CState) -> client_write(MsgId, Msg, noflow, CState).
 
+-spec sync_write(rabbit_types:msg_id(), msg(), client_msstate()) -> 'ok'.
+
+%% Used when upgrading to the modern index.
+sync_write(MsgId, Msg,
+             CState = #client_msstate { cur_file_cache_ets = CurFileCacheEts,
+                                        client_ref         = CRef }) ->
+    file_handle_cache_stats:update(msg_store_write),
+    ok = client_update_flying(+1, MsgId, CState),
+    ok = update_msg_cache(CurFileCacheEts, MsgId, Msg),
+    ok = server_call(CState, {write, CRef, MsgId, noflow}).
+
 -spec read(rabbit_types:msg_id(), client_msstate()) ->
                      {rabbit_types:ok(msg()) | 'not_found', client_msstate()}.
 
@@ -867,7 +879,12 @@ handle_call({read, MsgId}, From, State) ->
 
 handle_call({contains, MsgId}, From, State) ->
     State1 = contains_message(MsgId, From, State),
-    noreply(State1).
+    noreply(State1);
+
+%% Used when upgrading to the modern index.
+handle_call(Write = {write, _, _, _}, _From, State) ->
+    {noreply, State1, _} = handle_cast(Write, State),
+    reply(ok, State1).
 
 handle_cast({client_dying, CRef},
             State = #msstate { dying_clients       = DyingClients,

deps/rabbit/src/rabbit_queue_index.erl

@@ -31,6 +31,9 @@
 %% after a non-clean shutdown.
 -export([queue_index_walker_reader/2]).
 
+%% Used to upgrade to the modern index.
+-export([delete_segment_file_for_seq_id/2]).
+
 -define(CLEAN_FILENAME, "clean.dot").
 
 %%----------------------------------------------------------------------------
@@ -1543,3 +1546,22 @@ cleanup_global_recovery_terms() ->
 update_recovery_term(#resource{virtual_host = VHost} = QueueName, Term) ->
     Key = queue_name_to_dir_name(QueueName),
     rabbit_recovery_terms:store(VHost, Key, Term).
+
+
+%%----------------------------------------------------------------------------
+%% Upgrade to the modern index
+%%----------------------------------------------------------------------------
+
+%% This function is only used when upgrading to the new index.
+%% We delete the segment file without updating the state.
+%% We will drop the state later on so we don't care much
+%% about how accurate it is as long as we can read from
+%% subsequent segment files.
+delete_segment_file_for_seq_id(SeqId, #qistate { segments = Segments }) ->
+    {Seg, _} = seq_id_to_seg_and_rel_seq_id(SeqId),
+    case segment_find(Seg, Segments) of
+        {ok, #segment { path = Path }} ->
+            ok = rabbit_file:delete(Path);
+        error ->
+            ok
+    end.