WIP: more bugs fixed, closer to working again...
This commit is contained in:
parent
102c518269
commit
4bcd4b5de1
4 changed files with 66 additions and 59 deletions
|
@ -41,7 +41,11 @@
|
||||||
-include_lib("include/hanoidb.hrl").
|
-include_lib("include/hanoidb.hrl").
|
||||||
-include_lib("include/plain_rpc.hrl").
|
-include_lib("include/plain_rpc.hrl").
|
||||||
|
|
||||||
-record(state, { top, nursery, dir, opt, max_level }).
|
-record(state, { top :: pos_integer(),
|
||||||
|
nursery :: term(),
|
||||||
|
dir :: string(),
|
||||||
|
opt :: term(),
|
||||||
|
max_level :: pos_integer()}).
|
||||||
|
|
||||||
%% 0 means never expire
|
%% 0 means never expire
|
||||||
-define(DEFAULT_EXPIRY_SECS, 0).
|
-define(DEFAULT_EXPIRY_SECS, 0).
|
||||||
|
@ -249,24 +253,27 @@ drain_worker_and_return(MRef, PID, Value) ->
|
||||||
|
|
||||||
init([Dir, Opts0]) ->
|
init([Dir, Opts0]) ->
|
||||||
%% ensure expory_secs option is set in config
|
%% ensure expory_secs option is set in config
|
||||||
|
Opts =
|
||||||
case get_opt(expiry_secs, Opts0) of
|
case get_opt(expiry_secs, Opts0) of
|
||||||
undefined ->
|
undefined ->
|
||||||
Opts = [{expiry_secs, ?DEFAULT_EXPIRY_SECS}|Opts0];
|
[{expiry_secs, ?DEFAULT_EXPIRY_SECS}|Opts0];
|
||||||
N when is_integer(N), N >= 0 ->
|
N when is_integer(N), N >= 0 ->
|
||||||
Opts = [{expiry_secs, N}|Opts0]
|
[{expiry_secs, N}|Opts0]
|
||||||
end,
|
end,
|
||||||
|
|
||||||
hanoidb_util:ensure_expiry(Opts),
|
hanoidb_util:ensure_expiry(Opts),
|
||||||
|
|
||||||
|
{Nursery, MaxLevel, TopLevel} =
|
||||||
case file:read_file_info(Dir) of
|
case file:read_file_info(Dir) of
|
||||||
{ok, #file_info{ type=directory }} ->
|
{ok, #file_info{ type=directory }} ->
|
||||||
{ok, TopLevel, MaxLevel} = open_levels(Dir, Opts),
|
{ok, TL, ML} = open_levels(Dir, Opts),
|
||||||
{ok, Nursery} = hanoidb_nursery:recover(Dir, TopLevel, MaxLevel, Opts);
|
{ok, N0} = hanoidb_nursery:recover(Dir, TL, ML, Opts),
|
||||||
|
{N0, ML, TL};
|
||||||
{error, E} when E =:= enoent ->
|
{error, E} when E =:= enoent ->
|
||||||
ok = file:make_dir(Dir),
|
ok = file:make_dir(Dir),
|
||||||
{ok, TopLevel} = hanoidb_level:open(Dir, ?TOP_LEVEL, undefined, Opts, self()),
|
{ok, TL} = hanoidb_level:open(Dir, ?TOP_LEVEL, undefined, Opts, self()),
|
||||||
MaxLevel = ?TOP_LEVEL,
|
ML = ?TOP_LEVEL,
|
||||||
{ok, Nursery} = hanoidb_nursery:new(Dir, MaxLevel, Opts)
|
{ok, N0} = hanoidb_nursery:new(Dir, ML, Opts),
|
||||||
|
{N0, ML, TL}
|
||||||
end,
|
end,
|
||||||
{ok, #state{ top=TopLevel, dir=Dir, nursery=Nursery, opt=Opts, max_level=MaxLevel }}.
|
{ok, #state{ top=TopLevel, dir=Dir, nursery=Nursery, opt=Opts, max_level=MaxLevel }}.
|
||||||
|
|
||||||
|
|
|
@ -79,9 +79,9 @@ open(Name, Config) ->
|
||||||
%% read root position
|
%% read root position
|
||||||
{ok, <<RootPos:64/unsigned>>} = file:pread(File, FileInfo#file_info.size - 8, 8),
|
{ok, <<RootPos:64/unsigned>>} = file:pread(File, FileInfo#file_info.size - 8, 8),
|
||||||
{ok, <<BloomSize:32/unsigned>>} = file:pread(File, FileInfo#file_info.size - 12, 4),
|
{ok, <<BloomSize:32/unsigned>>} = file:pread(File, FileInfo#file_info.size - 12, 4),
|
||||||
{ok, BloomData} = file:pread(File, FileInfo#file_info.size-12-BloomSize ,BloomSize),
|
{ok, BloomData} = file:pread(File, (FileInfo#file_info.size - 12 - BloomSize), BloomSize),
|
||||||
|
|
||||||
{ok, Bloom} = binary_to_term(BloomData),
|
{ok, Bloom} = hanoidb_util:decode_bloom(BloomData),
|
||||||
|
|
||||||
%% suck in the root
|
%% suck in the root
|
||||||
{ok, Root} = read_node(File, RootPos),
|
{ok, Root} = read_node(File, RootPos),
|
||||||
|
|
|
@ -205,6 +205,15 @@ decode_kv_data(<<?TAG_TRANSACT, Rest/binary>>) ->
|
||||||
{ok, TX} = decode_crc_data(Rest, [], []),
|
{ok, TX} = decode_crc_data(Rest, [], []),
|
||||||
TX.
|
TX.
|
||||||
|
|
||||||
|
encode_bloom(Bloom) ->
|
||||||
|
case bloom:is_bloom(Bloom) of
|
||||||
|
true -> zlib:gzip(term_to_binary(Bloom));
|
||||||
|
false -> <<>>
|
||||||
|
end.
|
||||||
|
|
||||||
|
decode_bloom(Bin) ->
|
||||||
|
binary_to_term(zlib:gunzip(Bin)).
|
||||||
|
|
||||||
%% @doc Return number of seconds since 1970
|
%% @doc Return number of seconds since 1970
|
||||||
-spec tstamp() -> pos_integer().
|
-spec tstamp() -> pos_integer().
|
||||||
tstamp() ->
|
tstamp() ->
|
||||||
|
|
|
@ -94,11 +94,11 @@ init([Name, Options]) ->
|
||||||
case do_open(Name, Options, [exclusive]) of
|
case do_open(Name, Options, [exclusive]) of
|
||||||
{ok, IdxFile} ->
|
{ok, IdxFile} ->
|
||||||
file:write(IdxFile, ?FILE_FORMAT),
|
file:write(IdxFile, ?FILE_FORMAT),
|
||||||
BloomFilter = bloom:new(erlang:min(Size, 16#ffffffff), 0.01),
|
Bloom = bloom:new(erlang:min(Size, 16#ffffffff), 0.01),
|
||||||
BlockSize = hanoidb:get_opt(block_size, Options, ?NODE_SIZE),
|
BlockSize = hanoidb:get_opt(block_size, Options, ?NODE_SIZE),
|
||||||
{ok, #state{ name=Name,
|
{ok, #state{ name=Name,
|
||||||
index_file_pos=?FIRST_BLOCK_POS, index_file=IdxFile,
|
index_file_pos=?FIRST_BLOCK_POS, index_file=IdxFile,
|
||||||
bloom = BloomFilter,
|
bloom = Bloom,
|
||||||
block_size = BlockSize,
|
block_size = BlockSize,
|
||||||
compress = hanoidb:get_opt(compress, Options, none),
|
compress = hanoidb:get_opt(compress, Options, none),
|
||||||
opts = Options
|
opts = Options
|
||||||
|
@ -152,21 +152,18 @@ code_change(_OldVsn, State, _Extra) ->
|
||||||
% io:format("serializing ~p @ ~p~n", [State#state.name,
|
% io:format("serializing ~p @ ~p~n", [State#state.name,
|
||||||
% State#state.index_file_pos]),
|
% State#state.index_file_pos]),
|
||||||
serialize(#state{ bloom=Bloom, index_file=File, index_file_pos=Position }=State) ->
|
serialize(#state{ bloom=Bloom, index_file=File, index_file_pos=Position }=State) ->
|
||||||
|
|
||||||
%% assert that we're on track
|
|
||||||
case file:position(File, {eof, 0}) of
|
case file:position(File, {eof, 0}) of
|
||||||
{ok, Position} ->
|
{ok, Position} ->
|
||||||
ok;
|
ok;
|
||||||
{ok, WrongPosition} ->
|
{ok, WrongPosition} ->
|
||||||
exit({bad_position, Position, WrongPosition})
|
exit({bad_position, Position, WrongPosition})
|
||||||
end,
|
end,
|
||||||
|
|
||||||
ok = file:close(File),
|
ok = file:close(File),
|
||||||
erlang:term_to_binary( { State#state{ index_file=closed }, term_to_binary(Bloom) } ).
|
erlang:term_to_binary( { State#state{ index_file=closed }, hanoidb_util:encode_bloom(Bloom) } ).
|
||||||
|
|
||||||
deserialize(Binary) ->
|
deserialize(Binary) ->
|
||||||
{State, BinBloom} = erlang:binary_to_term(Binary),
|
{State, Bin} = erlang:binary_to_term(Binary),
|
||||||
{ok, Bloom} = term_to_binary(BinBloom),
|
Bloom = hanoidb_util:decode_bloom(Bin),
|
||||||
{ok, IdxFile} = do_open(State#state.name, State#state.opts, []),
|
{ok, IdxFile} = do_open(State#state.name, State#state.opts, []),
|
||||||
State#state{ bloom=Bloom, index_file=IdxFile }.
|
State#state{ bloom=Bloom, index_file=IdxFile }.
|
||||||
|
|
||||||
|
@ -178,10 +175,10 @@ do_open(Name, Options, OpenOpts) ->
|
||||||
|
|
||||||
|
|
||||||
%% @doc flush pending nodes and write trailer
|
%% @doc flush pending nodes and write trailer
|
||||||
|
flush_nodes(State=#state{ nodes=[#node{level=N, members=[{_,{Pos,_Len}}]}], last_node_pos=Pos }) when N>0 ->
|
||||||
|
%% stack consists of one node with one {pos,len} member. Just ignore this node.
|
||||||
|
flush_nodes(State#state{ nodes=[] });
|
||||||
flush_nodes(#state{ nodes=[], last_node_pos=LastNodePos, last_node_size=_LastNodeSize, bloom=Bloom }=State) ->
|
flush_nodes(#state{ nodes=[], last_node_pos=LastNodePos, last_node_size=_LastNodeSize, bloom=Bloom }=State) ->
|
||||||
BloomBin = term_to_binary(Bloom, [compressed]),
|
|
||||||
BloomSize = byte_size(BloomBin),
|
|
||||||
|
|
||||||
IdxFile = State#state.index_file,
|
IdxFile = State#state.index_file,
|
||||||
|
|
||||||
RootPos =
|
RootPos =
|
||||||
|
@ -194,6 +191,8 @@ flush_nodes(#state{ nodes=[], last_node_pos=LastNodePos, last_node_size=_LastNod
|
||||||
?FIRST_BLOCK_POS
|
?FIRST_BLOCK_POS
|
||||||
end,
|
end,
|
||||||
|
|
||||||
|
BloomBin = hanoidb_util:encode_bloom(Bloom),
|
||||||
|
BloomSize = byte_size(BloomBin),
|
||||||
Trailer = << 0:32, BloomBin/binary, BloomSize:32/unsigned, RootPos:64/unsigned >>,
|
Trailer = << 0:32, BloomBin/binary, BloomSize:32/unsigned, RootPos:64/unsigned >>,
|
||||||
|
|
||||||
ok = file:write(IdxFile, Trailer),
|
ok = file:write(IdxFile, Trailer),
|
||||||
|
@ -201,18 +200,17 @@ flush_nodes(#state{ nodes=[], last_node_pos=LastNodePos, last_node_size=_LastNod
|
||||||
ok = file:close(IdxFile),
|
ok = file:close(IdxFile),
|
||||||
|
|
||||||
{ok, State#state{ index_file=undefined, index_file_pos=undefined }};
|
{ok, State#state{ index_file=undefined, index_file_pos=undefined }};
|
||||||
|
|
||||||
%% stack consists of one node with one {pos,len} member. Just ignore this node.
|
|
||||||
flush_nodes(State=#state{ nodes=[#node{level=N, members=[{_,{Pos,_Len}}]}], last_node_pos=Pos }) when N>0 ->
|
|
||||||
flush_nodes(State#state{ nodes=[] });
|
|
||||||
|
|
||||||
flush_nodes(State) ->
|
flush_nodes(State) ->
|
||||||
{ok, State2} = close_node(State),
|
{ok, State2} = close_node(State),
|
||||||
flush_nodes(State2).
|
flush_nodes(State2).
|
||||||
|
|
||||||
add_record(Level, Key, Value,
|
|
||||||
#state{ nodes=[ #node{level=Level, members=List, size=NodeSize}=CurrNode | RestNodes ],
|
add_record(Level, Key, Value, State=#state{ nodes=[] }) ->
|
||||||
value_count=VC, tombstone_count=TC }=State) ->
|
add_record(Level, Key, Value, State#state{ nodes=[ #node{ level=Level } ] });
|
||||||
|
add_record(Level, Key, Value, State=#state{ nodes=[ #node{level=Level2 } |_]=Stack })
|
||||||
|
when Level < Level2 ->
|
||||||
|
add_record(Level, Key, Value, State#state{ nodes=[ #node{ level=(Level2 - 1) } | Stack] });
|
||||||
|
add_record(Level, Key, Value, #state{ nodes=[ #node{level=Level, members=List, size=NodeSize}=CurrNode | RestNodes ], value_count=VC, tombstone_count=TC }=State) ->
|
||||||
%% The top-of-stack node is at the level we wish to insert at.
|
%% The top-of-stack node is at the level we wish to insert at.
|
||||||
|
|
||||||
%% Assert that keys are increasing:
|
%% Assert that keys are increasing:
|
||||||
|
@ -236,10 +234,8 @@ add_record(Level, Key, Value,
|
||||||
end,
|
end,
|
||||||
|
|
||||||
{TC1, VC1} =
|
{TC1, VC1} =
|
||||||
case Level == 0 of
|
case Level of
|
||||||
true ->
|
0 ->
|
||||||
{TC, VC};
|
|
||||||
false ->
|
|
||||||
case Value of
|
case Value of
|
||||||
?TOMBSTONE ->
|
?TOMBSTONE ->
|
||||||
{TC+1, VC};
|
{TC+1, VC};
|
||||||
|
@ -247,7 +243,9 @@ add_record(Level, Key, Value,
|
||||||
{TC+1, VC};
|
{TC+1, VC};
|
||||||
_ ->
|
_ ->
|
||||||
{TC, VC+1}
|
{TC, VC+1}
|
||||||
end
|
end;
|
||||||
|
_ ->
|
||||||
|
{TC, VC}
|
||||||
end,
|
end,
|
||||||
|
|
||||||
NodeMembers = [{Key, Value} | List],
|
NodeMembers = [{Key, Value} | List],
|
||||||
|
@ -256,17 +254,10 @@ add_record(Level, Key, Value,
|
||||||
|
|
||||||
case NewSize >= State#state.block_size of
|
case NewSize >= State#state.block_size of
|
||||||
true ->
|
true ->
|
||||||
{ok, State2};
|
close_node(State2);
|
||||||
false ->
|
false ->
|
||||||
close_node(State2)
|
{ok, State2}
|
||||||
end;
|
end.
|
||||||
|
|
||||||
add_record(Level, Key, Value, State=#state{ nodes=[] }) ->
|
|
||||||
add_record(Level, Key, Value, State#state{ nodes=[ #node{ level=Level } ] });
|
|
||||||
|
|
||||||
add_record(Level, Key, Value, State=#state{ nodes=[ #node{level=Level2 } |_]=Stack }) when Level < Level2 ->
|
|
||||||
add_record(Level, Key, Value, State#state{ nodes=[ #node{ level=(Level2-1) } | Stack] }).
|
|
||||||
|
|
||||||
|
|
||||||
close_node(#state{nodes=[#node{ level=Level, members=NodeMembers }|RestNodes], compress=Compress} = State) ->
|
close_node(#state{nodes=[#node{ level=Level, members=NodeMembers }|RestNodes], compress=Compress} = State) ->
|
||||||
OrderedMembers = lists:reverse(NodeMembers),
|
OrderedMembers = lists:reverse(NodeMembers),
|
||||||
|
|
Loading…
Reference in a new issue