Improve tombstone management

include/bitcask.hrl

@@ -13,7 +13,10 @@
                     fd,       % File handle
                     hintfd,   % File handle for hints
                     hintcrc=0,% CRC-32 of current hint
-                    ofs }).   % Current offset for writing
+                    ofs,      % Current offset for writing
+                    l_ofs=0,  % Last offset written to data file
+                    l_hbytes=0,% Last # bytes written to hint file
+                    l_hintcrc=0}). % CRC-32 of current hint prior to last write
 
 -record(file_status, { filename,
                        fragmented,
@@ -26,8 +29,12 @@
 -define(FMT(Str, Args), lists:flatten(io_lib:format(Str, Args))).
 
 -define(TOMBSTONE, <<"bitcask_tombstone">>).
+-define(TOMBSTONE2_STR, "bitcask_tombstone2").
+-define(TOMBSTONE2,<<?TOMBSTONE2_STR>>).
 
--define(OFFSETFIELD,  64).
+-define(OFFSETFIELD_v1,  64).
+-define(TOMBSTONEFIELD_V2, 1).
+-define(OFFSETFIELD_V2,   63).
 -define(TSTAMPFIELD,  32).
 -define(KEYSIZEFIELD, 16).
 -define(TOTALSIZEFIELD, 32).
@@ -36,7 +43,7 @@
 -define(HEADER_SIZE,  14). % 4 + 4 + 2 + 4 bytes
 -define(MAXKEYSIZE, 2#1111111111111111).
 -define(MAXVALSIZE, 2#11111111111111111111111111111111).
--define(MAXOFFSET, 16#ffffffffffffffff). % max 64-bit unsigned
+-define(MAXOFFSET_V2, 16#7fffffffffffffff). % max 63-bit unsigned
 
 %% for hintfile validation
 -define(CHUNK_SIZE, 65535).

rebar.config.script

@@ -7,13 +7,15 @@ PulseBuild = case os:getenv("BITCASK_PULSE") of
 case PulseBuild of
     true ->
         PulseOpts =
-            [{pulse_no_side_effect,[{erlang,display,1}]},
+            [{pulse_no_side_effect,
+              [{erlang,display,1},
+               {bitcask_nifs, keydir_global_pending_frozen, 0}]},
              {pulse_side_effect,
               [ {bitcask_nifs, keydir_new, 0}
               , {bitcask_nifs, keydir_new, 1}
               , {bitcask_nifs, keydir_mark_ready, 1}
-              , {bitcask_nifs, keydir_put_int, 9}
-              , {bitcask_nifs, keydir_get_int, 4}
+              , {bitcask_nifs, keydir_put_int, 10}
+              , {bitcask_nifs, keydir_get_int, 5}
               , {bitcask_nifs, keydir_remove, 3}
               , {bitcask_nifs, keydir_remove_int, 6}
               , {bitcask_nifs, keydir_copy, 1}
@@ -23,8 +25,11 @@ case PulseBuild of
               , {bitcask_nifs, keydir_info, 1}
               , {bitcask_nifs, keydir_release, 1}
               , {bitcask_nifs, keydir_trim_fstats, 2}
+              %% For PULSE purposes, this func is side-effect free (see above)
+              %% bitcask_nifs:keydir_global_pending_frozen/0 
 
               , {bitcask_nifs, increment_file_id, 1}
+              , {bitcask_nifs, increment_file_id, 2}
 
               , {bitcask_nifs, lock_acquire_int,   2}
               , {bitcask_nifs, lock_release_int,   1}
@@ -41,6 +46,7 @@ case PulseBuild of
               , {bitcask_nifs, file_position_int, 2}
               , {bitcask_nifs, file_seekbof_int, 1}
 
+              , {bitcask_file, '_', '_'}
               , {bitcask_time, tstamp, 0}
 
               , {prim_file, '_', '_'}

src/bitcask_file.erl

@@ -23,6 +23,10 @@
 -compile(export_all).
 -behaviour(gen_server).
 
+-ifdef(PULSE).
+-compile({parse_transform, pulse_instrument}).
+-endif.
+
 %% API
 
 %% gen_server callbacks
@@ -70,6 +74,9 @@ file_position(Pid, Position) ->
 file_seekbof(Pid) ->
     file_request(Pid, file_seekbof).
 
+file_truncate(Pid) ->
+    file_request(Pid, file_truncate).
+
 %%%===================================================================
 %%% API helper functions
 %%%===================================================================
@@ -168,6 +175,9 @@ handle_call(file_seekbof, From, State=#state{fd=Fd}) ->
     check_owner(From, State),
     {ok, _} = file:position(Fd, bof),
     {reply, ok, State};
+handle_call(file_truncate, From, State=#state{fd=Fd}) ->
+    check_owner(From, State),
+    {reply, file:truncate(Fd), State};
 
 handle_call(_Request, _From, State) ->
     Reply = ok,

src/bitcask_fileops.erl

@@ -28,7 +28,7 @@
          close/1,
          close_for_writing/1,
          data_file_tstamps/1,
-         write/4,
+         write/5,
          read/3,
          sync/1,
          delete/1,
@@ -38,7 +38,8 @@
          filename/1,
          hintfile_name/1,
          file_tstamp/1,
-         check_write/4]).
+         check_write/4,
+         un_write/1]).
 -export([read_file_info/1, write_file_info/2, is_file/1]).
 
 -include_lib("kernel/include/file.hrl").
@@ -120,7 +121,7 @@ open_file(Filename) ->
         {ok, FD} ->
             {ok, #filestate{mode = read_only,
                             filename = Filename, tstamp = file_tstamp(Filename),
-                            fd = FD, ofs = 0 }};
+                            fd = FD, ofs = 0}};
         {error, Reason} ->
             {error, Reason}
     end.
@@ -150,7 +151,7 @@ close_hintfile(State = #filestate { hintfd = HintFd, hintcrc = HintCRC }) ->
     %% timestamp and offset as large as the file format supports so opening with
     %% an older version of bitcask will just reject the record at the end of the
     %% hintfile and otherwise work normally.
-    Iolist = hintfile_entry(<<>>, 0, {?MAXOFFSET, HintCRC}),
+    Iolist = hintfile_entry(<<>>, 0, 0, ?MAXOFFSET_V2, HintCRC),
     ok = bitcask_io:file_write(HintFd, Iolist),
     bitcask_io:file_sync(HintFd),
     bitcask_io:file_close(HintFd),
@@ -191,14 +192,14 @@ delete(#filestate{ filename = FN } = State) ->
 
 %% @doc Write a Key-named binary data field ("Value") to the Filestate.
 -spec write(#filestate{}, 
-            Key :: binary(), Value :: binary(), Tstamp :: integer()) ->
+            Key :: binary(), Value :: binary(), Tstamp :: integer(), TombstoneP :: boolean()) ->
         {ok, #filestate{}, Offset :: integer(), Size :: integer()} |
         {error, read_only}.
-write(#filestate { mode = read_only }, _K, _V, _Tstamp) ->
+write(#filestate { mode = read_only }, _K, _V, _Tstamp, _TombstoneP) ->
     {error, read_only};
 write(Filestate=#filestate{fd = FD, hintfd = HintFD, 
                            hintcrc = HintCRC0, ofs = Offset},
-      Key, Value, Tstamp) ->
+      Key, Value, Tstamp, TombstoneP) ->
     KeySz = size(Key),
     true = (KeySz =< ?MAXKEYSIZE),
     ValueSz = size(Value),
@@ -212,14 +213,32 @@ write(Filestate=#filestate{fd = FD, hintfd = HintFD,
     ok = bitcask_io:file_pwrite(FD, Offset, Bytes),
     %% Create and store the corresponding hint entry
     TotalSz = KeySz + ValueSz + ?HEADER_SIZE,
-    Iolist = hintfile_entry(Key, Tstamp, {Offset, TotalSz}),
+    TombInt = if TombstoneP -> 1;
+                 true       -> 0
+              end,
+    Iolist = hintfile_entry(Key, Tstamp, TombInt, Offset, TotalSz),
     ok = bitcask_io:file_write(HintFD, Iolist),
     %% Record our final offset
     TotalSz = iolist_size(Bytes),
     HintCRC = erlang:crc32(HintCRC0, Iolist), % compute crc of hint
     {ok, Filestate#filestate{ofs = Offset + TotalSz,
-                            hintcrc = HintCRC}, Offset, TotalSz}.
-
+                             hintcrc = HintCRC,
+                             l_ofs = Offset,
+                             l_hbytes = iolist_size(Iolist),
+                             l_hintcrc = HintCRC0}, Offset, TotalSz}.
+
+%% WARNING: We can only undo the last write.
+un_write(Filestate=#filestate{fd = FD, hintfd = HintFD, 
+                              l_ofs = LastOffset,
+                              l_hbytes = LastHintBytes,
+                              l_hintcrc = LastHintCRC}) ->
+    {ok, _O2} = bitcask_io:file_position(FD, LastOffset),
+    ok = bitcask_io:file_truncate(FD),
+    {ok, 0} = bitcask_io:file_position(FD, 0),
+    {ok, _HO2} = bitcask_io:file_position(HintFD, {cur, -LastHintBytes}),
+    ok = bitcask_io:file_truncate(HintFD),
+    {ok, Filestate#filestate{ofs = LastOffset,
+                             hintcrc = LastHintCRC}}.
 
 %% @doc Given an Offset and Size, get the corresponding k/v from Filename.
 -spec read(Filename :: string() | #filestate{}, Offset :: integer(),
@@ -406,7 +425,8 @@ read_crc(Fd) ->
         {ok, <<0:?TSTAMPFIELD, 
                0:?KEYSIZEFIELD,
                ExpectCRC:?TOTALSIZEFIELD, 
-               (?MAXOFFSET):?OFFSETFIELD>>} ->
+               _TombInt:?TOMBSTONEFIELD_V2,
+               (?MAXOFFSET_V2):?OFFSETFIELD_V2>>} ->
             ExpectCRC;
         _ -> error
     end.
@@ -466,7 +486,7 @@ fold_keys_loop(Fd, Offset, Fun, Acc0) ->
 
 fold_keys_int_loop(<<_Crc32:?CRCSIZEFIELD, Tstamp:?TSTAMPFIELD, 
                      KeySz:?KEYSIZEFIELD, ValueSz:?VALSIZEFIELD, 
-                     Key:KeySz/bytes, _:ValueSz/bytes,
+                     Key:KeySz/bytes, Value:ValueSz/bytes,
                      Rest/binary>>, 
                    Fun, Acc0, Consumed0, 
                    {Offset, AvgValSz0}, 
@@ -475,24 +495,13 @@ fold_keys_int_loop(<<_Crc32:?CRCSIZEFIELD, Tstamp:?TSTAMPFIELD,
     PosInfo = {Offset, TotalSz},
     Consumed = Consumed0 + TotalSz,
     AvgValSz = (AvgValSz0 + ValueSz) div 2,
-    Acc = Fun(Key, Tstamp, PosInfo, Acc0),
+    KeyPlus = case bitcask:is_tombstone(Value) of
+                  true  -> {tombstone, Key};
+                  false -> Key
+              end,
+    Acc = Fun(KeyPlus, Tstamp, PosInfo, Acc0),
     fold_keys_int_loop(Rest, Fun, Acc, Consumed, 
                        {Offset + TotalSz, AvgValSz}, EOI);
-%% in the case where values are very large, we don't actually want to 
-%% get a larger binary if we don't have to, so just issue a skip.
-fold_keys_int_loop(<<_Crc32:?CRCSIZEFIELD, Tstamp:?TSTAMPFIELD, 
-                     KeySz:?KEYSIZEFIELD, ValueSz:?VALSIZEFIELD, 
-                     Key:KeySz/bytes, 
-                     _Rest/binary>>, 
-                   Fun, Acc0, _Consumed0, 
-                   {Offset, AvgValSz0}, 
-                   _EOI) when AvgValSz0 > ?CHUNK_SIZE ->
-    TotalSz = KeySz + ValueSz + ?HEADER_SIZE,
-    PosInfo = {Offset, TotalSz},
-    Acc = Fun(Key, Tstamp, PosInfo, Acc0),
-    AvgValSz = (AvgValSz0 + ValueSz) div 2,
-    NewPos = Offset + TotalSz,
-    {skip, Acc, NewPos, {NewPos, AvgValSz}};
 fold_keys_int_loop(<<>>, _Fun, Acc, Consumed, _Args, EOI) when EOI =:= true ->
     {done, Acc, Consumed};
 fold_keys_int_loop(_Bytes, _Fun, Acc, Consumed, Args, EOI) when EOI =:= false ->
@@ -527,21 +536,26 @@ fold_hintfile(State, Fun, Acc0) ->
 %% hint record, three-tuple done indicates that we've exhausted all bytes, or
 %% it's an error
 fold_hintfile_loop(<<0:?TSTAMPFIELD, 0:?KEYSIZEFIELD,
-                     _ExpectCRC:?TOTALSIZEFIELD, (?MAXOFFSET):?OFFSETFIELD>>,
+                     _ExpectCRC:?TOTALSIZEFIELD,
+                     _TombInt:?TOMBSTONEFIELD_V2, (?MAXOFFSET_V2):?OFFSETFIELD_V2>>,
                    _Fun, Acc, Consumed, _Args, EOI) when EOI =:= true ->
     {done, Acc, Consumed + ?HINT_RECORD_SZ};
 %% main work loop here, containing the full match of hint record and key.
 %% if it gets a match, it proceeds to recurse over the rest of the big
 %% binary
 fold_hintfile_loop(<<Tstamp:?TSTAMPFIELD, KeySz:?KEYSIZEFIELD,
-                     TotalSz:?TOTALSIZEFIELD, Offset:?OFFSETFIELD,
+                     TotalSz:?TOTALSIZEFIELD,
+                     TombInt:?TOMBSTONEFIELD_V2, Offset:?OFFSETFIELD_V2,
                      Key:KeySz/bytes, Rest/binary>>, 
                    Fun, Acc0, Consumed0, {DataSize, HintFile} = Args,
                    EOI) ->
     case Offset + TotalSz =< DataSize + 1 of 
         true ->
             PosInfo = {Offset, TotalSz},
-            Acc = Fun(Key, Tstamp, PosInfo, Acc0),
+            KeyPlus = if TombInt == 1 -> {tombstone, Key};
+                         true         -> Key
+                      end,
+            Acc = Fun(KeyPlus, Tstamp, PosInfo, Acc0),
             Consumed = KeySz + ?HINT_RECORD_SZ + Consumed0,
             fold_hintfile_loop(Rest, Fun, Acc, 
                                Consumed, Args, EOI);
@@ -577,7 +591,6 @@ fold_hintfile_loop(_Bytes, _Fun, Acc0, Consumed0, Args, _EOI) ->
                                 {more, any(), integer(), any()} |
                                 {done, any()} |
                                 {done, any(), integer()} |
-                                {skip, any(), integer(), any()} |
                                 {error, any()}),
                      fun(), any(), any()) -> 
                             {error, any()} | any().
@@ -590,14 +603,8 @@ fold_file_loop(Fd, FoldFn, IntFoldFn, Acc0, Args0, Prev0, ChunkSz0) ->
     %% in datafile folds and key folds
     {Prev, ChunkSz}
         = case Prev0 of 
-              none -> {<<>>, ChunkSz0};
-              %if we're skipping around, we're likely too big
-              skip -> 
-                  CS = case ChunkSz0 >= (?MIN_CHUNK_SIZE * 2) of
-                           true -> ChunkSz0 div 2;
-                           false -> ?MIN_CHUNK_SIZE
-                       end,
-                  {<<>>, CS};
+              none ->
+                  {<<>>, ChunkSz0};
               Other -> 
                   CS = case byte_size(Other) of
                            %% to avoid having to rescan the same 
@@ -627,19 +634,6 @@ fold_file_loop(Fd, FoldFn, IntFoldFn, Acc0, Args0, Prev0, ChunkSz0) ->
                         end,
                     fold_file_loop(Fd, FoldFn, IntFoldFn, 
                                    Acc, Args, Rest, ChunkSz);
-                %% foldfuns should return skip when they have no need for 
-                %% the rest of the binary that they've been handed.
-                %% see fold_int_loop for proper usage.
-                {skip, Acc, SkipTo, Args} ->
-                    case bitcask_io:file_position(Fd, SkipTo) of
-                        {ok, SkipTo} ->
-                            fold_file_loop(Fd, FoldFn, IntFoldFn, 
-                                           Acc, Args, skip, ChunkSz);
-                        {error, Reason} ->
-                            {error, Reason};
-                        Other1 ->
-                            {error, {file_fold_error, Other1}}
-                    end;
                 %% the done two tuple is returned when we want to be 
                 %% unconditionally successfully finished, i.e. trailing data 
                 %% is a non-fatal error
@@ -677,10 +671,10 @@ open_hint_file(Filename, FinalOpts, Count) ->
             open_hint_file(Filename, FinalOpts, Count - 1)
     end.
 
-hintfile_entry(Key, Tstamp, {Offset, TotalSz}) ->
+hintfile_entry(Key, Tstamp, TombInt, Offset, TotalSz) ->
     KeySz = size(Key),
-    [<<Tstamp:?TSTAMPFIELD>>, <<KeySz:?KEYSIZEFIELD>>,
-     <<TotalSz:?TOTALSIZEFIELD>>, <<Offset:?OFFSETFIELD>>, Key].
+    [<<Tstamp:?TSTAMPFIELD, KeySz:?KEYSIZEFIELD, TotalSz:?TOTALSIZEFIELD,
+       TombInt:?TOMBSTONEFIELD_V2, Offset:?OFFSETFIELD_V2>>, Key].
 
 %% ===================================================================
 %% file/filelib avoidance code.
@@ -745,7 +739,7 @@ get_efile_port() ->
     Key = bitcask_efile_port,
     case get(Key) of
         undefined ->
-            case prim_file_drv_open(efile, [binary]) of
+            case prim_file_drv_open("efile", [binary]) of
                 {ok, Port} ->
                     put(Key, Port),
                     get_efile_port();

src/bitcask_io.erl

@@ -22,6 +22,10 @@
 -module(bitcask_io).
 -compile(export_all).
 
+-ifdef(PULSE).
+-compile({parse_transform, pulse_instrument}).
+-endif.
+
 file_open(Filename, Opts) ->
     M = file_module(),
     M:file_open(Filename, Opts).
@@ -58,6 +62,10 @@ file_position(Ref, Position) ->
     M = file_module(),
     M:file_position(Ref, Position).
 
+file_truncate(Ref) ->
+    M = file_module(),
+    M:file_truncate(Ref).
+
 file_module() ->
     case get(bitcask_file_mod) of
         undefined ->