diff options
| -rw-r--r-- | src/rabbit_msg_store.erl | 262 | ||||
| -rw-r--r-- | src/rabbit_msg_store_gc.erl | 15 | ||||
| -rw-r--r-- | src/rabbit_tests.erl | 20 | ||||
| -rw-r--r-- | src/rabbit_variable_queue.erl | 32 |
4 files changed, 198 insertions, 131 deletions
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl index f8b41ed388..fe5cbab89b 100644 --- a/src/rabbit_msg_store.erl +++ b/src/rabbit_msg_store.erl @@ -34,8 +34,8 @@ -behaviour(gen_server2). -export([start_link/4, successfully_recovered_state/1, - client_init/3, client_terminate/1, client_delete_and_terminate/1, - client_ref/1, + client_init/4, client_terminate/1, client_delete_and_terminate/1, + client_ref/1, close_all_indicated/1, write/3, read/2, contains/2, remove/2, release/2, sync/3]). -export([sync/1, set_maximum_since_use/2, @@ -82,10 +82,10 @@ dedup_cache_ets, %% tid of dedup cache table cur_file_cache_ets, %% tid of current file cache table dying_clients, %% set of dying clients - client_refs, %% set of references of all registered clients + clients, %% map of references of all registered clients + %% to callbacks successfully_recovered, %% boolean: did we recover state? file_size_limit, %% how big are our files allowed to get? - client_ondisk_callback, %% client ref to callback function mapping cref_to_guids %% client ref to synced messages mapping }). @@ -111,6 +111,7 @@ index_module, index_state, file_summary_ets, + file_handles_ets, msg_store }). @@ -124,6 +125,7 @@ index_module :: atom(), index_state :: any(), file_summary_ets :: ets:tid(), + file_handles_ets :: ets:tid(), msg_store :: server() }). @@ -146,13 +148,15 @@ {(fun ((A) -> 'finished' | {rabbit_guid:guid(), non_neg_integer(), A})), A}). -type(maybe_guid_fun() :: 'undefined' | fun ((gb_set()) -> any())). +-type(maybe_close_fds_fun() :: 'undefined' | fun (() -> 'ok')). +-type(deletion_thunk() :: fun (() -> boolean())). -spec(start_link/4 :: (atom(), file:filename(), [binary()] | 'undefined', startup_fun_state()) -> rabbit_types:ok_pid_or_error()). -spec(successfully_recovered_state/1 :: (server()) -> boolean()). --spec(client_init/3 :: (server(), client_ref(), maybe_guid_fun()) -> - client_msstate()). +-spec(client_init/4 :: (server(), client_ref(), maybe_guid_fun(), + maybe_close_fds_fun()) -> client_msstate()). -spec(client_terminate/1 :: (client_msstate()) -> 'ok'). -spec(client_delete_and_terminate/1 :: (client_msstate()) -> 'ok'). -spec(client_ref/1 :: (client_msstate()) -> client_ref()). @@ -169,8 +173,9 @@ -spec(set_maximum_since_use/2 :: (server(), non_neg_integer()) -> 'ok'). -spec(has_readers/2 :: (non_neg_integer(), gc_state()) -> boolean()). -spec(combine_files/3 :: (non_neg_integer(), non_neg_integer(), gc_state()) -> - 'ok'). --spec(delete_file/2 :: (non_neg_integer(), gc_state()) -> 'ok'). + 'ok' | deletion_thunk()). +-spec(delete_file/2 :: (non_neg_integer(), gc_state()) -> + 'ok' | deletion_thunk()). -endif. @@ -380,9 +385,9 @@ %% %% We use a separate set to keep track of the dying clients in order %% to keep that set, which is inspected on every write and remove, as -%% small as possible. Inspecting client_refs - the set of all clients -%% - would degrade performance with many healthy clients and few, if -%% any, dying clients, which is the typical case. +%% small as possible. Inspecting the set of all clients would degrade +%% performance with many healthy clients and few, if any, dying +%% clients, which is the typical case. %% %% For notes on Clean Shutdown and startup, see documentation in %% variable_queue. @@ -399,11 +404,11 @@ start_link(Server, Dir, ClientRefs, StartupFunState) -> successfully_recovered_state(Server) -> gen_server2:call(Server, successfully_recovered_state, infinity). -client_init(Server, Ref, MsgOnDiskFun) -> +client_init(Server, Ref, MsgOnDiskFun, CloseFDsFun) -> {IState, IModule, Dir, GCPid, FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts} = - gen_server2:call(Server, {new_client_state, Ref, MsgOnDiskFun}, - infinity), + gen_server2:call( + Server, {new_client_state, Ref, MsgOnDiskFun, CloseFDsFun}, infinity), #client_msstate { server = Server, client_ref = Ref, file_handle_cache = dict:new(), @@ -523,7 +528,8 @@ client_read3(#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 }) -> + gc_pid = GCPid, + client_ref = Ref }) -> Release = fun() -> ok = case ets:update_counter(FileSummaryEts, File, {#file_summary.readers, -1}) of @@ -570,9 +576,14 @@ client_read3(#msg_location { guid = Guid, file = File }, Defer, 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), + {ok, CState1} = close_all_indicated(CState), + %% We are now guaranteed that the mark_handle_open + %% call will either insert_new correctly, or will + %% fail, but find the value is open, not close. + mark_handle_open(FileHandlesEts, File, Ref), + %% Could the msg_store now mark the file to be + %% closed? No: marks for closing are issued only + %% when the msg_store has locked the file. {Msg, CState2} = %% This will never be the current file read_from_disk(MsgLocation, CState1, DedupCacheEts), Release(), %% this MUST NOT fail with badarg @@ -588,11 +599,10 @@ client_read3(#msg_location { guid = Guid, file = File }, Defer, end end. -clear_client_callback(CRef, - State = #msstate { client_ondisk_callback = CODC, - cref_to_guids = CTG }) -> - State #msstate { client_ondisk_callback = dict:erase(CRef, CODC), - cref_to_guids = dict:erase(CRef, CTG)}. +clear_client(CRef, State = #msstate { cref_to_guids = CTG, + dying_clients = DyingClients }) -> + State #msstate { cref_to_guids = dict:erase(CRef, CTG), + dying_clients = sets:del_element(CRef, DyingClients) }. %%---------------------------------------------------------------------------- @@ -625,7 +635,7 @@ init([Server, BaseDir, ClientRefs, StartupFunState]) -> {FileSummaryRecovered, FileSummaryEts} = recover_file_summary(AttemptFileSummaryRecovery, Dir), - {CleanShutdown, IndexState, ClientRefs1} = + {CleanShutdown, IndexState, Clients1} = recover_index_and_client_refs(IndexModule, FileSummaryRecovered, ClientRefs, Dir, Server), %% CleanShutdown => msg location index and file_summary both @@ -661,10 +671,9 @@ init([Server, BaseDir, ClientRefs, StartupFunState]) -> dedup_cache_ets = DedupCacheEts, cur_file_cache_ets = CurFileCacheEts, dying_clients = sets:new(), - client_refs = ClientRefs1, + clients = Clients1, successfully_recovered = CleanShutdown, file_size_limit = FileSizeLimit, - client_ondisk_callback = dict:new(), cref_to_guids = dict:new() }, @@ -684,6 +693,7 @@ init([Server, BaseDir, ClientRefs, StartupFunState]) -> index_module = IndexModule, index_state = IndexState, file_summary_ets = FileSummaryEts, + file_handles_ets = FileHandlesEts, msg_store = self() }), @@ -694,10 +704,10 @@ init([Server, BaseDir, ClientRefs, StartupFunState]) -> prioritise_call(Msg, _From, _State) -> case Msg of - successfully_recovered_state -> 7; - {new_client_state, _Ref, _MODC} -> 7; - {read, _Guid} -> 2; - _ -> 0 + successfully_recovered_state -> 7; + {new_client_state, _Ref, _MODC, _CloseFDsFun} -> 7; + {read, _Guid} -> 2; + _ -> 0 end. prioritise_cast(Msg, _State) -> @@ -713,7 +723,7 @@ prioritise_cast(Msg, _State) -> handle_call(successfully_recovered_state, _From, State) -> reply(State #msstate.successfully_recovered, State); -handle_call({new_client_state, CRef, Callback}, _From, +handle_call({new_client_state, CRef, MsgOnDiskFun, CloseFDsFun}, _From, State = #msstate { dir = Dir, index_state = IndexState, index_module = IndexModule, @@ -721,21 +731,15 @@ handle_call({new_client_state, CRef, Callback}, _From, file_summary_ets = FileSummaryEts, dedup_cache_ets = DedupCacheEts, cur_file_cache_ets = CurFileCacheEts, - client_refs = ClientRefs, - client_ondisk_callback = CODC, + clients = Clients, gc_pid = GCPid }) -> - CODC1 = case Callback of - undefined -> CODC; - _ -> dict:store(CRef, Callback, CODC) - end, + Clients1 = dict:store(CRef, {MsgOnDiskFun, CloseFDsFun}, Clients), reply({IndexState, IndexModule, Dir, GCPid, FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts}, - State #msstate { client_refs = sets:add_element(CRef, ClientRefs), - client_ondisk_callback = CODC1 }); + State #msstate { clients = Clients1 }); -handle_call({client_terminate, CRef}, _From, - State) -> - reply(ok, clear_client_callback(CRef, State)); +handle_call({client_terminate, CRef}, _From, State) -> + reply(ok, clear_client(CRef, State)); handle_call({read, Guid}, From, State) -> State1 = read_message(Guid, From, State), @@ -750,24 +754,19 @@ handle_cast({client_dying, CRef}, DyingClients1 = sets:add_element(CRef, DyingClients), write_message(CRef, <<>>, State #msstate { dying_clients = DyingClients1 }); -handle_cast({client_delete, CRef}, - State = #msstate { client_refs = ClientRefs, - dying_clients = DyingClients }) -> - State1 = clear_client_callback( - CRef, State #msstate { - client_refs = sets:del_element(CRef, ClientRefs), - dying_clients = sets:del_element(CRef, DyingClients) }), - noreply(remove_message(CRef, CRef, State1)); +handle_cast({client_delete, CRef}, State = #msstate { clients = Clients }) -> + State1 = State #msstate { clients = dict:erase(CRef, Clients) }, + noreply(remove_message(CRef, CRef, clear_client(CRef, State1))); handle_cast({write, CRef, Guid}, - State = #msstate { file_summary_ets = FileSummaryEts, - cur_file_cache_ets = CurFileCacheEts, - client_ondisk_callback = CODC, - cref_to_guids = CTG }) -> + State = #msstate { file_summary_ets = FileSummaryEts, + cur_file_cache_ets = CurFileCacheEts, + clients = Clients, + cref_to_guids = CTG }) -> true = 0 =< ets:update_counter(CurFileCacheEts, Guid, {3, -1}), [{Guid, Msg, _CacheRefCount}] = ets:lookup(CurFileCacheEts, Guid), - CTG1 = add_cref_to_guids_if_callback(CRef, Guid, CTG, CODC), + CTG1 = add_cref_to_guids_if_callback(CRef, Guid, CTG, Clients), State1 = State #msstate { cref_to_guids = CTG1 }, case should_mask_action(CRef, Guid, State) of {true, _Location} -> @@ -832,10 +831,11 @@ handle_cast(sync, State) -> handle_cast({combine_files, Source, Destination, Reclaimed}, State = #msstate { sum_file_size = SumFileSize, file_handles_ets = FileHandlesEts, - file_summary_ets = FileSummaryEts }) -> + file_summary_ets = FileSummaryEts, + clients = Clients }) -> ok = cleanup_after_file_deletion(Source, State), - %% see comment in cleanup_after_file_deletion - true = mark_handle_to_close(FileHandlesEts, Destination), + %% see comment in cleanup_after_file_deletion, and client_read3 + true = mark_handle_to_close(Clients, FileHandlesEts, Destination, false), true = ets:update_element(FileSummaryEts, Destination, {#file_summary.locked, false}), State1 = State #msstate { sum_file_size = SumFileSize - Reclaimed }, @@ -865,7 +865,7 @@ terminate(_Reason, State = #msstate { index_state = IndexState, file_summary_ets = FileSummaryEts, dedup_cache_ets = DedupCacheEts, cur_file_cache_ets = CurFileCacheEts, - client_refs = ClientRefs, + clients = Clients, dir = Dir }) -> %% stop the gc first, otherwise it could be working and we pull %% out the ets tables from under it. @@ -881,7 +881,7 @@ terminate(_Reason, State = #msstate { index_state = IndexState, [ets:delete(T) || T <- [FileSummaryEts, DedupCacheEts, FileHandlesEts, CurFileCacheEts]], IndexModule:terminate(IndexState), - store_recovery_terms([{client_refs, sets:to_list(ClientRefs)}, + store_recovery_terms([{client_refs, dict:fetch_keys(Clients)}, {index_module, IndexModule}], Dir), State3 #msstate { index_state = undefined, current_file_handle = undefined }. @@ -1136,41 +1136,45 @@ orddict_store(Key, Val, Dict) -> orddict:store(Key, Val, Dict). client_confirm(CRef, Guids, ActionTaken, - State = #msstate { client_ondisk_callback = CODC, - cref_to_guids = CTG }) -> - case dict:find(CRef, CODC) of - {ok, Fun} -> Fun(Guids, ActionTaken), - CTG1 = case dict:find(CRef, CTG) of - {ok, Gs} -> - Guids1 = gb_sets:difference(Gs, Guids), - case gb_sets:is_empty(Guids1) of - true -> dict:erase(CRef, CTG); - false -> dict:store(CRef, Guids1, CTG) - end; - error -> CTG - end, - State #msstate { cref_to_guids = CTG1 }; - error -> State + State = #msstate { clients = Clients, + cref_to_guids = CTG }) -> + case dict:fetch(CRef, Clients) of + {undefined, _CloseFDsFun} -> + State; + {MsgOnDiskFun, _CloseFDsFun} -> + MsgOnDiskFun(Guids, ActionTaken), + CTG1 = case dict:find(CRef, CTG) of + {ok, Gs} -> Guids1 = gb_sets:difference(Gs, Guids), + case gb_sets:is_empty(Guids1) of + true -> dict:erase(CRef, CTG); + false -> dict:store(CRef, Guids1, CTG) + end; + error -> CTG + end, + State #msstate { cref_to_guids = CTG1 } end. -add_cref_to_guids_if_callback(CRef, Guid, CTG, CODC) -> - case dict:find(CRef, CODC) of - {ok, _} -> dict:update(CRef, - fun (Guids) -> gb_sets:add(Guid, Guids) end, - gb_sets:singleton(Guid), CTG); - error -> CTG +add_cref_to_guids_if_callback(CRef, Guid, CTG, Clients) -> + case dict:fetch(CRef, Clients) of + {undefined, _CloseFDsFun} -> + CTG; + {_MsgOnDiskFun, _CloseFDsFun} -> + dict:update(CRef, fun (Guids) -> gb_sets:add(Guid, Guids) end, + gb_sets:singleton(Guid), CTG) end. client_confirm_if_on_disk(CRef, Guid, File, - State = #msstate { client_ondisk_callback = CODC, + State = #msstate { clients = Clients, current_file = CurFile, cref_to_guids = CTG }) -> CTG1 = case File of - CurFile -> add_cref_to_guids_if_callback(CRef, Guid, CTG, CODC); - _ -> case dict:find(CRef, CODC) of - {ok, Fun} -> Fun(gb_sets:singleton(Guid), written); - _ -> ok + CurFile -> add_cref_to_guids_if_callback(CRef, Guid, CTG, Clients); + _ -> case dict:fetch(CRef, Clients) of + {undefined, _CloseFDsFun} -> + ok; + {MsgOnDiskFun, _CloseFDsFun} -> + MsgOnDiskFun(gb_sets:singleton(Guid), written) end, CTG end, @@ -1221,29 +1225,56 @@ close_handle(Key, FHC) -> error -> FHC end. -mark_handle_open(FileHandlesEts, File) -> - %% This is fine to fail (already exists) - ets:insert_new(FileHandlesEts, {{self(), File}, open}), +mark_handle_open(FileHandlesEts, File, Ref) -> + %% This is fine to fail (already exists). Note it could fail with + %% the value being close, and not have it updated to open. + ets:insert_new(FileHandlesEts, {{Ref, File}, open}), true. -mark_handle_to_close(FileHandlesEts, File) -> - [ ets:update_element(FileHandlesEts, Key, {2, close}) - || {Key, open} <- ets:match_object(FileHandlesEts, {{'_', File}, open}) ], +%% See comment in client_read3 - only call this when the file is locked +mark_handle_to_close(ClientRefs, FileHandlesEts, File, Invoke) -> + [ begin + case ets:update_element(FileHandlesEts, Key, {2, close}) + andalso Invoke of + true -> case dict:fetch(Ref, ClientRefs) of + {_MsgOnDiskFun, undefined} -> ok; + {_MsgOnDiskFun, CloseFDsFun} -> ok = CloseFDsFun() + end; + false -> ok + end + end || + {{Ref, _File} = Key, open} <- + ets:match_object(FileHandlesEts, {{'_', File}, open}) ], true. -close_all_indicated(#client_msstate { file_handles_ets = FileHandlesEts } = +safe_file_delete_fun(File, Dir, FileHandlesEts) -> + fun () -> safe_file_delete(File, Dir, FileHandlesEts) end. + +safe_file_delete(File, Dir, FileHandlesEts) -> + %% do not match on any value - it's the absence of the row that + %% indicates the client has really closed the file. + case ets:match_object(FileHandlesEts, {{'_', File}, '_'}, 1) of + {[_|_], _Cont} -> + false; + _ -> + ok = file:delete(form_filename(Dir, filenum_to_name(File))), + true + end. + +close_all_indicated(#client_msstate { file_handles_ets = FileHandlesEts, + client_ref = Ref } = 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(), + Objs = ets:match_object(FileHandlesEts, {{Ref, '_'}, close}), + {ok, lists:foldl(fun ({Key = {_Ref, 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, + client_ref = Ref }) -> ok = dict:fold(fun (File, Hdl, ok) -> - true = ets:delete(FileHandlesEts, {Self, File}), + true = ets:delete(FileHandlesEts, {Ref, File}), file_handle_cache:close(Hdl) end, ok, FHC), CState #client_msstate { file_handle_cache = dict:new() }; @@ -1381,16 +1412,16 @@ index_delete_by_file(File, #msstate { index_module = Index, %%---------------------------------------------------------------------------- recover_index_and_client_refs(IndexModule, _Recover, undefined, Dir, _Server) -> - {false, IndexModule:new(Dir), sets:new()}; + {false, IndexModule:new(Dir), dict: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()}; + {false, IndexModule:new(Dir), dict: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()} + {false, IndexModule:new(Dir), dict:new()} end, case read_recovery_terms(Dir) of {false, Error} -> @@ -1403,7 +1434,8 @@ recover_index_and_client_refs(IndexModule, true, ClientRefs, Dir, Server) -> true -> case IndexModule:recover(Dir) of {ok, IndexState1} -> {true, IndexState1, - sets:from_list(ClientRefs)}; + dict:from_list([{CRef, {undefined, undefined}} + || CRef <- ClientRefs])}; {error, Error} -> Fresh("failed to recover index: ~p", [Error]) end; @@ -1744,7 +1776,8 @@ delete_file_if_empty(File, State = #msstate { cleanup_after_file_deletion(File, #msstate { file_handles_ets = FileHandlesEts, - file_summary_ets = FileSummaryEts }) -> + file_summary_ets = FileSummaryEts, + clients = Clients }) -> %% Ensure that any clients that have open fhs to the file 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 @@ -1752,7 +1785,7 @@ cleanup_after_file_deletion(File, %% 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, File), + true = mark_handle_to_close(Clients, FileHandlesEts, File, true), [#file_summary { left = Left, right = Right, locked = true, @@ -1781,6 +1814,7 @@ has_readers(File, #gc_state { file_summary_ets = FileSummaryEts }) -> combine_files(Source, Destination, State = #gc_state { file_summary_ets = FileSummaryEts, + file_handles_ets = FileHandlesEts, dir = Dir, msg_store = Server }) -> [#file_summary { @@ -1841,7 +1875,7 @@ combine_files(Source, Destination, SourceHdl, DestinationHdl, Destination, State), %% tidy up ok = file_handle_cache:close(DestinationHdl), - ok = file_handle_cache:delete(SourceHdl), + ok = file_handle_cache:close(SourceHdl), %% don't update dest.right, because it could be changing at the %% same time @@ -1851,9 +1885,11 @@ combine_files(Source, Destination, {#file_summary.file_size, TotalValidData}]), Reclaimed = SourceFileSize + DestinationFileSize - TotalValidData, - gen_server2:cast(Server, {combine_files, Source, Destination, Reclaimed}). + gen_server2:cast(Server, {combine_files, Source, Destination, Reclaimed}), + safe_file_delete_fun(Source, Dir, FileHandlesEts). delete_file(File, State = #gc_state { file_summary_ets = FileSummaryEts, + file_handles_ets = FileHandlesEts, dir = Dir, msg_store = Server }) -> [#file_summary { valid_total_size = 0, @@ -1861,8 +1897,8 @@ delete_file(File, State = #gc_state { file_summary_ets = FileSummaryEts, file_size = FileSize, readers = 0 }] = ets:lookup(FileSummaryEts, File), {[], 0} = load_and_vacuum_message_file(File, State), - ok = file:delete(form_filename(Dir, filenum_to_name(File))), - gen_server2:cast(Server, {delete_file, File, FileSize}). + gen_server2:cast(Server, {delete_file, File, FileSize}), + safe_file_delete_fun(File, Dir, FileHandlesEts). load_and_vacuum_message_file(File, #gc_state { dir = Dir, index_module = Index, diff --git a/src/rabbit_msg_store_gc.erl b/src/rabbit_msg_store_gc.erl index cd9fd4973f..68bf43380e 100644 --- a/src/rabbit_msg_store_gc.erl +++ b/src/rabbit_msg_store_gc.erl @@ -42,6 +42,7 @@ -record(state, { pending_no_readers, + on_action, msg_store_state }). @@ -89,6 +90,7 @@ init([MsgStoreState]) -> ok = file_handle_cache:register_callback(?MODULE, set_maximum_since_use, [self()]), {ok, #state { pending_no_readers = dict:new(), + on_action = [], msg_store_state = MsgStoreState }, hibernate, {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. @@ -131,13 +133,18 @@ code_change(_OldVsn, State, _Extra) -> attempt_action(Action, Files, State = #state { pending_no_readers = Pending, + on_action = Thunks, msg_store_state = MsgStoreState }) -> case [File || File <- Files, rabbit_msg_store:has_readers(File, MsgStoreState)] of - [] -> do_action(Action, Files, MsgStoreState), - State; - [File | _] -> Pending1 = dict:store(File, {Action, Files}, Pending), - State #state { pending_no_readers = Pending1 } + [] -> + Thunks1 = lists:filter( + fun (Thunk) -> not Thunk() end, + [do_action(Action, Files, MsgStoreState) | Thunks]), + State #state { on_action = Thunks1 }; + [File | _] -> + Pending1 = dict:store(File, {Action, Files}, Pending), + State #state { pending_no_readers = Pending1 } end. do_action(combine, [Source, Destination], MsgStoreState) -> diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index d913092cce..befcfd99ad 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -1502,12 +1502,13 @@ msg_store_remove(MsgStore, Ref, Guids) -> with_msg_store_client(MsgStore, Ref, Fun) -> rabbit_msg_store:client_terminate( - Fun(rabbit_msg_store:client_init(MsgStore, Ref, undefined))). + Fun(rabbit_msg_store:client_init(MsgStore, Ref, undefined, undefined))). foreach_with_msg_store_client(MsgStore, Ref, Fun, L) -> rabbit_msg_store:client_terminate( - lists:foldl(fun (Guid, MSCState) -> Fun(Guid, MSCState) end, - rabbit_msg_store:client_init(MsgStore, Ref, undefined), L)). + lists:foldl( + fun (Guid, MSCState) -> Fun(Guid, MSCState) end, + rabbit_msg_store:client_init(MsgStore, Ref, undefined, undefined), L)). test_msg_store() -> restart_msg_store_empty(), @@ -1516,7 +1517,7 @@ test_msg_store() -> {Guids1stHalf, Guids2ndHalf} = lists:split(50, Guids), Ref = rabbit_guid:guid(), MSCState = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref, - undefined), + undefined, undefined), %% check we don't contain any of the msgs we're about to publish false = msg_store_contains(false, Guids, MSCState), %% publish the first half @@ -1583,7 +1584,7 @@ test_msg_store() -> {Guid, 0, GuidsTail} end, Guids2ndHalf}), MSCState5 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref, - undefined), + undefined, undefined), %% check we have the right msgs left lists:foldl( fun (Guid, Bool) -> @@ -1593,7 +1594,7 @@ test_msg_store() -> %% restart empty restart_msg_store_empty(), MSCState6 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref, - undefined), + undefined, undefined), %% check we don't contain any of the msgs false = msg_store_contains(false, Guids, MSCState6), %% publish the first half again @@ -1602,7 +1603,7 @@ test_msg_store() -> ok = rabbit_msg_store:client_terminate( msg_store_read(Guids1stHalf, MSCState6)), MSCState7 = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, Ref, - undefined), + undefined, undefined), ok = rabbit_msg_store:remove(Guids1stHalf, MSCState7), ok = rabbit_msg_store:client_terminate(MSCState7), %% restart empty @@ -1661,7 +1662,7 @@ init_test_queue() -> Terms = rabbit_queue_index:shutdown_terms(TestQueue), PRef = proplists:get_value(persistent_ref, Terms, rabbit_guid:guid()), PersistentClient = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, - PRef, undefined), + PRef, undefined, undefined), Res = rabbit_queue_index:recover( TestQueue, Terms, false, fun (Guid) -> @@ -1695,7 +1696,8 @@ queue_index_publish(SeqIds, Persistent, Qi) -> true -> ?PERSISTENT_MSG_STORE; false -> ?TRANSIENT_MSG_STORE end, - MSCState = rabbit_msg_store:client_init(MsgStore, Ref, undefined), + MSCState = + rabbit_msg_store:client_init(MsgStore, Ref, undefined, undefined), {A, B = [{_SeqId, LastGuidWritten} | _]} = lists:foldl( fun (SeqId, {QiN, SeqIdsGuidsAcc}) -> diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl index 35e37df61e..9f02b6b76e 100644 --- a/src/rabbit_variable_queue.erl +++ b/src/rabbit_variable_queue.erl @@ -436,10 +436,10 @@ init(QueueName, true, true, MsgOnDiskFun, MsgIdxOnDiskFun) -> Terms}; _ -> {rabbit_guid:guid(), rabbit_guid:guid(), []} end, - PersistentClient = rabbit_msg_store:client_init(?PERSISTENT_MSG_STORE, - PRef, MsgOnDiskFun), - TransientClient = rabbit_msg_store:client_init(?TRANSIENT_MSG_STORE, - TRef, undefined), + PersistentClient = msg_store_client_init( + ?PERSISTENT_MSG_STORE, PRef, MsgOnDiskFun), + TransientClient = msg_store_client_init( + ?TRANSIENT_MSG_STORE, TRef, undefined), {DeltaCount, IndexState} = rabbit_queue_index:recover( QueueName, Terms1, @@ -937,7 +937,12 @@ with_immutable_msg_store_state(MSCState, IsPersistent, Fun) -> Res. msg_store_client_init(MsgStore, MsgOnDiskFun) -> - rabbit_msg_store:client_init(MsgStore, rabbit_guid:guid(), MsgOnDiskFun). + msg_store_client_init(MsgStore, rabbit_guid:guid(), MsgOnDiskFun). + +msg_store_client_init(MsgStore, Ref, MsgOnDiskFun) -> + rabbit_msg_store:client_init( + MsgStore, Ref, MsgOnDiskFun, + msg_store_close_fds_fun(MsgStore =:= ?PERSISTENT_MSG_STORE)). msg_store_write(MSCState, IsPersistent, Guid, Msg) -> with_immutable_msg_store_state( @@ -964,6 +969,23 @@ msg_store_sync(MSCState, IsPersistent, Guids, Callback) -> MSCState, IsPersistent, fun (MSCState1) -> rabbit_msg_store:sync(Guids, Callback, MSCState1) end). +msg_store_close_fds(MSCState, IsPersistent) -> + with_msg_store_state( + MSCState, IsPersistent, + fun (MSCState1) -> rabbit_msg_store:close_all_indicated(MSCState1) end). + +msg_store_close_fds_fun(IsPersistent) -> + Self = self(), + fun () -> + rabbit_amqqueue:maybe_run_queue_via_backing_queue_async( + Self, + fun (State = #vqstate { msg_store_clients = MSCState }) -> + {ok, MSCState1} = + msg_store_close_fds(MSCState, IsPersistent), + {[], State #vqstate { msg_store_clients = MSCState1 }} + end) + end. + maybe_write_delivered(false, _SeqId, IndexState) -> IndexState; maybe_write_delivered(true, SeqId, IndexState) -> |