% Since version 2.0 RC2
% Demonstrate the utility predicates "init_join" and "join" that allow truly CONCURRENT
%    parallel streams of processing to be joined.
% "init_join" is now a builtin while "join" is simplified to use a new builtin "join_test" behind the scenes.

% MUST BE run with "prova process_join_concurrent.prova -t -1" to have the agent non-terminating.

:- eval(consult("utils.prova")).

:- eval(process_join()).

process_join() :-
	println(["==========process_join_concurrent=========="]),
	iam(Me),
	unique_id(XID),
	% The result of the above is that we now have the conversation-id XID initialised.
	% Create a join predicate join_1 with the list of required input patterns in the last argument.
	init_join(XID,join_1,[c(_),b(_)]),
	% This will create two CONCURRENT parallel processing streams due to "switch_thread" moving
	%    the processing to new threads from the pool
	fork_a_b(Me,XID).

switch_thread() :-
	sendMsg(XID,async,0,switch,[]),
	rcvMsg(XID,async,From,switch,[]).

fork_a_b(Me,XID) :-
	switch_thread(),
	TH=Thread.currentThread(),
	println([TH,a]," > "),
	fork_c_d(Me,XID).
fork_a_b(Me,XID) :-
	switch_thread(),
	TH=Thread.currentThread(),
	println([TH,b]," > "),
	% Tell the join join_1 that a new pattern is ready
	join(Me,XID,join_1,b(1)).

fork_c_d(Me,XID) :-
	switch_thread(),
	TH=Thread.currentThread(),
	println([TH,c]," > "),
	% Tell the join join_1 that a new pattern is ready
	join(Me,XID,join_1,c(1)).
fork_c_d(Me,XID) :-
	switch_thread(),
	TH=Thread.currentThread(),
	println([TH,d]," > ").

% The following rule will be invoked by join once all the inputs are assembled.
join_1(Me,XID,Inputs) :-
	println(["Joined for XID=",XID," with inputs: ",Inputs]),
	TH=Thread.currentThread(),
	println([TH,joined]," > ").

% Typical output
% ==========process_join==========
% Thread[pool-2-thread-2,5,main] > b
% Thread[pool-2-thread-1,5,main] > a
% Thread[pool-2-thread-4,5,main] > d
% Thread[pool-2-thread-3,5,main] > c
% Joined for XID=1 with inputs: [["b",1], ["c",1]]
% Thread[pool-2-thread-3,5,main] > joined	