CQ shared store: Delete from index on remove or roll over

It is expensive to delete files because we must clean up
the index and to get the messages in the file we have to
scan it.

Instead of cleaning up the index on file delete this
commit deletes from the index as soon as possible.
There are two scenarios: messages that are removed
from the current write file, and messages that are
removed from other files. In the latter case, we
can just delete the index entry on remove. For messages
in the current write file, we want to keep the entry
in case fanout is used, because we don't want to write
the fanout message multiple times if we can avoid it.
So we keep track of removes in the current write file
and do a cleanup of these entries on file roll over.

Compared to the previous implementation we will no
longer increase the ref_count of messages that are
not in the current write file, meaning we may do more
writes in fanout scenarios. But at the same time the
file delete operation is much cheaper.

Author: lhoguin

deps/rabbit/src/rabbit_msg_store.erl

@@ -77,8 +77,10 @@
           current_file,
           %% current file handle since the last fsync?
           current_file_handle,
-          %% file handle cache
+          %% current write file offset
           current_file_offset,
+          %% messages that were potentially removed from the current write file
+          current_file_removes = [],
           %% TRef for our interval timer
           sync_timer_ref,
           %% files that had removes
@@ -1150,7 +1152,8 @@ write_message(MsgId, Msg, CRef,
               end, CRef, State1)
     end.
 
-remove_message(MsgId, CRef, State = #msstate{ index_ets = IndexEts }) ->
+remove_message(MsgId, CRef, State = #msstate{
+        index_ets = IndexEts, current_file = CurrentFile }) ->
     case should_mask_action(CRef, MsgId, State) of
         {true, _Location} ->
             State;
@@ -1162,22 +1165,32 @@ remove_message(MsgId, CRef, State = #msstate{ index_ets = IndexEts }) ->
             %% ets:lookup(FileSummaryEts, File),
             State;
         {_Mask, #msg_location { ref_count = RefCount, file = File,
-                                total_size = TotalSize }}
+                                total_size = TotalSize } = Entry}
           when RefCount > 0 ->
             %% only update field, otherwise bad interaction with
             %% concurrent GC
-            Dec = fun () -> index_update_ref_counter(IndexEts, MsgId, -1) end,
             case RefCount of
-                %% don't remove from cur_file_cache_ets here because
+                %% Don't remove from cur_file_cache_ets here because
                 %% there may be further writes in the mailbox for the
-                %% same msg.
-                1 -> ok = Dec(),
-                     delete_file_if_empty(
-                       File, gc_candidate(File,
-                         adjust_valid_total_size(
-                               File, -TotalSize, State)));
-                _ -> ok = Dec(),
-                     gc_candidate(File, State)
+                %% same msg. We will remove 0 ref_counts when rolling
+                %% over to the next write file.
+                1 when File =:= CurrentFile ->
+                    index_update_ref_counter(IndexEts, MsgId, -1),
+                    State1 = State#msstate{current_file_removes =
+                        [Entry#msg_location{ref_count=0}|Removes]},
+                    delete_file_if_empty(
+                      File, gc_candidate(File,
+                        adjust_valid_total_size(
+                              File, -TotalSize, State1)));
+                1 ->
+                    index_delete(IndexEts, MsgId),
+                    delete_file_if_empty(
+                      File, gc_candidate(File,
+                        adjust_valid_total_size(
+                              File, -TotalSize, State)));
+                _ ->
+                    index_update_ref_counter(IndexEts, MsgId, -1),
+                    gc_candidate(File, State)
             end
     end.
 
@@ -1239,7 +1252,9 @@ flush_or_roll_to_new_file(
                      cur_file_cache_ets  = CurFileCacheEts,
                      file_size_limit     = FileSizeLimit })
   when Offset >= FileSizeLimit ->
