diff --git a/src/bitcask.erl b/src/bitcask.erl index 267d086..430ea68 100644 --- a/src/bitcask.erl +++ b/src/bitcask.erl @@ -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. @@ -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(), @@ -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(), @@ -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)), @@ -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} -> @@ -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. @@ -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 @@ -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). @@ -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 @@ -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); @@ -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; @@ -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}.