From 5b54affbf0ee9b1e0f3bb488d315ff1221a275d4 Mon Sep 17 00:00:00 2001 From: Martin Sumner Date: Fri, 25 Jan 2019 14:32:41 +0000 Subject: [PATCH] Have inker reopen compacted files The inker cler will now close compacted files before prompting the inker to update the manifest. The inker should reopen those files, so that the file processes are linked to it and not the clerk. This also stops a stopped clerk leading to orphaned cdb files. --- src/leveled_iclerk.erl | 28 +++++++++++++++++----------- src/leveled_imanifest.erl | 3 +++ src/leveled_inker.erl | 10 ++++++++-- 3 files changed, 28 insertions(+), 13 deletions(-) diff --git a/src/leveled_iclerk.erl b/src/leveled_iclerk.erl index 0e2b2ae..076d73b 100644 --- a/src/leveled_iclerk.erl +++ b/src/leveled_iclerk.erl @@ -227,6 +227,7 @@ clerk_scorefilelist(Pid, []) -> clerk_scorefilelist(Pid, CandidateList) -> gen_server:cast(Pid, {score_filelist, CandidateList}). + %%%============================================================================ %%% gen_server callbacks %%%============================================================================ @@ -357,14 +358,14 @@ handle_cast(scoring_complete, State) -> end, BestRun1), leveled_log:log("IC002", [length(FilesToDelete)]), + ok = CloseFun(FilterServer), ok = leveled_inker:ink_clerkcomplete(State#state.inker, ManifestSlice, FilesToDelete), - ok = CloseFun(FilterServer), {noreply, State#state{scoring_state = undefined}}; false -> - ok = leveled_inker:ink_clerkcomplete(State#state.inker, [], []), ok = CloseFun(FilterServer), + ok = leveled_inker:ink_clerkcomplete(State#state.inker, [], []), {noreply, State#state{scoring_state = undefined}} end; handle_cast({trim, PersistedSQN, ManifestAsList}, State) -> @@ -794,8 +795,7 @@ write_values(KVCList, CDBopts, Journal0, ManSlice0, PressMethod) -> SQN, compact_journal), leveled_log:log("IC009", [FN]), - leveled_cdb:cdb_open_writer(FN, - CDBopts); + leveled_cdb:cdb_open_writer(FN, CDBopts); _ -> {ok, Journal0} end, @@ -1018,9 +1018,10 @@ compact_single_file_recovr_test() -> LedgerFun1, CompactFP, CDB} = compact_single_file_setup(), - [{LowSQN, FN, PidR, _LastKey}] = + CDBOpts = #cdb_options{binary_mode=true}, + [{LowSQN, FN, _PidOldR, LastKey}] = compact_files([Candidate], - #cdb_options{file_path=CompactFP, binary_mode=true}, + CDBOpts#cdb_options{file_path=CompactFP}, LedgerFun1, LedgerSrv1, 9, @@ -1028,6 +1029,7 @@ compact_single_file_recovr_test() -> native), io:format("FN of ~s~n", [FN]), ?assertMatch(2, LowSQN), + {ok, PidR} = leveled_cdb:cdb_reopen_reader(FN, LastKey, CDBOpts), ?assertMatch(probably, leveled_cdb:cdb_keycheck(PidR, {8, @@ -1047,6 +1049,7 @@ compact_single_file_recovr_test() -> test_ledgerkey("Key2")}), ?assertMatch({{_, _}, {"Value2", {[], infinity}}}, leveled_codec:from_inkerkv(RKV1)), + ok = leveled_cdb:cdb_close(PidR), ok = leveled_cdb:cdb_deletepending(CDB), ok = leveled_cdb:cdb_destroy(CDB). @@ -1057,9 +1060,10 @@ compact_single_file_retain_test() -> LedgerFun1, CompactFP, CDB} = compact_single_file_setup(), - [{LowSQN, FN, PidR, _LK}] = + CDBOpts = #cdb_options{binary_mode=true}, + [{LowSQN, FN, _PidOldR, LastKey}] = compact_files([Candidate], - #cdb_options{file_path=CompactFP, binary_mode=true}, + CDBOpts#cdb_options{file_path=CompactFP}, LedgerFun1, LedgerSrv1, 9, @@ -1067,6 +1071,7 @@ compact_single_file_retain_test() -> native), io:format("FN of ~s~n", [FN]), ?assertMatch(1, LowSQN), + {ok, PidR} = leveled_cdb:cdb_reopen_reader(FN, LastKey, CDBOpts), ?assertMatch(probably, leveled_cdb:cdb_keycheck(PidR, {8, @@ -1081,11 +1086,12 @@ compact_single_file_retain_test() -> stnd, test_ledgerkey("Key1")})), RKV1 = leveled_cdb:cdb_get(PidR, - {2, - stnd, - test_ledgerkey("Key2")}), + {2, + stnd, + test_ledgerkey("Key2")}), ?assertMatch({{_, _}, {"Value2", {[], infinity}}}, leveled_codec:from_inkerkv(RKV1)), + ok = leveled_cdb:cdb_close(PidR), ok = leveled_cdb:cdb_deletepending(CDB), ok = leveled_cdb:cdb_destroy(CDB). diff --git a/src/leveled_imanifest.erl b/src/leveled_imanifest.erl index fb72e84..24da2b8 100644 --- a/src/leveled_imanifest.erl +++ b/src/leveled_imanifest.erl @@ -52,6 +52,9 @@ generate_entry(Journal) -> case leveled_cdb:cdb_firstkey(PidR) of {StartSQN, _Type, _PK} -> LastKey = leveled_cdb:cdb_lastkey(PidR), + % close the file here. This will then be re-opened by the inker + % and so will be correctly linked to the inker not to the iclerk + ok = leveled_cdb:cdb_close(PidR), [{StartSQN, NewFN, PidR, LastKey}]; empty -> leveled_log:log("IC013", [NewFN]), diff --git a/src/leveled_inker.erl b/src/leveled_inker.erl index d25f6d9..22266e4 100644 --- a/src/leveled_inker.erl +++ b/src/leveled_inker.erl @@ -698,14 +698,20 @@ handle_call(doom, _From, State) -> handle_cast({clerk_complete, ManifestSnippet, FilesToDelete}, State) -> + CDBOpts = State#state.cdb_options, DropFun = fun(E, Acc) -> leveled_imanifest:remove_entry(Acc, E) end, Man0 = lists:foldl(DropFun, State#state.manifest, FilesToDelete), AddFun = - fun(E, Acc) -> - leveled_imanifest:add_entry(Acc, E, false) + fun(ManEntry, Acc) -> + {LowSQN, FN, _, LK_RO} = ManEntry, + % At this stage the FN has a .cdb extension, which will be + % stripped during add_entry - so need to add the .cdb here + {ok, Pid} = leveled_cdb:cdb_reopen_reader(FN, LK_RO, CDBOpts), + UpdEntry = {LowSQN, FN, Pid, LK_RO}, + leveled_imanifest:add_entry(Acc, UpdEntry, false) end, Man1 = lists:foldl(AddFun, Man0, ManifestSnippet), NewManifestSQN = State#state.manifest_sqn + 1,