Skip to content

Track expiry stats #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aef- wants to merge 1 commit into
base: openriak-3.4
Choose a base branch
from
Open

Track expiry stats #4

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
117 changes: 106 additions & 11 deletions src/bitcask.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@
max_file_size = 0 :: integer(), % Max. size of a written file
opts = [] :: list(), % Original options used to open the bitcask
key_transform :: function() | undefined,
stats_callback :: function() | undefined,
keydir :: reference(), % Key directory
read_write_p = 0 :: integer(), % integer() avoids atom -> NIF
% What tombstone style to write, for testing purposes only.
Expand All @@ -95,6 +96,11 @@
-type bitcask_set() :: set().
-endif.

-record(mstats, {
expired_keys = 0 :: integer(),
expired_bytes = 0 :: integer() % For each expired key, bytes represent: KeySize + ValueSize + HEADER_SIZE (constant)
}).

-record(mstate, { dirname :: string(),
merge_lock :: reference(),
max_file_size :: integer(),
Expand All @@ -108,6 +114,7 @@
del_keydir :: reference(),
expiry_time :: integer(),
expiry_grace_time :: integer(),
stats :: #mstats{},
key_transform :: function(),
read_write_p :: integer(), % integer() avoids atom -> NIF
opts :: list(),
Expand Down Expand Up @@ -148,6 +155,8 @@ open(Dirname, Opts) ->
%% Get the max file size parameter from opts
MaxFileSize = get_opt(max_file_size, Opts),

StatsCallback = get_stats_callback(get_opt(stats_callback, Opts)),

%% Get the number of seconds we are willing to wait for the keydir init to timeout
WaitTime = timer:seconds(get_opt(open_timeout, Opts)),

Expand Down Expand Up @@ -177,6 +186,7 @@ open(Dirname, Opts) ->
keydir = KeyDir,
key_transform = KeyTransformFun,
tombstone_version = TombstoneVersion,
stats_callback = StatsCallback,
read_write_p = ReadWriteI}),
Ref;
{error, Reason} ->
Expand Down Expand Up @@ -248,6 +258,8 @@ get(Ref, Key, TryNum) ->
E#bitcask_entry.file_id,
E#bitcask_entry.offset) of
ok ->
StatsCallback = State#bc_state.stats_callback,
StatsCallback({expired_keys, 1, E#bitcask_entry.total_sz}),
not_found;
already_exists ->
% Updated since last read, try again.
Expand Down Expand Up @@ -703,10 +715,12 @@ merge1(Dirname, Opts, FilesToMerge0, ExpiredFiles) ->
key_transform = KT,
read_write_p = 0,
opts = Opts,
stats = #mstats {},
delete_files = []},

%% Finally, start the merge process
ExpiredFilesFinished = expiry_merge(InExpiredFiles, LiveKeyDir, KT, []),
{ExpiredFilesFinished, KeysExpired, BytesExpired} = expiry_merge(InExpiredFiles, LiveKeyDir, KT, {[], 0, 0}),

State1 = merge_files(State),

%% Make sure to close the final output file
Expand Down Expand Up @@ -737,6 +751,13 @@ merge1(Dirname, Opts, FilesToMerge0, ExpiredFiles) ->
%% Explicitly release our keydirs instead of waiting for GC
bitcask_nifs:keydir_release(LiveKeyDir),
bitcask_nifs:keydir_release(DelKeyDir),

StatsCallback = get_stats_callback(get_opt(stats_callback, Opts)),
StatsCallback({
expired_keys,
KeysExpired + State1#mstate.stats#mstats.expired_keys,
BytesExpired + State1#mstate.stats#mstats.expired_bytes
}),

ok = bitcask_lockops:release(Lock).

Expand Down Expand Up @@ -1406,7 +1427,7 @@ merge_files(#mstate { dirname = Dirname,
end,
merge_files(State2#mstate { input_files = Rest }).

merge_single_entry(K, V, Tstamp, FileId, {_, _, Offset, _} = Pos, State) ->
merge_single_entry(K, V, Tstamp, FileId, {_, _, Offset, Bytes} = Pos, State) ->
case out_of_date(State, K, Tstamp, FileId, Pos, State#mstate.expiry_time,
false,
[State#mstate.live_keydir, State#mstate.del_keydir]) of
Expand All @@ -1429,7 +1450,19 @@ merge_single_entry(K, V, Tstamp, FileId, {_, _, Offset, _} = Pos, State) ->
%% Remove only if this is the current entry in the keydir
bitcask_nifs:keydir_remove(State#mstate.live_keydir, K,
Tstamp, FileId, Offset),
State;

OldStats = State#mstate.stats,
OldExpiredBytes = OldStats#mstats.expired_bytes,
OldExpiredKeys = OldStats#mstats.expired_keys,

NewStats = OldStats#mstats {
expired_bytes = OldExpiredBytes + Bytes,
expired_keys = OldExpiredKeys + 1
},

State#mstate {
stats = NewStats
};
not_found ->
%% First tombstone seen for this key during this merge
merge_single_tombstone(K,V, Tstamp, FileId, Offset, State);
Expand Down Expand Up @@ -1968,31 +2001,40 @@ expiry_merge([File | Files], LiveKeyDir, KT, Acc0) ->
FileId = bitcask_fileops:file_tstamp(File),
Fun = fun({tombstone, _}, _, _, Acc) ->
Acc;
(K0, Tstamp, {Offset, _TotalSz}, Acc) ->
(K0, Tstamp, {Offset, TotalSz}, {KeysExpired, BytesExpired} = Acc) ->
K = try KT(K0) catch TxErr -> {key_tx_error, TxErr} end,
case K of
{key_tx_error, KeyTxErr} ->
?LOG_ERROR("Invalid key on merge ~p: ~p",
[K0, KeyTxErr]);
[K0, KeyTxErr]),
Acc;
_ ->
bitcask_nifs:keydir_remove(LiveKeyDir, K, Tstamp,
FileId, Offset)
end,
Acc
FileId, Offset),
{KeysExpired + 1, BytesExpired + TotalSz}
end
end,
case bitcask_fileops:fold_keys(File, Fun, ok, default) of
case bitcask_fileops:fold_keys(File, Fun, {0, 0}, default) of
{error, Reason} ->
?LOG_ERROR("Error folding keys for ~p: ~p\n",
[File#filestate.filename,Reason]),
Acc = Acc0;
_ ->
{NewKeysExpired, NewBytesExpired} ->
?LOG_INFO("All keys expired in: ~p scheduling "
"file for deletion\n",
[File#filestate.filename]),
Acc = lists:append(Acc0, [File])

{FilesRemoved, KeysExpired, BytesExpired} = Acc0,
Acc = {[File | FilesRemoved], KeysExpired + NewKeysExpired, BytesExpired + NewBytesExpired}
end,
expiry_merge(Files, LiveKeyDir, KT, Acc).

get_stats_callback(StatsCallback)
when is_function(StatsCallback) ->
StatsCallback;
get_stats_callback(_State) ->
fun (_) -> ok end.

get_key_transform(KT)
when is_function(KT) ->
KT;
Expand Down Expand Up @@ -3477,6 +3519,59 @@ no_crash_on_key_transform_test() ->
ok = bitcask:close(B2),
ok.

get_expired_test_() ->
{foreach,
fun() ->
meck:new(stats, [non_strict]),
meck:expect(stats, callback, fun(_) -> ok end),
ok
end,
fun(_) ->
meck:unload()
end,
[{timeout, 4, fun get_call_expires_key/0},
{timeout, 6, fun stats_callback_is_called_only_when_expired_key_is_cleaned_up/0},
{timeout, 6, fun stats_callback_gets_called_twice_for_multiple_expirations/0}]}.

%% Tests that keys are no longer returned when a get is called on an expired key.
get_call_expires_key() ->
Dir = setup_testfolder("bc.get.expired.keys"),

B1 = bitcask:open(Dir, [read_write, {expiry_secs, 1}, {stats_callback, fun stats:callback/1}]),
ok = bitcask:put(B1,<<"k">>,<<"v">>),
ok = bitcask:put(B1,<<"k">>,<<"b">>),
timer:sleep(2000),
not_found = bitcask:get(B1,<<"k">>),
NumCalls = meck:num_calls(stats, callback, [{expired_keys, 1, 16}]),
?assertEqual(NumCalls, 1).

stats_callback_is_called_only_when_expired_key_is_cleaned_up() ->
Dir = setup_testfolder("bc.get.expired.keys"),

B1 = bitcask:open(Dir, [read_write, {expiry_secs, 1}, {stats_callback, fun stats:callback/1}]),
%% This first key, while "expired" will never count as expired.
%% This is because nothing (e.g. get) is triggering the clean-up
ok = bitcask:put(B1,<<"k">>,<<"v">>),
timer:sleep(2000),
ok = bitcask:put(B1,<<"k">>,<<"b">>),
timer:sleep(2000),
not_found = bitcask:get(B1,<<"k">>),
NumCalls = meck:num_calls(stats, callback, [{expired_keys, 1, 16}]),
?assertEqual(NumCalls, 1).

stats_callback_gets_called_twice_for_multiple_expirations() ->
Dir = setup_testfolder("bc.get.expired.keys"),

B1 = bitcask:open(Dir, [read_write, {expiry_secs, 1}, {stats_callback, fun stats:callback/1}]),
ok = bitcask:put(B1,<<"b">>,<<"v">>),
timer:sleep(2000),
not_found = bitcask:get(B1,<<"b">>),
ok = bitcask:put(B1,<<"k">>,<<"b">>),
timer:sleep(2000),
not_found = bitcask:get(B1,<<"k">>),
NumCalls = meck:num_calls(stats, callback, [{expired_keys, 1, 16}]),
?assertEqual(NumCalls, 2).


total_byte_stats_test_() ->
{timeout, 60, fun total_byte_stats_test2/0}.
Expand Down
Loading