-    State1 = internal_sync(State),
+    %% Cleanup the index of messages that were removed before rolling over.
+    State0 = cleanup_index_on_roll_over(State),
+    State1 = internal_sync(State0),
     ok = writer_close(CurHdl),
     NextFile = CurFile + 1,
     {ok, NextHdl} = writer_open(Dir, NextFile),
@@ -1267,6 +1282,8 @@ write_large_message(MsgId, MsgBodyBin,
                                   index_ets           = IndexEts,
                                   file_summary_ets    = FileSummaryEts,
                                   cur_file_cache_ets  = CurFileCacheEts }) ->
+    %% Cleanup the index of messages that were removed before rolling over.
+    State1 = cleanup_index_on_roll_over(State0),
     {LargeMsgFile, LargeMsgHdl} = case CurOffset of
         %% We haven't written in the file yet. Use it.
         0 ->
@@ -1286,13 +1303,13 @@ write_large_message(MsgId, MsgBodyBin,
     ok = index_insert(IndexEts,
            #msg_location { msg_id = MsgId, ref_count = 1, file = LargeMsgFile,
                            offset = 0, total_size = TotalSize }),
-    State1 = case CurFile of
+    State2 = case CurFile of
         %% We didn't open a new file. We must update the existing value.
         LargeMsgFile ->
             [_,_] = ets:update_counter(FileSummaryEts, LargeMsgFile,
                                        [{#file_summary.valid_total_size, TotalSize},
                                         {#file_summary.file_size,        TotalSize}]),
-            State0;
+            State1;
         %% We opened a new file. We can insert it all at once.
         %% We must also check whether we need to delete the previous
         %% current file, because if there is no valid data this is
@@ -1303,7 +1320,7 @@ write_large_message(MsgId, MsgBodyBin,
                                     valid_total_size = TotalSize,
                                     file_size        = TotalSize,
                                     locked           = false }),
-            delete_file_if_empty(CurFile, State0 #msstate { current_file_handle = LargeMsgHdl,
+            delete_file_if_empty(CurFile, State1 #msstate { current_file_handle = LargeMsgHdl,
                                                             current_file        = LargeMsgFile,
                                                             current_file_offset = TotalSize })
     end,
@@ -1318,11 +1335,21 @@ write_large_message(MsgId, MsgBodyBin,
     %% Delete messages from the cache that were written to disk.
     true = ets:match_delete(CurFileCacheEts, {'_', '_', 0}),
     %% Process confirms (this won't flush; we already did) and continue.
-    State = internal_sync(State1),
+    State = internal_sync(State2),
     State #msstate { current_file_handle = NextHdl,
                      current_file        = NextFile,
                      current_file_offset = 0 }.
 
+cleanup_index_on_roll_over(State = #msstate{ index_ets = IndexEts,
+        current_file = CurrentFile, current_file_removes = Removes}) ->
+    lists:foreach(fun(Entry) ->
+        %% We delete objects that have ref_count=0. If a message
+        %% got its ref_count increased, it will not be deleted.
+        %% We thus avoid extra index lookups to check for ref_count.
+        index_delete_object(IndexEts, Entry)
+    end, Removes),
+    State=#msstate{current_file_removes=[]}.
+
 contains_message(MsgId, From, State = #msstate{ index_ets = IndexEts }) ->
     MsgLocation = index_lookup_positive_ref_count(IndexEts, MsgId),
     gen_server2:reply(From, MsgLocation =/= not_found),
@@ -2134,7 +2161,6 @@ delete_file(File, State = #gc_state { file_summary_ets = FileSummaryEts,
         _ ->
             [#file_summary{ valid_total_size = 0,
                             file_size = FileSize }] = ets:lookup(FileSummaryEts, File),
-            [] = scan_and_vacuum_message_file(File, State),
             ok = file:delete(form_filename(Dir, filenum_to_name(File))),
             true = ets:delete(FileSummaryEts, File),
             rabbit_log:debug("Deleted empty file number ~tp; reclaimed ~tp bytes", [File, FileSize]),