mnemosyne_op.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 604 行 · 第 1/2 页

call_recursive_op(NextPid, Goal) ->
    call_recursive_op(NextPid, Goal, [], [], {[],[], []}).

call_recursive_op(NextPid, Goal, BssAcc, Acc, {Tid, PidL, EndC}) ->
    receive
	{bss, Bss, Max, EndCntrl, Stack, LastMarker} ->
	    NewEndC = update_endc (PidL, EndC, LastMarker),

	    {BssAcc1, SolutionList} =
		get_from_slg(Max, Acc, lists:append(BssAcc,Bss), Goal),
	    {First,Acc1} = split_list(Max, SolutionList),
	    case {Acc1, check_lastmarker (NewEndC, Goal#pred_sym.rec_count, LastMarker)} of
		{[], true} ->
		    NextPid ! {bss, First, Max, EndCntrl, Stack, 
			       {true, self ()}};
		_ ->
		    NextPid ! {bss, First, Max, EndCntrl, Stack, 
			       false}
	    end,
	    call_recursive_op(NextPid, Goal, BssAcc1, Acc1, 
			      {Tid, PidL, NewEndC});

	{fail, SomePid, Cause} ->
	    NextPid ! {fail, self(), Cause},
	    call_recursive_op(NextPid, Goal);

	{tid, NTid, Pid} ->
	    mnesia:put_activity_id(NTid),
	    NextPid ! {tid,NTid, self()},
	    call_recursive_op(NextPid, Goal, [], [], 
			      add_tid (Tid,PidL,EndC, NTid,Pid))
    end.

    

get_from_slg(Max, Acc, [Bs|Bss], Goal) when length(Acc)<Max ->
    GoalInst = mnemosyne_unify:instantiate(Goal,Bs),
    mnemosyne_slg:slg_init(),
    Acc1 = make_bindings(mnemosyne_slg:slg(GoalInst,none), Goal, Bs, Acc),
    get_from_slg(Max, Acc1, Bss, Goal);
get_from_slg(Max, Acc, Bss, Goal) ->
    {Bss, Acc}.


make_bindings([Answ|Answs], Goal, Bs, Acc) ->
    case mnemosyne_unify:unify(Answ,Goal,Bs) of
	fail -> make_bindings(Answs, Goal, Bs, Acc);
	NewBs -> make_bindings(Answs, Goal, Bs, [NewBs|Acc])
    end;
make_bindings([], Goal, Bs, Acc) ->
    Acc.


%%%================================================================
%%% 		Private

table_loop(NextPid, F, Acc, {Tid, PidL, EndC}) ->
    receive
	{bss, Bss, Max, EndCntrl, Stack, LastMarker} ->
?debugmsg(3,"~w got ~w\n", [self(), {bss, Bss, Max, EndCntrl, Stack}]),
	    NewEndC = update_endc (PidL, EndC, LastMarker),

	    List = lists:append(Acc#acc.bss,Bss),

	    case catch fact_produce(Max, Acc#acc{bss = List}, F) of
		{'EXIT', Cause} ->
		    NextPid ! {fail, self(), Cause},
		    table(NextPid, F);
		Acc1 ->
		    {First, Sols} = split_list(Max, Acc1#acc.sols),
		    case {Sols, check_lastmarker(NewEndC, F#pred_sym.rec_count, LastMarker)} of
			{[], true} ->
			    NextPid ! {bss, First, Max, EndCntrl, Stack, 
				       {true, self()}};
			_ ->
			    NextPid ! {bss, First, Max, EndCntrl, Stack, 
				       false}
		    end,
		    ?debugmsg(3,"~w: ~w ! ~w\n", 
			      [self(),NextPid, {bss, First, Max, EndCntrl, Stack}]),
		    table_loop(NextPid, F, Acc1#acc{sols = Sols}, 
			       {Tid, PidL, NewEndC})
	    end;
	
	{fail, SomePid, Cause} ->
	    NextPid ! {fail, self(), Cause},
	    table (NextPid, F);
	
	{tid, NTid, Pid} ->
	    mnesia:put_activity_id(NTid),
	    NextPid ! {tid,NTid, self ()},
	    table_loop(NextPid, F, #acc{}, add_tid(Tid,PidL,EndC,NTid,Pid))
    end.


%% add_tid called because a '{tid, Tid,Pid}' message arrived.
%% Tid is the new tid to use. If that Tid is a new one
%% it is a new transaction.
%% If it the same Tid as before then either we got it before
%% from another Pid (sender) --> add Pid to reference list
%% or we got it from a pid we already have it from
%% --> restart of question (using same pid)

%% Comments from dgud: The earlier version used a counter
%% to count the number of answers, due to a bug where the 
%% complicated loop failed(never returned)  when there was 
%% more than 1000 (i.e. prefetched) answers,
%% I rewrote it to use a decreasing PidList instead. So the 
%% various Count(s) variables is not used anymore.
%% This might be wrong I'm not 100% sure what I'm doing but 
%% it works much better now.

add_tid (Tid, PidL, EndC, Tid, Pid) ->
    case lists:member (Pid, PidL) of
	true ->		      % restart of trans with same tid
	    {Tid, [Pid], [Pid]};
	false ->	      % add one more pid to collect lastmarkers from
	    {Tid, [Pid | PidL], [Pid | PidL]}
    end;
add_tid (_, _, _, NTid, Pid) ->  % make new transaction
    {NTid, [Pid], [Pid]}.
    
update_endc (TidL, EndC, false) ->
    EndC;
update_endc (TidL, EndC, {true, Pid}) ->
    case lists:member(Pid, TidL) of
	true ->
	    lists:delete(Pid, EndC);
	_ ->   
	    EndC
    end.
check_lastmarker ([],BC,{true,_}) -> true;
check_lastmarker (AC,BC,Last) -> false.

%%%----------------
fact_produce(Max, Acc, F) ->
    {Sols,Bss} = fact_produce(Max, Acc#acc.sols, Acc#acc.bss, F),
    Acc#acc{bss=Bss, sols=Sols}.
    
fact_produce(Max, Sols, [Bs|Bss], F) when length(Sols)<Max ->
    fact_produce(Max, get_from_table(Bs,F,Sols), Bss, F);
fact_produce(Max, Sols, Bss, F) ->
    {Sols,Bss}.

get_from_table(Bs, F, Sols) ->
    InstPattern = mnemosyne_unify:instantiate(F#pred_sym.pattern, Bs),
    Table = F#pred_sym.functor,
    Matches = 
	case F#pred_sym.idx_method of
	    {dyn_idx,Pos} ->
		?not_yet('get_from_table: dyn_idx'),
		mnesia:match_object(Table, InstPattern, read);
	    
	    {stat_idx,Pos} ->
		mnesia:index_match_object(Table, InstPattern, Pos, read);

	    key ->
		mnesia:match_object(Table, InstPattern, read);

	    no_idx ->				% Occurs if no statistc optim
		mnesia:match_object(Table, InstPattern, read)
	end,
    NewBss = unify_list(Matches, F#pred_sym.back_pattern, Bs, Sols),
    ?debugmsg(4, "NewBss = ~s\n", 
	      [mnemosyne_pp:e(lists:map(fun(B)->{'#bindings',B} end,NewBss))]),
    NewBss.


%%%----------------------------------------------------------------
negation(NextPid, Count,HelpPid, Acc, {Tid,PidL,EndC}) ->
    receive
	%%---- From the ordinary flow
	{bss, Bss, Max, EndCntrl, Stack,LastMarker} ->
	    NewEndC = update_endc (PidL, EndC, LastMarker),
	    case check_lastmarker (NewEndC, Count, LastMarker) of
		true ->
		    HelpPid ! 
			{neg_bss_req, Bss, Max, EndCntrl, Stack, 
			 {true, self()}};
		_ ->
		    HelpPid ! 
			{neg_bss_req, Bss, Max, EndCntrl, Stack, 
			 {false, self()}}
	    end,
	    negation(NextPid, Count,HelpPid, Acc, {Tid,PidL,NewEndC});
		
	{fail, SomePid, Cause} ->
	    HelpPid ! {fail, self(), Cause},
	    NextPid ! {fail, self(), Cause},
	    negation(NextPid, Count, HelpPid);
	
	{tid, NTid,Pid} ->
	    HelpPid ! {tid_req, NTid, self()},
	    NextPid ! {tid, NTid, self()},
	    negation(NextPid, Count, HelpPid, [], add_tid(Tid,PidL,EndC,NTid,Pid));
	
	%%---- From the help process
	{neg_bss_repl, Bss, Max, EndCntrl, Stack, LastMarker} ->
	    NewLM = 
		case LastMarker of
		    {true, _} ->
			{true, self ()};
		    _ ->
			false
		end,
	    {First,Last} = split_list(Max, lists:append(Acc,Bss)),
	    NextPid ! {bss, First, Max, EndCntrl, Stack,NewLM},
	    negation(NextPid, Count, HelpPid, Last, {Tid,PidL,EndC})
	
    end.

    
negation_help(NextPid, Buddie, Acc0, {Tid,PidL, EndC}) ->
    receive
	%% Requests from the negation process ("Buddie")
	{neg_bss_req, Bss, Max, EndCntrl, Stack,LastMarker} ->
	    eval_neg_each_binding(Bss, NextPid, 
				  [{params,Max,EndCntrl}|Stack],
				  LastMarker),
	    negation_help(NextPid, Buddie, Acc0, {Tid, PidL, EndC});
	
	{tid_req,NTid,Pid} ->
%%	    mnesia:put_activity_id(Tid),
	    NextPid ! {tid,NTid,self()}, 
	    negation_help(NextPid, Buddie, [], {Tid, PidL, EndC});

	%% From the ordinary flow, that is, from the negated part
	{bss, Bss, all, [], [SendData,{params,Max,EndCntrl}|Stack],LastMarker} ->

	    NewEndC = update_endc (PidL, EndC, LastMarker),
	    case SendData of
		last when NewEndC == 0 ->
		    Buddie ! {neg_bss_repl, Acc0, Max, EndCntrl, 
			      Stack, {true, self()}},
		    negation_help(NextPid, Buddie, [], {Tid, PidL, NewEndC});
		{bs,BsOrig} ->
		    case Bss of
			[] ->
			    negation_help(NextPid,Buddie,[BsOrig|Acc0],
					 {Tid,PidL,NewEndC});
			[_|_] ->
			    negation_help(NextPid,Buddie,Acc0,
					 {Tid,PidL,NewEndC})
		    end
	    end;
	
	{fail, SomePid, Cause} ->
	    negation_help(NextPid, Buddie, [],{[],[],[]});
	
	{tid, NTid, Pid} ->
	    negation_help(NextPid, Buddie, Acc0, 
			  add_tid (Tid, PidL, EndC, NTid, Pid))
    end.
    


eval_neg_each_binding([Bs|Bss], NextPid, Stack, LastMarker) ->
    NextPid ! {bss, [Bs], all, [], [{bs,Bs}|Stack], false},
    eval_neg_each_binding(Bss, NextPid, Stack, LastMarker);
eval_neg_each_binding([], NextPid, Stack, LastMarker) ->
    NextPid ! {bss, [], all, [], [last|Stack], LastMarker}.

%%%----------------------------------------------------------------

bss_union(Bss, Bs0) ->
    bss_union(Bss, Bs0, []).

bss_union([Bs1|Bss], Bs0, Acc) ->
    case catch mnemosyne_unify:bs_union(Bs0,Bs1) of
	fail -> bss_union(Bss, Bs0, Acc);
	Bs -> bss_union(Bss, Bs0, [Bs|Acc])
    end;
bss_union([], _, Acc) ->
    Acc.

%%%----------------------------------------------------------------
send([Pid|Pids], Msg) when is_pid(Pid) ->  Pid ! Msg, send(Pids,Msg);
send([], _) -> ok;
send(Pid, Msg) when is_pid(Pid) -> Pid ! Msg.

%%%----
split_list(all, L) -> {L,[]};
split_list(N, L) when length(L)>N -> split_list(N, L, []);
split_list(_, L) -> {L,[]}.

split_list(N, [H|T], Acc) when N>0 -> split_list(N-1, T, [H|Acc]);
split_list(N, L, Acc) -> {Acc,L}.


%%%----
unify_list(Ms, Pattern, Bs, Sols) ->
    lists:foldl(
      fun (M,Acc) ->
	      case mnemosyne_unify:unify(Pattern, M, Bs) of
		  fail -> Acc; 
		  NewBs -> [NewBs|Acc]
	      end
      end, Sols, Ms).