WIP: tango_oid refactoring, all broken: infinite loop

This commit is contained in:
Scott Lystig Fritchie 2014-09-20 22:36:21 +09:00
parent 9a3ac02413
commit a0bb7ee23d
3 changed files with 33 additions and 13 deletions

View file

@ -38,28 +38,28 @@
%% TODO: for version 2: add strong checksum %% TODO: for version 2: add strong checksum
pack_v1(StreamList, Options, Page, PageSize) pack_v1(Stream_BackPs, Options, Page, PageSize)
when is_list(StreamList), is_list(Options), is_binary(Page), when is_list(Stream_BackPs), is_list(Options), is_binary(Page),
is_integer(PageSize), PageSize > 0 -> is_integer(PageSize), PageSize > 0 ->
StreamListBin = term_to_binary(StreamList), Stream_BackPsBin = term_to_binary(Stream_BackPs),
StreamListSize = byte_size(StreamListBin), Stream_BackPsSize = byte_size(Stream_BackPsBin),
OptionsInt = convert_options_list2int(Options), OptionsInt = convert_options_list2int(Options),
PageActualSize = byte_size(Page), PageActualSize = byte_size(Page),
pad_bin(PageSize, pad_bin(PageSize,
list_to_binary([<<?MAGIC_NUMBER_V1:32/big>>, list_to_binary([<<?MAGIC_NUMBER_V1:32/big>>,
<<OptionsInt:8/big>>, <<OptionsInt:8/big>>,
<<StreamListSize:16/big>>, <<Stream_BackPsSize:16/big>>,
StreamListBin, Stream_BackPsBin,
<<PageActualSize:16/big>>, <<PageActualSize:16/big>>,
Page])). Page])).
unpack_v1(<<?MAGIC_NUMBER_V1:32/big, unpack_v1(<<?MAGIC_NUMBER_V1:32/big,
_Options:8/big, _Options:8/big,
StreamListSize:16/big, StreamListBin:StreamListSize/binary, Stream_BackPsSize:16/big, Stream_BackPsBin:Stream_BackPsSize/binary,
PageActualSize:16/big, Page:PageActualSize/binary, PageActualSize:16/big, Page:PageActualSize/binary,
_/binary>>, Part) -> _/binary>>, Part) ->
if Part == stream_list -> if Part == stream_list ->
binary_to_term(StreamListBin); binary_to_term(Stream_BackPsBin);
Part == page -> Part == page ->
Page Page
end. end.
@ -102,11 +102,12 @@ scan_backward2(_Proj, _Stream, LastLPN, StopAtLPN, _NumPages, _WithPagesP)
when LastLPN =< StopAtLPN; LastLPN =< 0 -> when LastLPN =< StopAtLPN; LastLPN =< 0 ->
[]; [];
scan_backward2(Proj, Stream, LastLPN, StopAtLPN, NumPages, WithPagesP) -> scan_backward2(Proj, Stream, LastLPN, StopAtLPN, NumPages, WithPagesP) ->
%% ?D({scan, lastlpn, LastLPN}), ?D({scan, Stream, LastLPN}),
case corfurl:read_page(Proj, LastLPN) of case corfurl:read_page(Proj, LastLPN) of
{ok, FullPage} -> {ok, FullPage} ->
%% ?D({scan, LastLPN, ok}), ?D({scan, LastLPN, ok}),
%% ?D({scan, Stream, unpack_v1(FullPage, stream_list)}), ?D({scan, Stream, unpack_v1(FullPage, stream_list)}),
?D(proplists:get_value(Stream, unpack_v1(FullPage, stream_list))),
case proplists:get_value(Stream, unpack_v1(FullPage, stream_list)) of case proplists:get_value(Stream, unpack_v1(FullPage, stream_list)) of
undefined -> undefined ->
if NumPages == 0 -> if NumPages == 0 ->
@ -123,11 +124,14 @@ scan_backward2(Proj, Stream, LastLPN, StopAtLPN, NumPages, WithPagesP) ->
end; end;
[] -> [] ->
if WithPagesP -> if WithPagesP ->
?D(?LINE),
[{LastLPN, unpack_v1(FullPage, page)}]; [{LastLPN, unpack_v1(FullPage, page)}];
true -> true ->
?D(?LINE),
[LastLPN] [LastLPN]
end; end;
BackPs -> BackPs ->
?D(?LINE),
if WithPagesP -> if WithPagesP ->
%% ?D({bummer, BackPs}), %% ?D({bummer, BackPs}),
[{LastLPN, unpack_v1(FullPage, page)}| [{LastLPN, unpack_v1(FullPage, page)}|
@ -163,7 +167,8 @@ append_page(#proj{seq={Sequencer,_,_}, page_size=PageSize} = Proj,
{ok, LPN, BackPsList} = corfurl_sequencer:get_tails(Sequencer, 1, {ok, LPN, BackPsList} = corfurl_sequencer:get_tails(Sequencer, 1,
StreamList), StreamList),
%% pulse_tracing_add(write, LPN), %% pulse_tracing_add(write, LPN),
Page = tango:pack_v1(StreamList, [to_final_page], StreamBackPs = lists:zip(StreamList, BackPsList),
Page = tango:pack_v1(StreamBackPs, [to_final_page],
OrigPage, PageSize), OrigPage, PageSize),
append_page1(Proj, LPN, Page, StreamList, 5, OrigPage) append_page1(Proj, LPN, Page, StreamList, 5, OrigPage)
catch catch

View file

@ -39,7 +39,7 @@
-define(LONG_TIME, 30*1000). -define(LONG_TIME, 30*1000).
-define(D(X), io:format(user, "Dbg: ~s =\n ~p\n", [??X, X])). -define(D(X), io:format(user, "Dbg: ~s = ~p\n", [??X, X])).
-type lpn() :: non_neg_integer(). -type lpn() :: non_neg_integer().
@ -143,6 +143,7 @@ fetch_unread_pages(Proj, LastLPN, StopAtLPN)
%% ?D({fetch_unread_pages, LastLPN, StopAtLPN}), %% ?D({fetch_unread_pages, LastLPN, StopAtLPN}),
LPNandPages = tango:scan_backward(Proj, ?OID_STREAM_NUMBER, LastLPN, LPNandPages = tango:scan_backward(Proj, ?OID_STREAM_NUMBER, LastLPN,
StopAtLPN, true), StopAtLPN, true),
?D(LPNandPages),
{_LPNs, _Pages} = lists:unzip(LPNandPages). {_LPNs, _Pages} = lists:unzip(LPNandPages).
play_log_pages(Pages, SideEffectsP, play_log_pages(Pages, SideEffectsP,
@ -155,8 +156,11 @@ play_log_pages(Pages, I_State, CallbackMod, SideEffectsP) ->
roll_log_forward(#state{seq=SequencerPid, proj=Proj, all_back_ps=BackPs, roll_log_forward(#state{seq=SequencerPid, proj=Proj, all_back_ps=BackPs,
last_fetch_lpn=StopAtLPN} = State) -> last_fetch_lpn=StopAtLPN} = State) ->
?D(StopAtLPN),
LastLPN = find_last_lpn(SequencerPid), LastLPN = find_last_lpn(SequencerPid),
?D(LastLPN),
{LPNs, Pages} = fetch_unread_pages(Proj, LastLPN, StopAtLPN), {LPNs, Pages} = fetch_unread_pages(Proj, LastLPN, StopAtLPN),
?D(LPNs),
NewBPs = append_lpns(LPNs, BackPs), NewBPs = append_lpns(LPNs, BackPs),
play_log_pages(Pages, true, State#state{all_back_ps=NewBPs}). play_log_pages(Pages, true, State#state{all_back_ps=NewBPs}).
@ -193,15 +197,19 @@ play_log_mutate_i_state(Pages, SideEffectsP, I_State) ->
{Res, O2} = {Res, O2} =
case ?DICTMOD:find(Key, Dict) of case ?DICTMOD:find(Key, Dict) of
error -> error ->
?D(?LINE),
Dict2 = ?DICTMOD:store(Key, Next, Dict), Dict2 = ?DICTMOD:store(Key, Next, Dict),
{{ok, Next},O#oid_map{map=Dict2, {{ok, Next},O#oid_map{map=Dict2,
next=Next + 1}}; next=Next + 1}};
{ok, _} -> {ok, _} ->
?D(?LINE),
{already_exists, O} {already_exists, O}
end, end,
if SideEffectsP -> if SideEffectsP ->
?D(?LINE),
gen_server:reply(From, Res); gen_server:reply(From, Res);
true -> true ->
?D(?LINE),
ok ok
end, end,
O2 O2

