umgeher/leveled
0
Fork 0
Code Issues Pull requests Releases Activity Actions

Close Pending Deletes on shutdown

Browse source 
Pending deletes are not in the Manifest, and so were not being closed on
shutdown
This commit is contained in:
martinsumner 2017-01-17 16:54:54 +00:00
parent 7086717765
commit 850a524ad3
2 changed files with 19 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

18
src/leveled_pmanifest.erl
View file

@ -52,8 +52,8 @@
% The smallest snapshot manifest SQN in the snapshot % The smallest snapshot manifest SQN in the snapshot
% list % list
pending_deletes :: dict:dict(), pending_deletes :: dict:dict(),
% a dictionary mapping keys (filenames) to SQN when % a dictionary mapping keys (filenames) to SQN when the
% the deletion was made % deletion was made, and the original Manifest Entry
basement :: integer() basement :: integer()
% Currently the lowest level (the largest number) % Currently the lowest level (the largest number)
}). }).
@ -113,7 +113,13 @@ close_manifest(Manifest, CloseEntryFun) ->
Level = array:get(LevelIdx, Manifest#manifest.levels), Level = array:get(LevelIdx, Manifest#manifest.levels),
close_level(LevelIdx, Level, CloseEntryFun) close_level(LevelIdx, Level, CloseEntryFun)
end, end,
lists:foreach(CloseLevelFun, lists:seq(0, Manifest#manifest.basement)). lists:foreach(CloseLevelFun, lists:seq(0, Manifest#manifest.basement)),
ClosePDFun =
fun({_FN, {_SQN, ME}}) ->
CloseEntryFun(ME)
end,
lists:foreach(ClosePDFun, dict:to_list(Manifest#manifest.pending_deletes)).
save_manifest(Manifest, RootPath) -> save_manifest(Manifest, RootPath) ->
FP = filepath(RootPath, Manifest#manifest.manifest_sqn, current_manifest), FP = filepath(RootPath, Manifest#manifest.manifest_sqn, current_manifest),
@ -138,7 +144,9 @@ remove_manifest_entry(Manifest, ManSQN, LevelIdx, Entry) ->
leveled_log:log("PC019", ["remove", LevelIdx, UpdLevel]), leveled_log:log("PC019", ["remove", LevelIdx, UpdLevel]),
DelFun = DelFun =
fun(E, Acc) -> fun(E, Acc) ->
dict:store(E#manifest_entry.filename, ManSQN, Acc) dict:store(E#manifest_entry.filename,
{ManSQN, E},
Acc)
end, end,
Entries = Entries =
case is_list(Entry) of case is_list(Entry) of
@ -258,7 +266,7 @@ release_snapshot(Manifest, Pid) ->
end. end.
ready_to_delete(Manifest, Filename) -> ready_to_delete(Manifest, Filename) ->
ChangeSQN = dict:fetch(Filename, Manifest#manifest.pending_deletes), {ChangeSQN, _ME} = dict:fetch(Filename, Manifest#manifest.pending_deletes),
case Manifest#manifest.min_snapshot_sqn of case Manifest#manifest.min_snapshot_sqn of
0 -> 0 ->
% no shapshots % no shapshots

12
test/end_to_end/basic_SUITE.erl
View file

@ -12,12 +12,12 @@
]). ]).
all() -> [ all() -> [
simple_put_fetch_head_delete, % simple_put_fetch_head_delete,
many_put_fetch_head, % many_put_fetch_head,
journal_compaction, % journal_compaction,
fetchput_snapshot, % fetchput_snapshot,
load_and_count, % load_and_count,
load_and_count_withdelete, % load_and_count_withdelete,
space_clear_ondelete space_clear_ondelete
]. ].