path: root/src/rabbit_msg_store.erl

%%   The contents of this file are subject to the Mozilla Public License
%%   Version 1.1 (the "License"); you may not use this file except in
%%   compliance with the License. You may obtain a copy of the License at
%%   http://www.mozilla.org/MPL/
%%
%%   Software distributed under the License is distributed on an "AS IS"
%%   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
%%   License for the specific language governing rights and limitations
%%   under the License.
%%
%%   The Original Code is RabbitMQ.
%%
%%   The Initial Developers of the Original Code are LShift Ltd,
%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
%%
%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
%%   Technologies LLC, and Rabbit Technologies Ltd.
%%
%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
%%   (C) 2007-2010 Rabbit Technologies Ltd.
%%
%%   All Rights Reserved.
%%
%%   Contributor(s): ______________________________________.
%%

-module(rabbit_msg_store).

-behaviour(gen_server2).

-export([start_link/4, write/4, read/3, contains/2, remove/2, release/2,
         sync/3, client_init/2, client_terminate/2,
         client_delete_and_terminate/3, successfully_recovered_state/1]).

-export([sync/1, gc_done/4, set_maximum_since_use/2, gc/3]). %% internal

-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).

%%----------------------------------------------------------------------------

-include("rabbit_msg_store.hrl").

-define(SYNC_INTERVAL,  5).   %% milliseconds
-define(CLEAN_FILENAME, "clean.dot").
-define(FILE_SUMMARY_FILENAME, "file_summary.ets").

-define(BINARY_MODE,     [raw, binary]).
-define(READ_MODE,       [read]).
-define(READ_AHEAD_MODE, [read_ahead | ?READ_MODE]).
-define(WRITE_MODE,      [write]).

-define(FILE_EXTENSION,        ".rdq").
-define(FILE_EXTENSION_TMP,    ".rdt").

-define(HANDLE_CACHE_BUFFER_SIZE, 1048576). %% 1MB

%%----------------------------------------------------------------------------

-record(msstate,
        { dir,                    %% store directory
          index_module,           %% the module for index ops
          index_state,            %% where are messages?
          current_file,           %% current file name as number
          current_file_handle,    %% current file handle since the last fsync?
          file_handle_cache,      %% file handle cache
          on_sync,                %% pending sync requests
          sync_timer_ref,         %% TRef for our interval timer
          sum_valid_data,         %% sum of valid data in all files
          sum_file_size,          %% sum of file sizes
          pending_gc_completion,  %% things to do once GC completes
          gc_active,              %% is the GC currently working?
          gc_pid,                 %% pid of our GC
          file_handles_ets,       %% tid of the shared file handles table
          file_summary_ets,       %% tid of the file summary table
          dedup_cache_ets,        %% tid of dedup cache table
          cur_file_cache_ets,     %% tid of current file cache table
          client_refs,            %% set of references of all registered clients
          successfully_recovered, %% boolean: did we recover state?
          file_size_limit         %% how big are our files allowed to get?
         }).

-record(client_msstate,
        { file_handle_cache,
          index_state,
          index_module,
          dir,
          gc_pid,
          file_handles_ets,
          file_summary_ets,
          dedup_cache_ets,
          cur_file_cache_ets
         }).

-record(file_summary,
        {file, valid_total_size, contiguous_top, left, right, file_size,
         locked, readers}).

%%----------------------------------------------------------------------------

-ifdef(use_specs).

-type(server() :: pid() | atom()).
-type(file_num() :: non_neg_integer()).
-type(client_msstate() :: #client_msstate {
                      file_handle_cache  :: dict:dictionary(),
                      index_state        :: any(),
                      index_module       :: atom(),
                      dir                :: file:filename(),
                      gc_pid             :: pid(),
                      file_handles_ets   :: ets:tid(),
                      file_summary_ets   :: ets:tid(),
                      dedup_cache_ets    :: ets:tid(),
                      cur_file_cache_ets :: ets:tid() }).
-type(startup_fun_state() ::
        {(fun ((A) -> 'finished' | {rabbit_guid:guid(), non_neg_integer(), A})),
         A}).

-spec(start_link/4 ::
        (atom(), file:filename(), [binary()] | 'undefined',
         startup_fun_state()) -> rabbit_types:ok_pid_or_error()).
-spec(write/4 :: (server(), rabbit_guid:guid(), msg(), client_msstate()) ->
                      rabbit_types:ok(client_msstate())).
-spec(read/3 :: (server(), rabbit_guid:guid(), client_msstate()) ->
                     {rabbit_types:ok(msg()) | 'not_found', client_msstate()}).
-spec(contains/2 :: (server(), rabbit_guid:guid()) -> boolean()).
-spec(remove/2 :: (server(), [rabbit_guid:guid()]) -> 'ok').
-spec(release/2 :: (server(), [rabbit_guid:guid()]) -> 'ok').
-spec(sync/3 :: (server(), [rabbit_guid:guid()], fun (() -> any())) -> 'ok').
-spec(gc_done/4 :: (server(), non_neg_integer(), file_num(), file_num()) ->
                        'ok').
-spec(set_maximum_since_use/2 :: (server(), non_neg_integer()) -> 'ok').
-spec(client_init/2 :: (server(), binary()) -> client_msstate()).
-spec(client_terminate/2 :: (client_msstate(), server()) -> 'ok').
-spec(client_delete_and_terminate/3 ::
        (client_msstate(), server(), binary()) -> 'ok').
-spec(successfully_recovered_state/1 :: (server()) -> boolean()).

-spec(gc/3 :: (non_neg_integer(), non_neg_integer(),
               {ets:tid(), file:filename(), atom(), any()}) ->
                   'concurrent_readers' | non_neg_integer()).

-endif.

%%----------------------------------------------------------------------------

%% We run GC whenever (garbage / sum_file_size) > ?GARBAGE_FRACTION
%% It is not recommended to set this to < 0.5
-define(GARBAGE_FRACTION,      0.5).