View file

@ -28,6 +28,8 @@
-endif. -endif.
-endif. -endif.
-define(D(X), io:format(user, "Dbg: ~s = ~p\n", [??X, X])).
-ifdef(TEST). -ifdef(TEST).
-ifndef(PULSE). -ifndef(PULSE).
@ -52,8 +54,13 @@ tango_oid_one_test_int(PageSize, Seq, Proj) ->
K1 = foo, K1 = foo,
K2 = bar, K2 = bar,
OID_Num1 = 1, OID_Num1 = 1,
error = tango_oid:get(OID_Map, "does not exist"),
?D(?LINE),
{ok, OID_Num1} = tango_oid:new(OID_Map, K1), {ok, OID_Num1} = tango_oid:new(OID_Map, K1),
?D(?LINE),
{ok, OID_Num1} = tango_oid:get(OID_Map, K1), {ok, OID_Num1} = tango_oid:get(OID_Map, K1),
?D(?LINE),
already_exists = tango_oid:new(OID_Map, K1), already_exists = tango_oid:new(OID_Map, K1),
%% The V2 put should *not* have clobbered the previous value %% The V2 put should *not* have clobbered the previous value
{ok, OID_Num1} = tango_oid:get(OID_Map, K1), {ok, OID_Num1} = tango_oid:get(OID_Map, K1),