Packages

Bitcask key value store

Current section

Files

Jump to
bitcask src bitcask_fileops.erl
Raw

src/bitcask_fileops.erl

%% -------------------------------------------------------------------
%%
%% bitcask: Eric Brewer-inspired key/value store
%%
%% Copyright (c) 2007-2010 Basho Technologies, Inc. All Rights Reserved.
%%
%% This file is provided to you under the Apache License,
%% Version 2.0 (the "License"); you may not use this file
%% except in compliance with the License. You may obtain
%% a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied. See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
%% -------------------------------------------------------------------
%% @doc Basic file i/o operations for bitcask.
-module(bitcask_fileops).
-export([create_file/3,
open_file/1,
open_file/2,
close/1,
close_all/1,
close_for_writing/1,
data_file_tstamps/1,
write/4,
read/3,
sync/1,
delete/1,
fold/3,
fold_keys/3, fold_keys/4,
mk_filename/2,
filename/1,
hintfile_name/1,
file_tstamp/1,
check_write/4,
un_write/1]).
-export([read_file_info/1, write_file_info/2, is_file/1]).
-include_lib("kernel/include/file.hrl").
-include("bitcask.hrl").
-define(HINT_RECORD_SZ, 18). % Tstamp(4) + KeySz(2) + TotalSz(4) + Offset(8)
-ifdef(PULSE).
-compile({parse_transform, pulse_instrument}).
-endif.
-ifdef(TEST).
-ifdef(EQC).
-include_lib("eqc/include/eqc.hrl").
-include_lib("eqc/include/eqc_fsm.hrl").
-endif.
-compile(export_all).
-include_lib("eunit/include/eunit.hrl").
-endif.
%% @doc Open a new file for writing.
%% Called on a Dirname, will open a fresh file in that directory.
-spec create_file(Dirname :: string(), Opts :: [any()],
reference()) ->
{ok, #filestate{}} | {error, term()}.
create_file(DirName, Opts0, Keydir) ->
Opts = [create|Opts0],
case get_create_lock(DirName) of
{ok, Lock} ->
try
{ok, Newest} = bitcask_nifs:increment_file_id(Keydir),
Filename = mk_filename(DirName, Newest),
ok = ensure_dir(Filename),
%% Check for o_sync strategy and add to opts
FinalOpts =
case bitcask:get_opt(sync_strategy, Opts) of
o_sync ->
[o_sync | Opts];
_ ->
Opts
end,
{ok, FD} = bitcask_io:file_open(Filename, FinalOpts),
HintFD = open_hint_file(Filename, FinalOpts),
{ok, #filestate{mode = read_write,
filename = Filename,
tstamp = file_tstamp(Filename),
hintfd = HintFD, fd = FD, ofs = 0}}
catch Error:Reason ->
%% if we fail somehow, do we need to nuke any partial
%% state?
{Error, Reason}
after
bitcask_lockops:release(Lock)
end;
Else ->
Else
end.
get_create_lock(DirName) ->
get_create_lock(DirName, 100).
get_create_lock(_DirName, 0) ->
error(lock_failure);
get_create_lock(DirName, N) ->
timer:sleep(100-N),
case bitcask_lockops:acquire(create, DirName) of
{ok, Lock} ->
{ok, Lock};
{error, locked} ->
get_create_lock(DirName, N - 1);
{error, _} = Else ->
Else
end.
%% @doc Open an existing file for reading.
%% Called with fully-qualified filename.
-spec open_file(Filename :: string()) -> {ok, #filestate{}} | {error, any()}.
open_file(Filename) ->
open_file(Filename, readonly).
open_file(Filename, append) ->
case bitcask_io:file_open(Filename, []) of
{ok, FD} ->
case bitcask_io:file_position(FD, {eof, 0}) of
{ok, 0} ->
% File was deleted and we just opened a new one, undo.
bitcask_io:file_close(FD),
_ = file:delete(Filename),
{error, enoent};
{ok, Ofs} ->
case reopen_hintfile(Filename) of
{error, enoent} ->
bitcask_io:file_close(FD),
{error, enoent};
{undefined, _} ->
bitcask_io:file_close(FD),
{error, enoent};
{HintFD, HintCRC} ->
{ok,
#filestate{mode = read_write,
filename = Filename,
tstamp = file_tstamp(Filename),
fd = FD,
hintfd = HintFD,
hintcrc = HintCRC,
ofs = Ofs
}}
end
end;
{error, _Reason} = Err ->
Err
end;
open_file(Filename, readonly) ->
case bitcask_io:file_open(Filename, [readonly]) of
{ok, FD} ->
{ok, #filestate{mode = read_only,
filename = Filename, tstamp = file_tstamp(Filename),
fd = FD, ofs = 0}};
{error, Reason} ->
{error, Reason}
end.
% Re-open hintfile for appending.
-spec reopen_hintfile(string() | #filestate{}) ->
{error, enoent} | {HintFD::port() | undefined, CRC :: non_neg_integer()}.
reopen_hintfile(Filename) ->
case (catch open_hint_file(Filename, [])) of
couldnt_open_hintfile ->
{undefined, 0};
HintFD ->
HintFilename = hintfile_name(Filename),
{ok, HintI} = read_file_info(HintFilename),
HintSize = HintI#file_info.size,
case bitcask_io:file_position(HintFD, HintSize) of
{ok, 0} ->
bitcask_io:file_close(HintFD),
_ = file:delete(HintFilename),
{error, enoent};
{ok, _FileSize} ->
prepare_hintfile_for_append(HintFD)
end
end.
% Removes the final CRC record so more records can be added to the file.
-spec prepare_hintfile_for_append(HintFD :: port()) ->
{HintFD :: port() | undefined, CRC :: non_neg_integer()}.
prepare_hintfile_for_append(HintFD) ->
case bitcask_io:file_position(HintFD,
{eof, -?HINT_RECORD_SZ}) of
{ok, _} ->
case read_crc(HintFD) of
error ->
bitcask_io:file_close(HintFD),
{undefined, 0};
HintCRC ->
bitcask_io:file_position(HintFD,
{eof, -?HINT_RECORD_SZ}),
bitcask_io:file_truncate(HintFD),
{HintFD, HintCRC}
end;
_ ->
bitcask_io:file_close(HintFD),
{undefined, 0}
end.
%% @doc Use when done writing a file. (never open for writing again)
-spec close(#filestate{} | fresh | undefined) -> ok.
close(fresh) -> ok;
close(undefined) -> ok;
close(State = #filestate{ fd = FD }) ->
_ = close_hintfile(State),
bitcask_io:file_close(FD),
ok.
%% @doc Use when closing multiple files. (never open for writing again)
-spec close_all([#filestate{} | fresh | undefined]) -> ok.
close_all(FileStates) ->
lists:foreach(fun ?MODULE:close/1, FileStates),
ok.
%% @doc Close a file for writing, but leave it open for reads.
-spec close_for_writing(#filestate{} | fresh | undefined) -> #filestate{} | ok.
close_for_writing(fresh) -> ok;
close_for_writing(undefined) -> ok;
close_for_writing(State = #filestate{ mode = read_write, fd = Fd }) ->
S2 = close_hintfile(State),
bitcask_io:file_sync(Fd),
S2#filestate { mode = read_only }.
close_hintfile(State = #filestate { hintfd = undefined }) ->
State;
close_hintfile(State = #filestate { hintfd = HintFd, hintcrc = HintCRC }) ->
%% Write out CRC check at end of hint file. Write with an empty key, zero
%% 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, 0, ?MAXOFFSET_V2, HintCRC),
_ = bitcask_io:file_write(HintFd, Iolist),
_ = bitcask_io:file_sync(HintFd),
_ = bitcask_io:file_close(HintFd),
State#filestate { hintfd = undefined, hintcrc = 0 }.
%% Build a list of {tstamp, filename} for all files in the directory that
%% match our regex.
-spec data_file_tstamps(Dirname :: string()) -> [{integer(), string()}].
data_file_tstamps(Dirname) ->
case list_dir(Dirname) of
{ok, Files} ->
lists:foldl(
fun(Filename, Acc) ->
case string:tokens(Filename, ".") of
[TSString, _, "data"] ->
[{list_to_integer(TSString),
filename:join(Dirname, Filename)}
| Acc];
_ ->
Acc
end
end,
[], Files)
end.
%% @doc Use only after merging, to permanently delete a data file.
-spec delete(#filestate{}) -> ok | {error, atom()}.
delete(#filestate{ filename = FN } = State) ->
_ = file:delete(FN),
case has_hintfile(State) of
true ->
file:delete(hintfile_name(State));
false ->
ok
end.
%% @doc Write a Key-named binary data field ("Value") to the Filestate.
-spec write(#filestate{},
Key :: binary(), Value :: binary(), Tstamp :: integer()) ->
{ok, #filestate{}, Offset :: integer(), Size :: integer()} |
{error, read_only}.
write(#filestate { mode = read_only }, _K, _V, _Tstamp) ->
{error, read_only};
write(Filestate=#filestate{fd = FD, hintfd = HintFD,
hintcrc = HintCRC0, ofs = Offset},
Key, Value, Tstamp) ->
KeySz = size(Key),
true = (KeySz =< ?MAXKEYSIZE),
ValueSz = size(Value),
true = (ValueSz =< ?MAXVALSIZE),
%% Setup io_list for writing -- avoid merging binaries if we can help it
Bytes0 = [<<Tstamp:?TSTAMPFIELD>>, <<KeySz:?KEYSIZEFIELD>>,
<<ValueSz:?VALSIZEFIELD>>, Key, Value],
Bytes = [<<(erlang:crc32(Bytes0)):?CRCSIZEFIELD>> | Bytes0],
%% Store the full entry in the data file
try
ok = bitcask_io:file_pwrite(FD, Offset, Bytes),
%% Create and store the corresponding hint entry
TotalSz = iolist_size(Bytes),
TombInt = case bitcask:is_tombstone(Value) of
true -> 1;
false -> 0
end,
Iolist = hintfile_entry(Key, Tstamp, TombInt, Offset, TotalSz),
case HintFD of
undefined ->
ok;
_ ->
ok = bitcask_io:file_write(HintFD, Iolist)
end,
%% Record our final offset
HintCRC = erlang:crc32(HintCRC0, Iolist), % compute crc of hint
{ok, Filestate#filestate{ofs = Offset + TotalSz,
hintcrc = HintCRC,
l_ofs = Offset,
l_hbytes = iolist_size(Iolist),
l_hintcrc = HintCRC0}, Offset, TotalSz}
catch
error:{badmatch,Error} ->
Error
end.
%% 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(),
Size :: integer()) ->
{ok, Key :: binary(), Value :: binary()} |
{error, bad_crc} | {error, atom()}.
read(Filename, Offset, Size) when is_list(Filename) ->
case open_file(Filename) of
{ok, Fstate} ->
read(Fstate, Offset, Size);
{error, Reason} ->
{error, Reason}
end;
read(#filestate { fd = FD }, Offset, Size) ->
case bitcask_io:file_pread(FD, Offset, Size) of
{ok, <<Crc32:?CRCSIZEFIELD/unsigned, Bytes/binary>>} ->
%% Verify the CRC of the data
case erlang:crc32(Bytes) of
Crc32 ->
%% Unpack the actual data
<<_Tstamp:?TSTAMPFIELD,
KeySz:?KEYSIZEFIELD, ValueSz:?VALSIZEFIELD,
Key:KeySz/bytes, Value:ValueSz/bytes>> = Bytes,
{ok, Key, Value};
_BadCrc ->
{error, bad_crc}
end;
eof ->
{error, eof};
{error, Reason} ->
{error, Reason}
end.
%% @doc Call the OS's fsync(2) system call on the cask and hint files.
-spec sync(#filestate{}) -> ok.
sync(#filestate { mode = read_write, fd = Fd, hintfd = HintFd }) ->
ok = bitcask_io:file_sync(Fd),
ok = bitcask_io:file_sync(HintFd).
-spec fold(fresh | #filestate{},
fun((binary(), binary(), integer(),
{list(), integer(), integer(), integer()}, any()) -> any()),
any()) ->
any() | {error, any()}.
fold(fresh, _Fun, Acc) -> Acc;
fold(#filestate { fd=Fd, filename=Filename, tstamp=FTStamp }, Fun, Acc0) ->
%% TODO: Add some sort of check that this is a read-only file
ok = bitcask_io:file_seekbof(Fd),
case fold_file_loop(Fd, regular, fun fold_int_loop/5, Fun, Acc0,
{Filename, FTStamp, 0, 0}) of
{error, Reason} ->
{error, Reason};
Acc -> Acc
end.
-type key_fold_fun() :: fun((binary(), integer(), {integer(), integer()}, any()) -> any()).
-type key_fold_mode() :: datafile | hintfile | default | recovery.
-spec fold_keys(fresh | #filestate{}, key_fold_fun(), any()) ->
any() | {error, any()}.
fold_keys(fresh, _Fun, Acc) -> Acc;
fold_keys(State, Fun, Acc) ->
fold_keys(State, Fun, Acc, default).
-spec fold_keys(fresh | #filestate{}, key_fold_fun(), any(), key_fold_mode()) ->
any() | {error, any()}.
fold_keys(State, Fun, Acc, datafile) ->
fold_keys_loop(State, 0, Fun, Acc);
fold_keys(#filestate { fd = _Fd } = State, Fun, Acc, hintfile) ->
fold_hintfile(State, Fun, Acc);
fold_keys(State, Fun, Acc, Mode) ->
fold_keys(State, Fun, Acc, Mode, has_hintfile(State)).
fold_keys(State, Fun, Acc, default, true) ->
fold_hintfile(State, Fun, Acc);
fold_keys(State, Fun, Acc, default, false) ->
fold_keys_loop(State, 0, Fun, Acc);
fold_keys(State, Fun, Acc, recovery, true) ->
fold_keys(State, Fun, Acc, recovery, true, has_valid_hintfile(State));
fold_keys(State, Fun, Acc, recovery, false) ->
fold_keys_loop(State, 0, Fun, Acc).
fold_keys(State, Fun, Acc, recovery, _, true) ->
case fold_hintfile(State, Fun, Acc) of
{error, {trunc_hintfile, Acc0}} ->
Acc0;
{error, Reason} ->
HintFile = hintfile_name(State),
error_logger:warning_msg("Hintfile '~s' failed fold: ~p\n",
[HintFile, Reason]),
fold_keys_loop(State, 0, Fun, Acc);
Acc1 ->
Acc1
end;
fold_keys(State, Fun, Acc, recovery, _, false) ->
HintFile = hintfile_name(State),
error_logger:warning_msg("Hintfile '~s' invalid\n",
[HintFile]),
fold_keys_loop(State, 0, Fun, Acc).
-spec mk_filename(string(), integer()) -> string().
mk_filename(Dirname, Tstamp) ->
filename:join(Dirname,
lists:concat([integer_to_list(Tstamp),".bitcask.data"])).
-spec filename(#filestate{}) -> string().
filename(#filestate { filename = Fname }) ->
Fname.
-spec hintfile_name(string() | #filestate{}) -> string().
hintfile_name(Filename) when is_list(Filename) ->
filename:rootname(Filename, ".data") ++ ".hint";
hintfile_name(#filestate { filename = Fname }) ->
hintfile_name(Fname).
-spec file_tstamp(#filestate{} | string()) -> integer().
file_tstamp(#filestate{tstamp=Tstamp}) ->
Tstamp;
file_tstamp(Filename) when is_list(Filename) ->
list_to_integer(filename:basename(Filename, ".bitcask.data")).
-spec check_write(fresh | #filestate{}, binary(), non_neg_integer(), integer()) ->
fresh | wrap | ok.
check_write(fresh, _Key, _ValSize, _MaxSize) ->
%% for the very first write, special-case
fresh;
check_write(#filestate { ofs = Offset }, Key, ValSize, MaxSize) ->
Size = ?HEADER_SIZE + size(Key) + ValSize,
case (Offset + Size) > MaxSize of
true ->
wrap;
false ->
ok
end.
has_hintfile(#filestate { filename = Fname }) ->
is_file(hintfile_name(Fname)).
%% Return true if there is a hintfile and it has
%% a valid CRC check
has_valid_hintfile(State) ->
HintFile = hintfile_name(State),
case bitcask_io:file_open(HintFile, [readonly, read_ahead]) of
{ok, HintFd} ->
{ok, HintI} = read_file_info(HintFile),
HintSize = HintI#file_info.size,
hintfile_validate_loop(HintFd, 0, HintSize);
_ ->
false
end.
hintfile_validate_loop(Fd, CRC0, Rem) ->
{ReadLen, HasCRC} =
case Rem =< ?CHUNK_SIZE of
true ->
case Rem < ?HINT_RECORD_SZ of
true ->
{0, error};
false ->
{Rem - ?HINT_RECORD_SZ, true}
end;
false ->
{?CHUNK_SIZE, false}
end,
case bitcask_io:file_read(Fd, ReadLen) of
{ok, Bytes} ->
case HasCRC of
true ->
ExpectCRC = read_crc(Fd),
CRC = erlang:crc32(CRC0, Bytes),
ExpectCRC =:= CRC;
false ->
hintfile_validate_loop(Fd,
erlang:crc32(CRC0, Bytes),
Rem - ReadLen);
error ->
false
end;
_ -> false
end.
read_crc(Fd) ->
case bitcask_io:file_read(Fd, ?HINT_RECORD_SZ) of
{ok, <<0:?TSTAMPFIELD,
0:?KEYSIZEFIELD,
ExpectCRC:?TOTALSIZEFIELD,
_TombInt:?TOMBSTONEFIELD_V2,
(?MAXOFFSET_V2):?OFFSETFIELD_V2>>} ->
ExpectCRC;
_ -> error
end.
%% ===================================================================
%% Internal functions
%% ===================================================================
fold_int_loop(_Bytes, _Fun, Acc, _Consumed, {Filename, _, Offset, 20}) ->
error_logger:error_msg("fold_loop: CRC error limit at file ~p offset ~p\n",
[Filename, Offset]),
{done, Acc};
fold_int_loop(<<Crc32:?CRCSIZEFIELD, Tstamp:?TSTAMPFIELD,
KeySz:?KEYSIZEFIELD, ValueSz:?VALSIZEFIELD,
Key:KeySz/bytes, Value:ValueSz/bytes, Rest/binary>>,
Fun, Acc0, Consumed0,
{Filename, FTStamp, Offset, CrcSkipCount}) ->
TotalSz = KeySz + ValueSz + ?HEADER_SIZE,
case erlang:crc32([<<Tstamp:?TSTAMPFIELD, KeySz:?KEYSIZEFIELD,
ValueSz:?VALSIZEFIELD>>, Key, Value]) of
Crc32 ->
PosInfo = {Filename, FTStamp, Offset, TotalSz},
Acc = Fun(Key, Value, Tstamp, PosInfo, Acc0),
fold_int_loop(Rest, Fun, Acc, Consumed0 + TotalSz,
{Filename, FTStamp, Offset + TotalSz,
CrcSkipCount});
_ ->
error_logger:error_msg("fold_loop: CRC error at file ~s offset ~p, "
"skipping ~p bytes\n",
[Filename, Offset, TotalSz]),
fold_int_loop(Rest, Fun, Acc0, Consumed0 + TotalSz,
{Filename, FTStamp, Offset + TotalSz,
CrcSkipCount + 1})
end;
fold_int_loop(_Bytes, _Fun, Acc, Consumed, Args) ->
{more, Acc, Consumed, Args}.
fold_keys_loop(#filestate{fd=Fd, filename=Filename, tstamp=FTStamp}, Offset,
Fun, Acc0) ->
case bitcask_io:file_position(Fd, Offset) of
{ok, Offset} -> ok;
Other -> error(Other)
end,
case fold_file_loop(Fd, regular, fun fold_keys_int_loop/5, Fun, Acc0,
{Filename, FTStamp, Offset, 0}) of
{error, Reason} ->
{error, Reason};
Acc -> Acc
end.
fold_keys_int_loop(_Bytes, _Fun, Acc, _Consumed, {Filename, _, Offset, 20}) ->
error_logger:error_msg("fold_loop: CRC error limit at file ~p offset ~p\n",
[Filename, Offset]),
{done, Acc};
fold_keys_int_loop(<<Crc32:?CRCSIZEFIELD, Tstamp:?TSTAMPFIELD,
KeySz:?KEYSIZEFIELD, ValueSz:?VALSIZEFIELD,
Key:KeySz/bytes, Value:ValueSz/bytes, Rest/binary>>,
Fun, Acc0, Consumed0,
{Filename, FTStamp, Offset, CrcSkipCount}) ->
TotalSz = KeySz + ValueSz + ?HEADER_SIZE,
case erlang:crc32([<<Tstamp:?TSTAMPFIELD, KeySz:?KEYSIZEFIELD,
ValueSz:?VALSIZEFIELD>>, Key, Value]) of
Crc32 ->
PosInfo = {Offset, TotalSz},
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, Consumed0 + TotalSz,
{Filename, FTStamp, Offset + TotalSz,
CrcSkipCount});
_ ->
error_logger:error_msg("fold_loop: CRC error at file ~s offset ~p, "
"skipping ~p bytes\n",
[Filename, Offset, TotalSz]),
fold_keys_int_loop(Rest, Fun, Acc0, Consumed0 + TotalSz,
{Filename, FTStamp, Offset + TotalSz,
CrcSkipCount + 1})
end;
fold_keys_int_loop(_Bytes, _Fun, Acc, Consumed, Args) ->
{more, Acc, Consumed, Args}.
fold_hintfile(State, Fun, Acc0) ->
HintFile = hintfile_name(State),
case bitcask_io:file_open(HintFile, [readonly, read_ahead]) of
{ok, HintFd} ->
try
{ok, DataI} = read_file_info(State#filestate.filename),
DataSize = DataI#file_info.size,
case fold_file_loop(HintFd, hint, fun fold_hintfile_loop/5, Fun,
Acc0, {DataSize, HintFile}) of
{error, Reason} ->
{error, Reason};
Acc ->
Acc
end
after
bitcask_io:file_close(HintFd)
end;
{error, Reason} ->
{error, {fold_hintfile, Reason}}
end.
%% conditional end match here, checking that we get the expected CRC-containing
%% 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,
_TombInt:?TOMBSTONEFIELD_V2, (?MAXOFFSET_V2):?OFFSETFIELD_V2>>,
_Fun, Acc, Consumed, _Args) ->
{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,
TombInt:?TOMBSTONEFIELD_V2, Offset:?OFFSETFIELD_V2,
Key:KeySz/bytes, Rest/binary>>,
Fun, Acc0, Consumed0, {DataSize, HintFile} = Args) ->
case Offset + TotalSz =< DataSize + 1 of
true ->
PosInfo = {Offset, TotalSz},
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);
false ->
error_logger:warning_msg("Hintfile '~s' contains pointer ~p ~p "
"that is greater than total data size ~p\n",
[HintFile, Offset, TotalSz, DataSize]),
{error, {trunc_hintfile, Acc0}}
end;
%% catchall case where we don't get enough bytes from fold_file_loop
fold_hintfile_loop(_Bytes, _Fun, Acc0, Consumed0, Args) ->
{more, Acc0, Consumed0, Args}.
%% @doc scaffolding for faster folds over large files.
%% The somewhat tricky thing here is the FoldFn, which is a /6
%% that does all the actual work. see fold_hintfile_loop as a
%% commented example
-spec fold_file_loop(port(), atom(),
fun((binary(), fun(), any(), integer(),
any()) ->
{more, any(), integer(), any()} |
{done, any()} |
{done, any(), integer()} |
{skip, any(), integer(), any()} |
{error, any()}),
fun(), any(), any()) ->
{error, any()} | any().
fold_file_loop(Fd, Type, FoldFn, IntFoldFn, Acc, Args) ->
fold_file_loop(Fd, Type, FoldFn, IntFoldFn, Acc, Args, none, ?CHUNK_SIZE).
fold_file_loop(Fd, Type, FoldFn, IntFoldFn, Acc0, Args0, Prev0, ChunkSz0) ->
%% analyze what happened in the last loop to determine whether or
%% not to change the read size. This is an optimization for large values
%% in datafile folds and key folds
{Prev, ChunkSz}
= case Prev0 of
none -> {<<>>, ChunkSz0};
Other ->
CS = case byte_size(Other) of
%% to avoid having to rescan the same
%% binaries over and over again.
N when N >= ?MAX_CHUNK_SIZE ->
?MAX_CHUNK_SIZE;
N when N > ChunkSz0 ->
ChunkSz0 * 2;
_ -> ChunkSz0
end,
{Other, CS}
end,
case bitcask_io:file_read(Fd, ChunkSz) of
{ok, <<Bytes0/binary>>} ->
Bytes = <<Prev/binary, Bytes0/binary>>,
case FoldFn(Bytes, IntFoldFn, Acc0, 0, Args0) of
%% foldfuns should return more when they don't have enough
%% bytes to satisfy their main binary match.
{more, Acc, Consumed, Args} ->
Rest =
case Consumed > byte_size(Bytes) of
true -> <<>>;
false ->
<<_:Consumed/bytes, R/binary>> = Bytes,
R
end,
fold_file_loop(Fd, Type, FoldFn, IntFoldFn,
Acc, Args, Rest, ChunkSz);
%% the done two tuple is returned when we want to be
%% unconditionally successfully finished,
%% i.e. trailing data is a non-fatal error
{done, Acc} ->
Acc;
%% three tuple done requires full consumption of all
%% bytes given to the internal fold function, to
%% satisfy the pre-existing semantics of hintfile
%% folds.
{done, Acc, Consumed} ->
case Consumed =:= byte_size(Bytes) of
true -> Acc;
false ->
{error, {partial_fold, Consumed, Bytes, Bytes0}}
end;
{error, Reason} ->
{error, Reason}
end;
eof ->
%% when we reach the end of the file, if it's a hintfile,
%% we need to make sure that require_hint_crc is honored
%% (or not as the case may be).
case Prev == <<>> andalso Type == hint of
false ->
Acc0;
true ->
case application:get_env(bitcask, require_hint_crc) of
{ok, true} ->
{error, {incomplete_hint, 4}};
_ ->
Acc0
end
end;
{error, Reason} ->
{error, Reason}
end.
open_hint_file(Filename, FinalOpts) ->
open_hint_file(Filename, FinalOpts, 10).
open_hint_file(_Filename, _FinalOpts, 0) ->
throw(couldnt_open_hintfile);
open_hint_file(Filename, FinalOpts, Count) ->
case bitcask_io:file_open(hintfile_name(Filename), FinalOpts) of
{ok, FD} ->
FD;
{error, eexist} ->
timer:sleep(50),
open_hint_file(Filename, FinalOpts, Count - 1)
end.
hintfile_entry(Key, Tstamp, TombInt, Offset, TotalSz) ->
KeySz = size(Key),
[<<Tstamp:?TSTAMPFIELD, KeySz:?KEYSIZEFIELD, TotalSz:?TOTALSIZEFIELD,
TombInt:?TOMBSTONEFIELD_V2, Offset:?OFFSETFIELD_V2>>, Key].
%% ===================================================================
%% file/filelib avoidance code.
%% ===================================================================
read_file_info(FileName) ->
prim_file:read_file_info(FileName).
write_file_info(FileName, Info) ->
prim_file:write_file_info(FileName, Info).
is_file(File) ->
case read_file_info(File) of
{ok, #file_info{type=regular}} ->
true;
{ok, #file_info{type=directory}} ->
true;
_ ->
false
end.
is_dir(File) ->
case read_file_info(File) of
{ok, #file_info{type=directory}} ->
true;
_ ->
false
end.
ensure_dir("/") ->
ok;
ensure_dir(F) ->
Dir = filename:dirname(F),
case is_dir(Dir) of
true ->
ok;
false when Dir =:= F ->
%% Protect against infinite loop
{error,einval};
false ->
_ = ensure_dir(Dir),
%% this is rare enough that the serialization
%% isn't super important, maybe
case file:make_dir(Dir) of
{error,eexist}=EExist ->
case is_dir(Dir) of
true ->
ok;
false ->
EExist
end;
Err ->
Err
end
end.
list_dir(Dir) ->
list_dir(Dir, 1).
list_dir(_, 0) ->
{error, efile_driver_unavailable};
list_dir(Directory, Retries) when is_integer(Retries), Retries > 0 ->
Port = get_efile_port(),
case prim_file:list_dir(Port, Directory) of
{error, einval} ->
clear_efile_port(),
list_dir(Directory, Retries-1);
Result ->
Result
end.
get_efile_port() ->
Key = bitcask_efile_port,
case get(Key) of
undefined ->
case prim_file_drv_open("efile", [binary]) of
{ok, Port} ->
put(Key, Port),
get_efile_port();
Err ->
error_logger:error_msg("get_efile_port: ~p\n", [Err]),
timer:sleep(1000),
get_efile_port()
end;
Port ->
Port
end.
clear_efile_port() ->
erase(bitcask_efile_port).
prim_file_drv_open(Driver, Portopts) ->
try erlang:open_port({spawn, Driver}, Portopts) of
Port ->
{ok, Port}
catch
error:Reason ->
{error, Reason}
end.