%% The components:
%%
%% Index: this is a mapping from Guid to #msg_location{}:
%%        {Guid, RefCount, File, Offset, TotalSize}
%%        By default, it's in ets, but it's also pluggable.
%% FileSummary: this is an ets table which maps File to #file_summary{}:
%%        {File, ValidTotalSize, ContiguousTop, Left, Right,
%%         FileSize, Locked, Readers}
%%
%% The basic idea is that messages are appended to the current file up
%% until that file becomes too big (> file_size_limit). At that point,
%% the file is closed and a new file is created on the _right_ of the
%% old file which is used for new messages. Files are named
%% numerically ascending, thus the file with the lowest name is the
%% eldest file.
%%
%% We need to keep track of which messages are in which files (this is
%% the Index); how much useful data is in each file and which files
%% are on the left and right of each other. This is the purpose of the
%% FileSummary ets table.
%%
%% As messages are removed from files, holes appear in these
%% files. The field ValidTotalSize contains the total amount of useful
%% data left in the file, whilst ContiguousTop contains the amount of
%% valid data right at the start of each file. These are needed for
%% garbage collection.
%%
%% When we discover that a file is now empty, we delete it. When we
%% discover that it can be combined with the useful data in either its
%% left or right neighbour, and overall, across all the files, we have
%% ((the amount of garbage) / (the sum of all file sizes)) >
%% ?GARBAGE_FRACTION, we start a garbage collection run concurrently,
%% which will compact the two files together. This keeps disk
%% utilisation high and aids performance. We deliberately do this
%% lazily in order to prevent doing GC on files which are soon to be
%% emptied (and hence deleted) soon.
%%
%% Given the compaction between two files, the left file (i.e. elder
%% file) is considered the ultimate destination for the good data in
%% the right file. If necessary, the good data in the left file which
%% is fragmented throughout the file is written out to a temporary
%% file, then read back in to form a contiguous chunk of good data at
%% the start of the left file. Thus the left file is garbage collected
%% and compacted. Then the good data from the right file is copied
%% onto the end of the left file. Index and FileSummary tables are
%% updated.
%%
%% On non-clean startup, we scan the files we discover, dealing with
%% the possibilites of a crash having occured during a compaction
%% (this consists of tidyup - the compaction is deliberately designed
%% such that data is duplicated on disk rather than risking it being
%% lost), and rebuild the FileSummary ets table and Index.
%%
%% So, with this design, messages move to the left. Eventually, they
%% should end up in a contiguous block on the left and are then never
%% rewritten. But this isn't quite the case. If in a file there is one
%% message that is being ignored, for some reason, and messages in the
%% file to the right and in the current block are being read all the
%% time then it will repeatedly be the case that the good data from
%% both files can be combined and will be written out to a new
%% file. Whenever this happens, our shunned message will be rewritten.
%%
%% So, provided that we combine messages in the right order,
%% (i.e. left file, bottom to top, right file, bottom to top),
%% eventually our shunned message will end up at the bottom of the
%% left file. The compaction/combining algorithm is smart enough to
%% read in good data from the left file that is scattered throughout
%% (i.e. C and D in the below diagram), then truncate the file to just
%% above B (i.e. truncate to the limit of the good contiguous region
%% at the start of the file), then write C and D on top and then write
%% E, F and G from the right file on top. Thus contiguous blocks of
%% good data at the bottom of files are not rewritten (yes, this is
%% the data the size of which is tracked by the ContiguousTop
%% variable. Judicious use of a mirror is required).
%%
%% +-------+    +-------+         +-------+
%% |   X   |    |   G   |         |   G   |
%% +-------+    +-------+         +-------+
%% |   D   |    |   X   |         |   F   |
%% +-------+    +-------+         +-------+
%% |   X   |    |   X   |         |   E   |
%% +-------+    +-------+         +-------+
%% |   C   |    |   F   |   ===>  |   D   |
%% +-------+    +-------+         +-------+
%% |   X   |    |   X   |         |   C   |
%% +-------+    +-------+         +-------+
%% |   B   |    |   X   |         |   B   |
%% +-------+    +-------+         +-------+
%% |   A   |    |   E   |         |   A   |
%% +-------+    +-------+         +-------+
%%   left         right             left
%%
%% From this reasoning, we do have a bound on the number of times the
%% message is rewritten. From when it is inserted, there can be no
%% files inserted between it and the head of the queue, and the worst
%% case is that everytime it is rewritten, it moves one position lower
%% in the file (for it to stay at the same position requires that
%% there are no holes beneath it, which means truncate would be used
%% and so it would not be rewritten at all). Thus this seems to
%% suggest the limit is the number of messages ahead of it in the
%% queue, though it's likely that that's pessimistic, given the
%% requirements for compaction/combination of files.
%%
%% The other property is that we have is the bound on the lowest
%% utilisation, which should be 50% - worst case is that all files are
%% fractionally over half full and can't be combined (equivalent is
%% alternating full files and files with only one tiny message in
%% them).
%%
%% Messages are reference-counted. When a message with the same guid
%% is written several times we only store it once, and only remove it
%% from the store when it has been removed the same number of times.
%%
%% The reference counts do not persist. Therefore the initialisation
%% function must be provided with a generator that produces ref count
%% deltas for all recovered messages. This is only used on startup
%% when the shutdown was non-clean.
%%
%% Read messages with a reference count greater than one are entered
%% into a message cache. The purpose of the cache is not especially
%% performance, though it can help there too, but prevention of memory
%% explosion. It ensures that as messages with a high reference count
%% are read from several processes they are read back as the same
%% binary object rather than multiples of identical binary
%% objects.
%%
%% Reads can be performed directly by clients without calling to the
%% server. This is safe because multiple file handles can be used to
%% read files. However, locking is used by the concurrent GC to make
%% sure that reads are not attempted from files which are in the
%% process of being garbage collected.
%%
%% The server automatically defers reads, removes and contains calls
%% that occur which refer to files which are currently being
%% GC'd. Contains calls are only deferred in order to ensure they do
%% not overtake removes.
%%
%% The current file to which messages are being written has a
%% write-back cache. This is written to immediately by clients and can
%% be read from by clients too. This means that there are only ever
%% writes made to the current file, thus eliminating delays due to
%% flushing write buffers in order to be able to safely read from the
%% current file. The one exception to this is that on start up, the
%% cache is not populated with msgs found in the current file, and
%% thus in this case only, reads may have to come from the file
%% itself. The effect of this is that even if the msg_store process is
%% heavily overloaded, clients can still write and read messages with
%% very low latency and not block at all.
%%
%% For notes on Clean Shutdown and startup, see documentation in
%% variable_queue.

%%----------------------------------------------------------------------------
%% public API
%%----------------------------------------------------------------------------

start_link(Server, Dir, ClientRefs, StartupFunState) ->
    gen_server2:start_link({local, Server}, ?MODULE,
                           [Server, Dir, ClientRefs, StartupFunState],
                           [{timeout, infinity}]).

write(Server, Guid, Msg,
      CState = #client_msstate { cur_file_cache_ets = CurFileCacheEts }) ->
    ok = update_msg_cache(CurFileCacheEts, Guid, Msg),
    {gen_server2:cast(Server, {write, Guid}), CState}.

read(Server, Guid,
     CState = #client_msstate { dedup_cache_ets    = DedupCacheEts,
                                cur_file_cache_ets = CurFileCacheEts }) ->
    %% 1. Check the dedup cache
    case fetch_and_increment_cache(DedupCacheEts, Guid) of
        not_found ->
            %% 2. Check the cur file cache
            case ets:lookup(CurFileCacheEts, Guid) of
                [] ->
                    Defer = fun() -> {gen_server2:pcall(
                                        Server, 2, {read, Guid}, infinity),
                                      CState} end,
                    case index_lookup(Guid, CState) of
                        not_found   -> Defer();
                        MsgLocation -> client_read1(Server, MsgLocation, Defer,
                                                    CState)
                    end;
                [{Guid, Msg, _CacheRefCount}] ->
                    %% Although we've found it, we don't know the
                    %% refcount, so can't insert into dedup cache
                    {{ok, Msg}, CState}
            end;
        Msg ->
            {{ok, Msg}, CState}
    end.

contains(Server, Guid) -> gen_server2:call(Server, {contains, Guid}, infinity).
remove(_Server, [])    -> ok;
remove(Server, Guids)  -> gen_server2:cast(Server, {remove, Guids}).
release(_Server, [])   -> ok;
release(Server, Guids) -> gen_server2:cast(Server, {release, Guids}).
sync(Server, Guids, K) -> gen_server2:cast(Server, {sync, Guids, K}).
sync(Server)           -> gen_server2:pcast(Server, 8, sync). %% internal

gc_done(Server, Reclaimed, Source, Destination) ->
    gen_server2:pcast(Server, 8, {gc_done, Reclaimed, Source, Destination}).

set_maximum_since_use(Server, Age) ->
    gen_server2:pcast(Server, 8, {set_maximum_since_use, Age}).

client_init(Server, Ref) ->
    {IState, IModule, Dir, GCPid,
     FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts} =
        gen_server2:pcall(Server, 7, {new_client_state, Ref}, infinity),
    #client_msstate { file_handle_cache  = dict:new(),
                      index_state        = IState,
                      index_module       = IModule,
                      dir                = Dir,
                      gc_pid             = GCPid,
                      file_handles_ets   = FileHandlesEts,
                      file_summary_ets   = FileSummaryEts,
                      dedup_cache_ets    = DedupCacheEts,
                      cur_file_cache_ets = CurFileCacheEts }.

client_terminate(CState, Server) ->
    close_all_handles(CState),
    ok = gen_server2:call(Server, client_terminate, infinity).

client_delete_and_terminate(CState, Server, Ref) ->
    close_all_handles(CState),
    ok = gen_server2:cast(Server, {client_delete, Ref}).

successfully_recovered_state(Server) ->
    gen_server2:pcall(Server, 7, successfully_recovered_state, infinity).

%%----------------------------------------------------------------------------
%% Client-side-only helpers
%%----------------------------------------------------------------------------

client_read1(Server,
             #msg_location { guid = Guid, file = File } = MsgLocation,
             Defer,
             CState = #client_msstate { file_summary_ets = FileSummaryEts }) ->
    case ets:lookup(FileSummaryEts, File) of
        [] -> %% File has been GC'd and no longer exists. Go around again.
            read(Server, Guid, CState);
        [#file_summary { locked = Locked, right = Right }] ->
            client_read2(Server, Locked, Right, MsgLocation, Defer, CState)
    end.

client_read2(_Server, false, undefined, _MsgLocation, Defer, _CState) ->
    %% Although we've already checked both caches and not found the
    %% message there, the message is apparently in the
    %% current_file. We can only arrive here if we are trying to read
    %% a message which we have not written, which is very odd, so just
    %% defer.
    %%
    %% OR, on startup, the cur_file_cache is not populated with the
    %% contents of the current file, thus reads from the current file
    %% will end up here and will need to be deferred.
    Defer();
client_read2(_Server, true, _Right, _MsgLocation, Defer, _CState) ->
    %% Of course, in the mean time, the GC could have run and our msg
    %% is actually in a different file, unlocked. However, defering is
    %% the safest and simplest thing to do.
    Defer();
client_read2(Server, false, _Right,
             MsgLocation = #msg_location { guid = Guid, file = File },
             Defer,
             CState = #client_msstate { file_summary_ets = FileSummaryEts }) ->
    %% It's entirely possible that everything we're doing from here on
    %% is for the wrong file, or a non-existent file, as a GC may have
    %% finished.
    safe_ets_update_counter(
      FileSummaryEts, File, {#file_summary.readers, +1},
      fun (_) -> client_read3(Server, MsgLocation, Defer, CState) end,
      fun () -> read(Server, Guid, CState) end).

client_read3(Server, #msg_location { guid = Guid, file = File }, Defer,
             CState = #client_msstate { file_handles_ets = FileHandlesEts,
                                        file_summary_ets = FileSummaryEts,
                                        dedup_cache_ets  = DedupCacheEts,
                                        gc_pid           = GCPid }) ->
    Release =
        fun() -> ok = case ets:update_counter(FileSummaryEts, File,
                                              {#file_summary.readers, -1}) of
                          0 -> case ets:lookup(FileSummaryEts, File) of
                                   [#file_summary { locked = true }] ->
                                       rabbit_msg_store_gc:no_readers(
                                         GCPid, File);
                                   _ -> ok
                               end;
                          _ -> ok
                      end
        end,
    %% If a GC involving the file hasn't already started, it won't
    %% start now. Need to check again to see if we've been locked in
    %% the meantime, between lookup and update_counter (thus GC
    %% started before our +1. In fact, it could have finished by now
    %% too).
    case ets:lookup(FileSummaryEts, File) of
        [] -> %% GC has deleted our file, just go round again.
            read(Server, Guid, CState);
        [#file_summary { locked = true }] ->
            %% If we get a badarg here, then the GC has finished and
            %% deleted our file. Try going around again. Otherwise,
            %% just defer.
            %%
            %% badarg scenario: we lookup, msg_store locks, GC starts,
            %% GC ends, we +1 readers, msg_store ets:deletes (and
            %% unlocks the dest)
            try Release(),
                Defer()
            catch error:badarg -> read(Server, Guid, CState)
            end;
        [#file_summary { locked = false }] ->
            %% Ok, we're definitely safe to continue - a GC involving
            %% the file cannot start up now, and isn't running, so
            %% nothing will tell us from now on to close the handle if
            %% it's already open.
            %%
            %% Finally, we need to recheck that the msg is still at
            %% the same place - it's possible an entire GC ran between
            %% us doing the lookup and the +1 on the readers. (Same as
            %% badarg scenario above, but we don't have a missing file
            %% - we just have the /wrong/ file).
            case index_lookup(Guid, CState) of
                #msg_location { file = File } = MsgLocation ->
                    %% Still the same file.
                    mark_handle_open(FileHandlesEts, File),

                    CState1 = close_all_indicated(CState),
                    {Msg, CState2} = %% This will never be the current file
                        read_from_disk(MsgLocation, CState1, DedupCacheEts),
                    Release(), %% this MUST NOT fail with badarg
                    {{ok, Msg}, CState2};
                MsgLocation -> %% different file!
                    Release(), %% this MUST NOT fail with badarg
                    client_read1(Server, MsgLocation, Defer, CState)
            end
    end.

%%----------------------------------------------------------------------------
%% gen_server callbacks
%%----------------------------------------------------------------------------

init([Server, BaseDir, ClientRefs, StartupFunState]) ->
    process_flag(trap_exit, true),

    ok = file_handle_cache:register_callback(?MODULE, set_maximum_since_use,
                                             [self()]),

    Dir = filename:join(BaseDir, atom_to_list(Server)),

    {ok, IndexModule} = application:get_env(msg_store_index_module),
    rabbit_log:info("~w: using ~p to provide index~n", [Server, IndexModule]),

    AttemptFileSummaryRecovery =
        case ClientRefs of
            undefined -> ok = rabbit_misc:recursive_delete([Dir]),
                         ok = filelib:ensure_dir(filename:join(Dir, "nothing")),
                         false;
            _         -> ok = filelib:ensure_dir(filename:join(Dir, "nothing")),
                         recover_crashed_compactions(Dir)
        end,

    %% if we found crashed compactions we trust neither the
    %% file_summary nor the location index. Note the file_summary is
    %% left empty here if it can't be recovered.
    {FileSummaryRecovered, FileSummaryEts} =
        recover_file_summary(AttemptFileSummaryRecovery, Dir),

    {CleanShutdown, IndexState, ClientRefs1} =
        recover_index_and_client_refs(IndexModule, FileSummaryRecovered,
                                      ClientRefs, Dir, Server),
    %% CleanShutdown => msg location index and file_summary both
    %% recovered correctly.
    true = case {FileSummaryRecovered, CleanShutdown} of
               {true, false} -> ets:delete_all_objects(FileSummaryEts);
               _             -> true
           end,
    %% CleanShutdown <=> msg location index and file_summary both
    %% recovered correctly.

    DedupCacheEts   = ets:new(rabbit_msg_store_dedup_cache, [set, public]),
    FileHandlesEts  = ets:new(rabbit_msg_store_shared_file_handles,
                              [ordered_set, public]),
    CurFileCacheEts = ets:new(rabbit_msg_store_cur_file, [set, public]),

    {ok, FileSizeLimit} = application:get_env(msg_store_file_size_limit),

    State = #msstate { dir                    = Dir,
                       index_module           = IndexModule,
                       index_state            = IndexState,
                       current_file           = 0,
                       current_file_handle    = undefined,
                       file_handle_cache      = dict:new(),
                       on_sync                = [],
                       sync_timer_ref         = undefined,
                       sum_valid_data         = 0,
                       sum_file_size          = 0,
                       pending_gc_completion  = [],
                       gc_active              = false,
                       gc_pid                 = undefined,
                       file_handles_ets       = FileHandlesEts,
                       file_summary_ets       = FileSummaryEts,
                       dedup_cache_ets        = DedupCacheEts,
                       cur_file_cache_ets     = CurFileCacheEts,
                       client_refs            = ClientRefs1,
                       successfully_recovered = CleanShutdown,
                       file_size_limit        = FileSizeLimit
                      },

    %% If we didn't recover the msg location index then we need to
    %% rebuild it now.
    {Offset, State1 = #msstate { current_file = CurFile }} =
        build_index(CleanShutdown, StartupFunState, State),

    %% read is only needed so that we can seek
    {ok, CurHdl} = open_file(Dir, filenum_to_name(CurFile),
                             [read | ?WRITE_MODE]),
    {ok, Offset} = file_handle_cache:position(CurHdl, Offset),
    ok = file_handle_cache:truncate(CurHdl),

    {ok, GCPid} = rabbit_msg_store_gc:start_link(Dir, IndexState, IndexModule,
                                                 FileSummaryEts),

    {ok, maybe_compact(
           State1 #msstate { current_file_handle = CurHdl, gc_pid = GCPid }),
     hibernate,
     {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.

handle_call({read, Guid}, From, State) ->
    State1 = read_message(Guid, From, State),
    noreply(State1);

handle_call({contains, Guid}, From, State) ->
    State1 = contains_message(Guid, From, State),
    noreply(State1);

handle_call({new_client_state, CRef}, _From,
            State = #msstate { dir                = Dir,
                               index_state        = IndexState,
                               index_module       = IndexModule,
                               file_handles_ets   = FileHandlesEts,
                               file_summary_ets   = FileSummaryEts,
                               dedup_cache_ets    = DedupCacheEts,
                               cur_file_cache_ets = CurFileCacheEts,
                               client_refs        = ClientRefs,
                               gc_pid             = GCPid }) ->
    reply({IndexState, IndexModule, Dir, GCPid,
           FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts},
          State #msstate { client_refs = sets:add_element(CRef, ClientRefs) });

handle_call(successfully_recovered_state, _From, State) ->
    reply(State #msstate.successfully_recovered, State);

handle_call(client_terminate, _From, State) ->
    reply(ok, State).

handle_cast({write, Guid},
            State = #msstate { current_file_handle = CurHdl,
                               current_file        = CurFile,
                               sum_valid_data      = SumValid,
                               sum_file_size       = SumFileSize,
                               file_summary_ets    = FileSummaryEts,
                               cur_file_cache_ets  = CurFileCacheEts }) ->
    true = 0 =< ets:update_counter(CurFileCacheEts, Guid, {3, -1}),
    [{Guid, Msg, _CacheRefCount}] = ets:lookup(CurFileCacheEts, Guid),
    case index_lookup(Guid, State) of
        not_found ->
            %% New message, lots to do
            {ok, CurOffset} = file_handle_cache:current_virtual_offset(CurHdl),
            {ok, TotalSize} = rabbit_msg_file:append(CurHdl, Guid, Msg),
            ok = index_insert(#msg_location {
                                guid = Guid, ref_count = 1, file = CurFile,
                                offset = CurOffset, total_size = TotalSize },
                              State),
            [#file_summary { valid_total_size = ValidTotalSize,
                             contiguous_top   = ContiguousTop,
                             right            = undefined,
                             locked           = false,
                             file_size        = FileSize }] =
                ets:lookup(FileSummaryEts, CurFile),
            ValidTotalSize1 = ValidTotalSize + TotalSize,
            ContiguousTop1 = case CurOffset =:= ContiguousTop of
                                 true  -> ValidTotalSize1;
                                 false -> ContiguousTop
                             end,
            true = ets:update_element(
                     FileSummaryEts, CurFile,
                     [{#file_summary.valid_total_size, ValidTotalSize1},
                      {#file_summary.contiguous_top,   ContiguousTop1},
                      {#file_summary.file_size,        FileSize + TotalSize}]),
            NextOffset = CurOffset + TotalSize,
            noreply(
              maybe_roll_to_new_file(
                NextOffset, State #msstate {
                              sum_valid_data = SumValid + TotalSize,
                              sum_file_size  = SumFileSize + TotalSize }));
        #msg_location { ref_count = RefCount } ->
            %% We already know about it, just update counter. Only
            %% update field otherwise bad interaction with concurrent GC
            ok = index_update_fields(Guid,
                                     {#msg_location.ref_count, RefCount + 1},
                                     State),
            noreply(State)
    end;

handle_cast({remove, Guids}, State) ->
    State1 = lists:foldl(
               fun (Guid, State2) -> remove_message(Guid, State2) end,
               State, Guids),
    noreply(maybe_compact(State1));

handle_cast({release, Guids}, State =
                #msstate { dedup_cache_ets = DedupCacheEts }) ->
    lists:foreach(
      fun (Guid) -> decrement_cache(DedupCacheEts, Guid) end, Guids),
    noreply(State);

handle_cast({sync, Guids, K},
            State = #msstate { current_file        = CurFile,
                               current_file_handle = CurHdl,
                               on_sync             = Syncs }) ->
    {ok, SyncOffset} = file_handle_cache:last_sync_offset(CurHdl),
    case lists:any(fun (Guid) ->
                           #msg_location { file = File, offset = Offset } =
                               index_lookup(Guid, State),
                           File =:= CurFile andalso Offset >= SyncOffset
                   end, Guids) of
        false -> K(),
                 noreply(State);
        true  -> noreply(State #msstate { on_sync = [K | Syncs] })
    end;

handle_cast(sync, State) ->
    noreply(internal_sync(State));

handle_cast({gc_done, Reclaimed, Src, Dst},
            State = #msstate { sum_file_size    = SumFileSize,
                               gc_active        = {Src, Dst},
                               file_handles_ets = FileHandlesEts,
                               file_summary_ets = FileSummaryEts }) ->
    %% GC done, so now ensure that any clients that have open fhs to
    %% those files close them before using them again. This has to be
    %% done here (given it's done in the msg_store, and not the gc),
    %% and not when starting up the GC, because if done when starting
    %% up the GC, the client could find the close, and close and
    %% reopen the fh, whilst the GC is waiting for readers to
    %% disappear, before it's actually done the GC.
    true = mark_handle_to_close(FileHandlesEts, Src),
    true = mark_handle_to_close(FileHandlesEts, Dst),
    %% we always move data left, so Src has gone and was on the
    %% right, so need to make dest = source.right.left, and also
    %% dest.right = source.right
    [#file_summary { left    = Dst,
                     right   = SrcRight,
                     locked  = true,
                     readers = 0 }] = ets:lookup(FileSummaryEts, Src),
    %% this could fail if SrcRight =:= undefined
    ets:update_element(FileSummaryEts, SrcRight, {#file_summary.left, Dst}),
    true = ets:update_element(FileSummaryEts, Dst,
                              [{#file_summary.locked, false},
                               {#file_summary.right,  SrcRight}]),
    true = ets:delete(FileSummaryEts, Src),
    noreply(
      maybe_compact(run_pending(
                      State #msstate { sum_file_size = SumFileSize - Reclaimed,
                                       gc_active     = false })));

handle_cast({set_maximum_since_use, Age}, State) ->
    ok = file_handle_cache:set_maximum_since_use(Age),
    noreply(State);

handle_cast({client_delete, CRef},
            State = #msstate { client_refs = ClientRefs }) ->
    noreply(
      State #msstate { client_refs = sets:del_element(CRef, ClientRefs) }).

handle_info(timeout, State) ->
    noreply(internal_sync(State));

handle_info({'EXIT', _Pid, Reason}, State) ->
    {stop, Reason, State}.

terminate(_Reason, State = #msstate { index_state         = IndexState,
                                      index_module        = IndexModule,
                                      current_file_handle = CurHdl,
                                      gc_pid              = GCPid,
                                      file_handles_ets    = FileHandlesEts,
                                      file_summary_ets    = FileSummaryEts,
                                      dedup_cache_ets     = DedupCacheEts,
                                      cur_file_cache_ets  = CurFileCacheEts,
                                      client_refs         = ClientRefs,
                                      dir                 = Dir }) ->
    %% stop the gc first, otherwise it could be working and we pull
    %% out the ets tables from under it.
    ok = rabbit_msg_store_gc:stop(GCPid),
    State1 = case CurHdl of
                 undefined -> State;
                 _         -> State2 = internal_sync(State),
                              file_handle_cache:close(CurHdl),
                              State2
             end,
    State3 = close_all_handles(State1),
    store_file_summary(FileSummaryEts, Dir),
    [ets:delete(T) ||
        T <- [FileSummaryEts, DedupCacheEts, FileHandlesEts, CurFileCacheEts]],
    IndexModule:terminate(IndexState),
    store_recovery_terms([{client_refs, sets:to_list(ClientRefs)},
                          {index_module, IndexModule}], Dir),
    State3 #msstate { index_state         = undefined,
                      current_file_handle = undefined }.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%----------------------------------------------------------------------------
%% general helper functions
%%----------------------------------------------------------------------------

noreply(State) ->
    {State1, Timeout} = next_state(State),
    {noreply, State1, Timeout}.

reply(Reply, State) ->
    {State1, Timeout} = next_state(State),
    {reply, Reply, State1, Timeout}.

next_state(State = #msstate { on_sync = [], sync_timer_ref = undefined }) ->
    {State, hibernate};
next_state(State = #msstate { sync_timer_ref = undefined }) ->
    {start_sync_timer(State), 0};
next_state(State = #msstate { on_sync = [] }) ->
    {stop_sync_timer(State), hibernate};
next_state(State) ->
    {State, 0}.

start_sync_timer(State = #msstate { sync_timer_ref = undefined }) ->
    {ok, TRef} = timer:apply_after(?SYNC_INTERVAL, ?MODULE, sync, [self()]),
    State #msstate { sync_timer_ref = TRef }.

stop_sync_timer(State = #msstate { sync_timer_ref = undefined }) ->
    State;
stop_sync_timer(State = #msstate { sync_timer_ref = TRef }) ->
    {ok, cancel} = timer:cancel(TRef),
    State #msstate { sync_timer_ref = undefined }.

internal_sync(State = #msstate { current_file_handle = CurHdl,
                                 on_sync = Syncs }) ->
    State1 = stop_sync_timer(State),
    case Syncs of
        [] -> State1;
        _  -> ok = file_handle_cache:sync(CurHdl),
              lists:foreach(fun (K) -> K() end, lists:reverse(Syncs)),
              State1 #msstate { on_sync = [] }
    end.

read_message(Guid, From,
             State = #msstate { dedup_cache_ets = DedupCacheEts }) ->
    case index_lookup(Guid, State) of
        not_found ->
            gen_server2:reply(From, not_found),
            State;
        MsgLocation ->
            case fetch_and_increment_cache(DedupCacheEts, Guid) of
                not_found -> read_message1(From, MsgLocation, State);
                Msg       -> gen_server2:reply(From, {ok, Msg}),
                             State
            end
    end.

read_message1(From, #msg_location { guid = Guid, ref_count = RefCount,
                                    file = File, offset = Offset } = MsgLoc,
              State = #msstate { current_file        = CurFile,
                                 current_file_handle = CurHdl,
                                 file_summary_ets    = FileSummaryEts,
                                 dedup_cache_ets     = DedupCacheEts,
                                 cur_file_cache_ets  = CurFileCacheEts }) ->
    case File =:= CurFile of
        true  -> {Msg, State1} =
                     %% can return [] if msg in file existed on startup
                     case ets:lookup(CurFileCacheEts, Guid) of
                         [] ->
                             {ok, RawOffSet} =
                                 file_handle_cache:current_raw_offset(CurHdl),
                             ok = case Offset >= RawOffSet of
                                      true  -> file_handle_cache:flush(CurHdl);
                                      false -> ok
                                  end,
                             read_from_disk(MsgLoc, State, DedupCacheEts);
                         [{Guid, Msg1, _CacheRefCount}] ->
                             ok = maybe_insert_into_cache(
                                    DedupCacheEts, RefCount, Guid, Msg1),
                             {Msg1, State}
                     end,
                 gen_server2:reply(From, {ok, Msg}),
                 State1;
        false -> [#file_summary { locked = Locked }] =
                     ets:lookup(FileSummaryEts, File),
                 case Locked of
                     true  -> add_to_pending_gc_completion({read, Guid, From},
                                                           State);
                     false -> {Msg, State1} =
                                  read_from_disk(MsgLoc, State, DedupCacheEts),
                              gen_server2:reply(From, {ok, Msg}),
                              State1
                 end
    end.

read_from_disk(#msg_location { guid = Guid, ref_count = RefCount,
                               file = File, offset = Offset,
                               total_size = TotalSize },
               State, DedupCacheEts) ->
    {Hdl, State1} = get_read_handle(File, State),
    {ok, Offset} = file_handle_cache:position(Hdl, Offset),
    {ok, {Guid, Msg}} =
        case rabbit_msg_file:read(Hdl, TotalSize) of
            {ok, {Guid, _}} = Obj ->
                Obj;
            Rest ->
                {error, {misread, [{old_state, State},
                                   {file_num,  File},
                                   {offset,    Offset},
                                   {guid,      Guid},
                                   {read,      Rest},
                                   {proc_dict, get()}
                                  ]}}
        end,
    ok = maybe_insert_into_cache(DedupCacheEts, RefCount, Guid, Msg),
    {Msg, State1}.

contains_message(Guid, From, State = #msstate { gc_active = GCActive }) ->
    case index_lookup(Guid, State) of
        not_found ->
            gen_server2:reply(From, false),
            State;
        #msg_location { file = File } ->
            case GCActive of
                {A, B} when File =:= A orelse File =:= B ->
                    add_to_pending_gc_completion(
                      {contains, Guid, From}, State);
                _ ->
                    gen_server2:reply(From, true),
                    State
            end
    end.

remove_message(Guid, State = #msstate { sum_valid_data   = SumValid,
                                        file_summary_ets = FileSummaryEts,
                                        dedup_cache_ets  = DedupCacheEts }) ->
    #msg_location { ref_count = RefCount, file = File,
                    offset = Offset, total_size = TotalSize } =
        index_lookup(Guid, State),
    case RefCount of
        1 ->
            %% don't remove from CUR_FILE_CACHE_ETS_NAME here because
            %% there may be further writes in the mailbox for the same
            %% msg.
            ok = remove_cache_entry(DedupCacheEts, Guid),
            [#file_summary { valid_total_size = ValidTotalSize,
                             contiguous_top   = ContiguousTop,
                             locked           = Locked }] =
                ets:lookup(FileSummaryEts, File),
            case Locked of
                true ->
                    add_to_pending_gc_completion({remove, Guid}, State);
                false ->
                    ok = index_delete(Guid, State),
                    ContiguousTop1 = lists:min([ContiguousTop, Offset]),
                    ValidTotalSize1 = ValidTotalSize - TotalSize,
                    true = ets:update_element(
                             FileSummaryEts, File,
                             [{#file_summary.valid_total_size, ValidTotalSize1},
                              {#file_summary.contiguous_top, ContiguousTop1}]),
                    State1 = delete_file_if_empty(File, State),
                    State1 #msstate { sum_valid_data = SumValid - TotalSize }
            end;
        _ when 1 < RefCount ->
            ok = decrement_cache(DedupCacheEts, Guid),
            %% only update field, otherwise bad interaction with concurrent GC
            ok = index_update_fields(Guid,
                                     {#msg_location.ref_count, RefCount - 1},
                                     State),
            State
    end.

add_to_pending_gc_completion(
  Op, State = #msstate { pending_gc_completion = Pending }) ->
    State #msstate { pending_gc_completion = [Op | Pending] }.

run_pending(State = #msstate { pending_gc_completion = [] }) ->
    State;
run_pending(State = #msstate { pending_gc_completion = Pending }) ->
    State1 = State #msstate { pending_gc_completion = [] },
    lists:foldl(fun run_pending/2, State1, lists:reverse(Pending)).

run_pending({read, Guid, From}, State) ->
    read_message(Guid, From, State);
run_pending({contains, Guid, From}, State) ->
    contains_message(Guid, From, State);
run_pending({remove, Guid}, State) ->
    remove_message(Guid, State).

safe_ets_update_counter(Tab, Key, UpdateOp, SuccessFun, FailThunk) ->
    try
        SuccessFun(ets:update_counter(Tab, Key, UpdateOp))
    catch error:badarg -> FailThunk()
    end.

safe_ets_update_counter_ok(Tab, Key, UpdateOp, FailThunk) ->
    safe_ets_update_counter(Tab, Key, UpdateOp, fun (_) -> ok end, FailThunk).

%%----------------------------------------------------------------------------
%% file helper functions
%%----------------------------------------------------------------------------

open_file(Dir, FileName, Mode) ->
    file_handle_cache:open(form_filename(Dir, FileName), ?BINARY_MODE ++ Mode,
                           [{write_buffer, ?HANDLE_CACHE_BUFFER_SIZE}]).

close_handle(Key, CState = #client_msstate { file_handle_cache = FHC }) ->
    CState #client_msstate { file_handle_cache = close_handle(Key, FHC) };

close_handle(Key, State = #msstate { file_handle_cache = FHC }) ->
    State #msstate { file_handle_cache = close_handle(Key, FHC) };

close_handle(Key, FHC) ->
    case dict:find(Key, FHC) of
        {ok, Hdl} -> ok = file_handle_cache:close(Hdl),
                     dict:erase(Key, FHC);
        error     -> FHC
    end.

mark_handle_open(FileHandlesEts, File) ->
    %% This is fine to fail (already exists)
    ets:insert_new(FileHandlesEts, {{self(), File}, open}),
    true.

mark_handle_to_close(FileHandlesEts, File) ->
    [ ets:update_element(FileHandlesEts, Key, {2, close})
      || {Key, open} <- ets:match_object(FileHandlesEts, {{'_', File}, open}) ],
    true.

close_all_indicated(#client_msstate { file_handles_ets = FileHandlesEts } =
                    CState) ->
    Objs = ets:match_object(FileHandlesEts, {{self(), '_'}, close}),
    lists:foldl(fun ({Key = {_Self, File}, close}, CStateM) ->
                        true = ets:delete(FileHandlesEts, Key),
                        close_handle(File, CStateM)
                end, CState, Objs).

close_all_handles(CState = #client_msstate { file_handles_ets = FileHandlesEts,
                                             file_handle_cache = FHC }) ->
    Self = self(),
    ok = dict:fold(fun (File, Hdl, ok) ->
                           true = ets:delete(FileHandlesEts, {Self, File}),
                           file_handle_cache:close(Hdl)
                   end, ok, FHC),
    CState #client_msstate { file_handle_cache = dict:new() };

close_all_handles(State = #msstate { file_handle_cache = FHC }) ->
    ok = dict:fold(fun (_Key, Hdl, ok) -> file_handle_cache:close(Hdl) end,
                   ok, FHC),
    State #msstate { file_handle_cache = dict:new() }.

get_read_handle(FileNum, CState = #client_msstate { file_handle_cache = FHC,
                                                    dir = Dir }) ->
    {Hdl, FHC2} = get_read_handle(FileNum, FHC, Dir),
    {Hdl, CState #client_msstate { file_handle_cache = FHC2 }};

get_read_handle(FileNum, State = #msstate { file_handle_cache = FHC,
                                            dir = Dir }) ->
    {Hdl, FHC2} = get_read_handle(FileNum, FHC, Dir),
    {Hdl, State #msstate { file_handle_cache = FHC2 }}.

get_read_handle(FileNum, FHC, Dir) ->
    case dict:find(FileNum, FHC) of
        {ok, Hdl} -> {Hdl, FHC};
        error     -> {ok, Hdl} = open_file(Dir, filenum_to_name(FileNum),
                                           ?READ_MODE),
                     {Hdl, dict:store(FileNum, Hdl, FHC)}
    end.

preallocate(Hdl, FileSizeLimit, FinalPos) ->
    {ok, FileSizeLimit} = file_handle_cache:position(Hdl, FileSizeLimit),
    ok = file_handle_cache:truncate(Hdl),
    {ok, FinalPos} = file_handle_cache:position(Hdl, FinalPos),
    ok.

truncate_and_extend_file(Hdl, Lowpoint, Highpoint) ->
    {ok, Lowpoint} = file_handle_cache:position(Hdl, Lowpoint),
    ok = file_handle_cache:truncate(Hdl),
    ok = preallocate(Hdl, Highpoint, Lowpoint).

form_filename(Dir, Name) -> filename:join(Dir, Name).

filenum_to_name(File) -> integer_to_list(File) ++ ?FILE_EXTENSION.

filename_to_num(FileName) -> list_to_integer(filename:rootname(FileName)).

list_sorted_file_names(Dir, Ext) ->
    lists:sort(fun (A, B) -> filename_to_num(A) < filename_to_num(B) end,
               filelib:wildcard("*" ++ Ext, Dir)).

%%----------------------------------------------------------------------------
%% message cache helper functions
%%----------------------------------------------------------------------------

maybe_insert_into_cache(DedupCacheEts, RefCount, Guid, Msg)
  when RefCount > 1 ->
    update_msg_cache(DedupCacheEts, Guid, Msg);
maybe_insert_into_cache(_DedupCacheEts, _RefCount, _Guid, _Msg) ->
    ok.

update_msg_cache(CacheEts, Guid, Msg) ->
    case ets:insert_new(CacheEts, {Guid, Msg, 1}) of
        true  -> ok;
        false -> safe_ets_update_counter_ok(
                   CacheEts, Guid, {3, +1},
                   fun () -> update_msg_cache(CacheEts, Guid, Msg) end)
    end.

remove_cache_entry(DedupCacheEts, Guid) ->
    true = ets:delete(DedupCacheEts, Guid),
    ok.

fetch_and_increment_cache(DedupCacheEts, Guid) ->
    case ets:lookup(DedupCacheEts, Guid) of
        [] ->
            not_found;
        [{_Guid, Msg, _RefCount}] ->
            safe_ets_update_counter_ok(
              DedupCacheEts, Guid, {3, +1},
              %% someone has deleted us in the meantime, insert us
              fun () -> ok = update_msg_cache(DedupCacheEts, Guid, Msg) end),
            Msg
    end.

decrement_cache(DedupCacheEts, Guid) ->
    true = safe_ets_update_counter(
             DedupCacheEts, Guid, {3, -1},
             fun (N) when N =< 0 -> true = ets:delete(DedupCacheEts, Guid);
                 (_N)            -> true
             end,
             %% Guid is not in there because although it's been
             %% delivered, it's never actually been read (think:
             %% persistent message held in RAM)
             fun () -> true end),
    ok.

%%----------------------------------------------------------------------------
%% index
%%----------------------------------------------------------------------------

index_lookup(Key, #client_msstate { index_module = Index,
                                    index_state  = State }) ->
    Index:lookup(Key, State);

index_lookup(Key, #msstate { index_module = Index, index_state = State }) ->
    Index:lookup(Key, State).

index_insert(Obj, #msstate { index_module = Index, index_state = State }) ->
    Index:insert(Obj, State).

index_update(Obj, #msstate { index_module = Index, index_state = State }) ->
    Index:update(Obj, State).

index_update_fields(Key, Updates, #msstate { index_module = Index,
                                             index_state  = State }) ->
    Index:update_fields(Key, Updates, State).

index_delete(Key, #msstate { index_module = Index, index_state = State }) ->
    Index:delete(Key, State).

index_delete_by_file(File, #msstate { index_module = Index,
                                      index_state  = State }) ->
    Index:delete_by_file(File, State).

%%----------------------------------------------------------------------------
%% shutdown and recovery
%%----------------------------------------------------------------------------

recover_index_and_client_refs(IndexModule, _Recover, undefined, Dir, _Server) ->
    {false, IndexModule:new(Dir), sets:new()};
recover_index_and_client_refs(IndexModule, false, _ClientRefs, Dir, Server) ->
    rabbit_log:warning("~w: rebuilding indices from scratch~n", [Server]),
    {false, IndexModule:new(Dir), sets:new()};
recover_index_and_client_refs(IndexModule, true, ClientRefs, Dir, Server) ->
    Fresh = fun (ErrorMsg, ErrorArgs) ->
                    rabbit_log:warning("~w: " ++ ErrorMsg ++ "~n"
                                       "rebuilding indices from scratch~n",
                                       [Server | ErrorArgs]),
                    {false, IndexModule:new(Dir), sets:new()}
            end,
    case read_recovery_terms(Dir) of
        {false, Error} ->
            Fresh("failed to read recovery terms: ~p", [Error]);
        {true, Terms} ->
            RecClientRefs  = proplists:get_value(client_refs, Terms, []),
            RecIndexModule = proplists:get_value(index_module, Terms),
            case (lists:sort(ClientRefs) =:= lists:sort(RecClientRefs)
                  andalso IndexModule =:= RecIndexModule) of
                true  -> case IndexModule:recover(Dir) of
                             {ok, IndexState1} ->
                                 {true, IndexState1,
                                  sets:from_list(ClientRefs)};
                             {error, Error} ->
                                 Fresh("failed to recover index: ~p", [Error])
                         end;
                false -> Fresh("recovery terms differ from present", [])
            end
    end.

store_recovery_terms(Terms, Dir) ->
    rabbit_misc:write_term_file(filename:join(Dir, ?CLEAN_FILENAME), Terms).

read_recovery_terms(Dir) ->
    Path = filename:join(Dir, ?CLEAN_FILENAME),
    case rabbit_misc:read_term_file(Path) of
        {ok, Terms}    -> case file:delete(Path) of
                              ok             -> {true,  Terms};
                              {error, Error} -> {false, Error}
                          end;
        {error, Error} -> {false, Error}
    end.

store_file_summary(Tid, Dir) ->
    ok = ets:tab2file(Tid, filename:join(Dir, ?FILE_SUMMARY_FILENAME),
                      [{extended_info, [object_count]}]).

recover_file_summary(false, _Dir) ->
    %% TODO: the only reason for this to be an *ordered*_set is so
    %% that a) maybe_compact can start a traversal from the eldest
    %% file, and b) build_index in fast recovery mode can easily
    %% identify the current file. It's awkward to have both that
    %% odering and the left/right pointers in the entries - replacing
    %% the former with some additional bit of state would be easy, but
    %% ditching the latter would be neater.
    {false, ets:new(rabbit_msg_store_file_summary,
                    [ordered_set, public, {keypos, #file_summary.file}])};
recover_file_summary(true, Dir) ->
    Path = filename:join(Dir, ?FILE_SUMMARY_FILENAME),
    case ets:file2tab(Path) of
        {ok, Tid}       -> file:delete(Path),
                          {true, Tid};
        {error, _Error} -> recover_file_summary(false, Dir)
    end.

count_msg_refs(Gen, Seed, State) ->
    case Gen(Seed) of
        finished ->
            ok;
        {_Guid, 0, Next} ->
            count_msg_refs(Gen, Next, State);
        {Guid, Delta, Next} ->
            ok = case index_lookup(Guid, State) of
                     not_found ->
                         index_insert(#msg_location { guid = Guid,
                                                      file = undefined,
                                                      ref_count = Delta },
                                      State);
                     #msg_location { ref_count = RefCount } = StoreEntry ->
                         NewRefCount = RefCount + Delta,
                         case NewRefCount of
                             0 -> index_delete(Guid, State);
                             _ -> index_update(StoreEntry #msg_location {
                                                 ref_count = NewRefCount },
                                               State)
                         end
                 end,
            count_msg_refs(Gen, Next, State)
    end.

recover_crashed_compactions(Dir) ->
    FileNames =    list_sorted_file_names(Dir, ?FILE_EXTENSION),
    TmpFileNames = list_sorted_file_names(Dir, ?FILE_EXTENSION_TMP),
    lists:foreach(
      fun (TmpFileName) ->
              NonTmpRelatedFileName =
                  filename:rootname(TmpFileName) ++ ?FILE_EXTENSION,
              true = lists:member(NonTmpRelatedFileName, FileNames),
              ok = recover_crashed_compaction(
                     Dir, TmpFileName, NonTmpRelatedFileName)
      end, TmpFileNames),
    TmpFileNames == [].

recover_crashed_compaction(Dir, TmpFileName, NonTmpRelatedFileName) ->
    %% Because a msg can legitimately appear multiple times in the
    %% same file, identifying the contents of the tmp file and where
    %% they came from is non-trivial. If we are recovering a crashed
    %% compaction then we will be rebuilding the index, which can cope
    %% with duplicates appearing. Thus the simplest and safest thing
    %% to do is to append the contents of the tmp file to its main
    %% file.
    {ok, TmpHdl}  = open_file(Dir, TmpFileName, ?READ_MODE),
    {ok, MainHdl} = open_file(Dir, NonTmpRelatedFileName,
                              ?READ_MODE ++ ?WRITE_MODE),
    {ok, _End} = file_handle_cache:position(MainHdl, eof),
    Size = filelib:file_size(form_filename(Dir, TmpFileName)),
    {ok, Size} = file_handle_cache:copy(TmpHdl, MainHdl, Size),
    ok = file_handle_cache:close(MainHdl),
    ok = file_handle_cache:delete(TmpHdl),
    ok.

scan_file_for_valid_messages(Dir, FileName) ->
    case open_file(Dir, FileName, ?READ_MODE) of
        {ok, Hdl}       -> Valid = rabbit_msg_file:scan(
                                     Hdl, filelib:file_size(
                                            form_filename(Dir, FileName))),
                           %% if something really bad has happened,
                           %% the close could fail, but ignore
                           file_handle_cache:close(Hdl),
                           Valid;
        {error, enoent} -> {ok, [], 0};
        {error, Reason} -> {error, {unable_to_scan_file, FileName, Reason}}
    end.

%% Takes the list in *ascending* order (i.e. eldest message
%% first). This is the opposite of what scan_file_for_valid_messages
%% produces. The list of msgs that is produced is youngest first.
find_contiguous_block_prefix(L) -> find_contiguous_block_prefix(L, 0, []).

find_contiguous_block_prefix([], ExpectedOffset, Guids) ->
    {ExpectedOffset, Guids};
find_contiguous_block_prefix([{Guid, TotalSize, ExpectedOffset} | Tail],
                             ExpectedOffset, Guids) ->
    ExpectedOffset1 = ExpectedOffset + TotalSize,
    find_contiguous_block_prefix(Tail, ExpectedOffset1, [Guid | Guids]);
find_contiguous_block_prefix([_MsgAfterGap | _Tail], ExpectedOffset, Guids) ->
    {ExpectedOffset, Guids}.

build_index(true, _StartupFunState,
            State = #msstate { file_summary_ets = FileSummaryEts }) ->
    ets:foldl(
      fun (#file_summary { valid_total_size = ValidTotalSize,
                           file_size        = FileSize,
                           file             = File },
           {_Offset, State1 = #msstate { sum_valid_data = SumValid,
                                         sum_file_size  = SumFileSize }}) ->
              {FileSize, State1 #msstate {
                           sum_valid_data = SumValid + ValidTotalSize,
                           sum_file_size  = SumFileSize + FileSize,
                           current_file   = File }}
      end, {0, State}, FileSummaryEts);
build_index(false, {MsgRefDeltaGen, MsgRefDeltaGenInit},
            State = #msstate { dir = Dir }) ->
    ok = count_msg_refs(MsgRefDeltaGen, MsgRefDeltaGenInit, State),
    {ok, Pid} = gatherer:start_link(),
    case [filename_to_num(FileName) ||
             FileName <- list_sorted_file_names(Dir, ?FILE_EXTENSION)] of
        []     -> build_index(Pid, undefined, [State #msstate.current_file],
                              State);
        Files  -> {Offset, State1} = build_index(Pid, undefined, Files, State),
                  {Offset, lists:foldl(fun delete_file_if_empty/2,
                                       State1, Files)}
    end.

build_index(Gatherer, Left, [],
            State = #msstate { file_summary_ets = FileSummaryEts,
                               sum_valid_data   = SumValid,
                               sum_file_size    = SumFileSize }) ->
    case gatherer:out(Gatherer) of
        empty ->
            ok = gatherer:stop(Gatherer),
            ok = rabbit_misc:unlink_and_capture_exit(Gatherer),
            ok = index_delete_by_file(undefined, State),
            Offset = case ets:lookup(FileSummaryEts, Left) of
                         []                                       -> 0;
                         [#file_summary { file_size = FileSize }] -> FileSize
                     end,
            {Offset, State #msstate { current_file = Left }};
        {value, #file_summary { valid_total_size = ValidTotalSize,
                                file_size = FileSize } = FileSummary} ->
            true = ets:insert_new(FileSummaryEts, FileSummary),
            build_index(Gatherer, Left, [],
                        State #msstate {
                          sum_valid_data = SumValid + ValidTotalSize,
                          sum_file_size  = SumFileSize + FileSize })
    end;
build_index(Gatherer, Left, [File|Files], State) ->
    ok = gatherer:fork(Gatherer),
    ok = worker_pool:submit_async(
           fun () -> build_index_worker(Gatherer, State,
                                        Left, File, Files)
           end),
    build_index(Gatherer, File, Files, State).

build_index_worker(Gatherer, State = #msstate { dir = Dir },
                   Left, File, Files) ->
    {ok, Messages, FileSize} =
        scan_file_for_valid_messages(Dir, filenum_to_name(File)),
    {ValidMessages, ValidTotalSize} =
        lists:foldl(
          fun (Obj = {Guid, TotalSize, Offset}, {VMAcc, VTSAcc}) ->
                  case index_lookup(Guid, State) of
                      #msg_location { file = undefined } = StoreEntry ->
                          ok = index_update(StoreEntry #msg_location {
                                              file = File, offset = Offset,
                                              total_size = TotalSize },
                                            State),
                          {[Obj | VMAcc], VTSAcc + TotalSize};
                      _ ->
                          {VMAcc, VTSAcc}
                  end
          end, {[], 0}, Messages),
    %% foldl reverses lists, find_contiguous_block_prefix needs
    %% msgs eldest first, so, ValidMessages is the right way round
    {ContiguousTop, _} = find_contiguous_block_prefix(ValidMessages),
    {Right, FileSize1} =
        case Files of
            %% if it's the last file, we'll truncate to remove any
            %% rubbish above the last valid message. This affects the
            %% file size.
            []    -> {undefined, case ValidMessages of
                                     [] -> 0;
                                     _  -> {_Guid, TotalSize, Offset} =
                                               lists:last(ValidMessages),
                                           Offset + TotalSize
                                 end};
            [F|_] -> {F, FileSize}
        end,
    ok = gatherer:in(Gatherer, #file_summary {
                       file             = File,
                       valid_total_size = ValidTotalSize,
                       contiguous_top   = ContiguousTop,
                       left             = Left,
                       right            = Right,
                       file_size        = FileSize1,
                       locked           = false,
                       readers          = 0 }),
    ok = gatherer:finish(Gatherer).

%%----------------------------------------------------------------------------
%% garbage collection / compaction / aggregation -- internal
%%----------------------------------------------------------------------------

maybe_roll_to_new_file(
  Offset,
  State = #msstate { dir                 = Dir,
                     current_file_handle = CurHdl,
                     current_file        = CurFile,
                     file_summary_ets    = FileSummaryEts,
                     cur_file_cache_ets  = CurFileCacheEts,
                     file_size_limit     = FileSizeLimit })
  when Offset >= FileSizeLimit ->
    State1 = internal_sync(State),
    ok = file_handle_cache:close(CurHdl),
    NextFile = CurFile + 1,
    {ok, NextHdl} = open_file(Dir, filenum_to_name(NextFile), ?WRITE_MODE),
    true = ets:insert_new(FileSummaryEts, #file_summary {
                            file             = NextFile,
                            valid_total_size = 0,
                            contiguous_top   = 0,
                            left             = CurFile,
                            right            = undefined,
                            file_size        = 0,
                            locked           = false,
                            readers          = 0 }),
    true = ets:update_element(FileSummaryEts, CurFile,
                              {#file_summary.right, NextFile}),
    true = ets:match_delete(CurFileCacheEts, {'_', '_', 0}),
    maybe_compact(State1 #msstate { current_file_handle = NextHdl,
                                    current_file        = NextFile });
maybe_roll_to_new_file(_, State) ->
    State.

maybe_compact(State = #msstate { sum_valid_data   = SumValid,
                                 sum_file_size    = SumFileSize,
                                 gc_active        = false,
                                 gc_pid           = GCPid,
                                 file_summary_ets = FileSummaryEts,
                                 file_size_limit  = FileSizeLimit })
  when (SumFileSize > 2 * FileSizeLimit andalso
        (SumFileSize - SumValid) / SumFileSize > ?GARBAGE_FRACTION) ->
    %% TODO: the algorithm here is sub-optimal - it may result in a
    %% complete traversal of FileSummaryEts.
    case ets:first(FileSummaryEts) of
        '$end_of_table' ->
            State;
        First ->
            case find_files_to_gc(FileSummaryEts, FileSizeLimit,
                                  ets:lookup(FileSummaryEts, First)) of
                not_found ->
                    State;
                {Src, Dst} ->
                    State1 = close_handle(Src, close_handle(Dst, State)),
                    true = ets:update_element(FileSummaryEts, Src,
                                              {#file_summary.locked, true}),
                    true = ets:update_element(FileSummaryEts, Dst,
                                              {#file_summary.locked, true}),
                    ok = rabbit_msg_store_gc:gc(GCPid, Src, Dst),
                    State1 #msstate { gc_active = {Src, Dst} }
            end
    end;
maybe_compact(State) ->
    State.

find_files_to_gc(FileSummaryEts, FileSizeLimit,
                 [#file_summary { file             = Dst,
                                  valid_total_size = DstValid,
                                  right            = Src }]) ->
    case Src of
        undefined ->
            not_found;
        _   ->
            [#file_summary { file             = Src,
                             valid_total_size = SrcValid,
                             left             = Dst,
                             right            = SrcRight }] = Next =
                ets:lookup(FileSummaryEts, Src),
            case SrcRight of
                undefined -> not_found;
                _         -> case DstValid + SrcValid =< FileSizeLimit of
                                 true  -> {Src, Dst};
                                 false -> find_files_to_gc(
                                            FileSummaryEts, FileSizeLimit, Next)
                             end
            end
    end.

delete_file_if_empty(File, State = #msstate { current_file = File }) ->
    State;
delete_file_if_empty(File, State = #msstate {
                             dir              = Dir,
                             sum_file_size    = SumFileSize,
                             file_handles_ets = FileHandlesEts,
                             file_summary_ets = FileSummaryEts }) ->
    [#file_summary { valid_total_size = ValidData,
                     left             = Left,
                     right            = Right,
                     file_size        = FileSize,
                     locked           = false }] =
        ets:lookup(FileSummaryEts, File),
    case ValidData of
        %% we should NEVER find the current file in here hence right
        %% should always be a file, not undefined
        0 -> case {Left, Right} of
                 {undefined, _} when Right =/= undefined ->
                     %% the eldest file is empty.
                     true = ets:update_element(
                              FileSummaryEts, Right,
                              {#file_summary.left, undefined});
                 {_, _} when Right =/= undefined ->
                     true = ets:update_element(FileSummaryEts, Right,
                                               {#file_summary.left, Left}),
                     true = ets:update_element(FileSummaryEts, Left,
                                               {#file_summary.right, Right})
             end,
             true = mark_handle_to_close(FileHandlesEts, File),
             true = ets:delete(FileSummaryEts, File),
             State1 = close_handle(File, State),
             ok = file:delete(form_filename(Dir, filenum_to_name(File))),
             State1 #msstate { sum_file_size = SumFileSize - FileSize };
        _ -> State
    end.

%%----------------------------------------------------------------------------
%% garbage collection / compaction / aggregation -- external
%%----------------------------------------------------------------------------

gc(SrcFile, DstFile, State = {FileSummaryEts, _Dir, _Index, _IndexState}) ->
    [SrcObj = #file_summary {
       readers          = SrcReaders,
       left             = DstFile,
       file_size        = SrcFileSize,
       locked           = true }] = ets:lookup(FileSummaryEts, SrcFile),
    [DstObj = #file_summary {
       readers          = DstReaders,
       right            = SrcFile,
       file_size        = DstFileSize,
       locked           = true }] = ets:lookup(FileSummaryEts, DstFile),

    case SrcReaders =:= 0 andalso DstReaders =:= 0 of
        true  -> TotalValidData = combine_files(SrcObj, DstObj, State),
                 %% don't update dest.right, because it could be
                 %% changing at the same time
                 true = ets:update_element(
                          FileSummaryEts, DstFile,
                          [{#file_summary.valid_total_size, TotalValidData},
                           {#file_summary.contiguous_top,   TotalValidData},
                           {#file_summary.file_size,        TotalValidData}]),
                 SrcFileSize + DstFileSize - TotalValidData;
        false -> concurrent_readers
    end.

combine_files(#file_summary { file             = Source,
                              valid_total_size = SourceValid,
                              left             = Destination },
              #file_summary { file             = Destination,
                              valid_total_size = DestinationValid,
                              contiguous_top   = DestinationContiguousTop,
                              right            = Source },
              State = {_FileSummaryEts, Dir, _Index, _IndexState}) ->
    SourceName      = filenum_to_name(Source),
    DestinationName = filenum_to_name(Destination),
    {ok, SourceHdl}      = open_file(Dir, SourceName,
                                     ?READ_AHEAD_MODE),
    {ok, DestinationHdl} = open_file(Dir, DestinationName,
                                     ?READ_AHEAD_MODE ++ ?WRITE_MODE),
    ExpectedSize = SourceValid + DestinationValid,
    %% if DestinationValid =:= DestinationContiguousTop then we don't
    %% need a tmp file
    %% if they're not equal, then we need to write out everything past
    %%   the DestinationContiguousTop to a tmp file then truncate,
    %%   copy back in, and then copy over from Source
    %% otherwise we just truncate straight away and copy over from Source
    case DestinationContiguousTop =:= DestinationValid of
        true ->
            ok = truncate_and_extend_file(
                   DestinationHdl, DestinationContiguousTop, ExpectedSize);
        false ->
            {DestinationWorkList, DestinationValid} =
                find_unremoved_messages_in_file(Destination, State),
            Worklist =
                lists:dropwhile(
                  fun (#msg_location { offset = Offset })
                      when Offset =/= DestinationContiguousTop ->
                          %% it cannot be that Offset =:=
                          %% DestinationContiguousTop because if it
                          %% was then DestinationContiguousTop would
                          %% have been extended by TotalSize
                          Offset < DestinationContiguousTop
                  end, DestinationWorkList),
            Tmp = filename:rootname(DestinationName) ++ ?FILE_EXTENSION_TMP,
            {ok, TmpHdl} = open_file(Dir, Tmp, ?READ_AHEAD_MODE ++ ?WRITE_MODE),
            ok = copy_messages(
                   Worklist, DestinationContiguousTop, DestinationValid,
                   DestinationHdl, TmpHdl, Destination, State),
            TmpSize = DestinationValid - DestinationContiguousTop,
            %% so now Tmp contains everything we need to salvage from
            %% Destination, and index_state has been updated to
            %% reflect the compaction of Destination so truncate
            %% Destination and copy from Tmp back to the end
            {ok, 0} = file_handle_cache:position(TmpHdl, 0),
            ok = truncate_and_extend_file(
                   DestinationHdl, DestinationContiguousTop, ExpectedSize),
            {ok, TmpSize} =
                file_handle_cache:copy(TmpHdl, DestinationHdl, TmpSize),
            %% position in DestinationHdl should now be DestinationValid
            ok = file_handle_cache:sync(DestinationHdl),
            ok = file_handle_cache:delete(TmpHdl)
    end,
    {SourceWorkList, SourceValid} =
        find_unremoved_messages_in_file(Source, State),
    ok = copy_messages(SourceWorkList, DestinationValid, ExpectedSize,
                       SourceHdl, DestinationHdl, Destination, State),
    %% tidy up
    ok = file_handle_cache:close(DestinationHdl),
    ok = file_handle_cache:delete(SourceHdl),
    ExpectedSize.

find_unremoved_messages_in_file(File,
                                {_FileSummaryEts, Dir, Index, IndexState}) ->
    %% Messages here will be end-of-file at start-of-list
    {ok, Messages, _FileSize} =
        scan_file_for_valid_messages(Dir, filenum_to_name(File)),
    %% foldl will reverse so will end up with msgs in ascending offset order
    lists:foldl(fun ({Guid, TotalSize, Offset}, Acc = {List, Size}) ->
                        case Index:lookup(Guid, IndexState) of
                            #msg_location { file = File, total_size = TotalSize,
                                            offset = Offset } = Entry ->
                                {[ Entry | List ], TotalSize + Size};
                            _ ->
                                Acc
                        end
                end, {[], 0}, Messages).

copy_messages(WorkList, InitOffset, FinalOffset, SourceHdl, DestinationHdl,
              Destination, {_FileSummaryEts, _Dir, Index, IndexState}) ->
    Copy = fun ({BlockStart, BlockEnd}) ->
                   BSize = BlockEnd - BlockStart,
                   {ok, BlockStart} =
                       file_handle_cache:position(SourceHdl, BlockStart),
                   {ok, BSize} =
                       file_handle_cache:copy(SourceHdl, DestinationHdl, BSize)
           end,
    case
        lists:foldl(
          fun (#msg_location { guid = Guid, offset = Offset,
                               total_size = TotalSize },
               {CurOffset, Block = {BlockStart, BlockEnd}}) ->
                  %% CurOffset is in the DestinationFile.
                  %% Offset, BlockStart and BlockEnd are in the SourceFile
                  %% update MsgLocation to reflect change of file and offset
                  ok = Index:update_fields(Guid,
                                           [{#msg_location.file, Destination},
                                            {#msg_location.offset, CurOffset}],
                                           IndexState),
                  {CurOffset + TotalSize,
                   case BlockEnd of
                       undefined ->
                           %% base case, called only for the first list elem
                           {Offset, Offset + TotalSize};
                       Offset ->
                           %% extend the current block because the
                           %% next msg follows straight on
                           {BlockStart, BlockEnd + TotalSize};
                       _ ->
                           %% found a gap, so actually do the work for
                           %% the previous block
                           Copy(Block),
                           {Offset, Offset + TotalSize}
                   end}
          end, {InitOffset, {undefined, undefined}}, WorkList) of
        {FinalOffset, Block} ->
            case WorkList of
                [] -> ok;
                _  -> Copy(Block), %% do the last remaining block
                      ok = file_handle_cache:sync(DestinationHdl)
            end;
        {FinalOffsetZ, _Block} ->
            {gc_error, [{expected, FinalOffset},
                        {got, FinalOffsetZ},
                        {destination, Destination}]}
    end.