/* Copyright (C) 1990 Swedish Institute of Computer Science */

/*
	Modified by K.Y. Hui to use the
	Aberdeen socket library for SICStus, SWI-Prolog & ECLiPSe.

	31-Jan-2002	1)	I dropped Quintus support as it is
				no longer available in our dept.
			2)	This version is modified to support
				the KRAFT monitor (making duplicates).
				Other than that, it is a normal Linda server.

				KH
*/

:- module(linda_server,[	linda/0,
				linda/1,
				linda/2
			]).

:- ['../util/prolog_detect'].				%load Prolog detecting module
:- prolog_detect:prolog(Prolog),
	(	(	Prolog=sicstus,
			!,
			prolog_flag(character_escapes,_,on),		% allow '\n' notation
			prolog_flag(redefine_warnings,_,off),		%turn off redefine warnings
			use_module(library(lists)),
			use_module(library(fastrw))
		);
		(	Prolog=eclipse,
			use_module(library(iso)),
			['../util/retractall']
		);
		true
	).

:- ['../util/newline'].			%load Prolog newline module
:- ['../util/delete_element'].		%load delete module
:- ['../messaging/message_format'].	%KRAFT message-format module needed for duplicating messages for monitor
:- ['../socket_lib/a_sockets'].		%load Aberdeen socket lib

:- dynamic ts/1.
:- dynamic tuple/1.
%:- dynamic waiting_list/6.
:- dynamic waiting_list/5.
:- dynamic linda_trace/0.
:- dynamic connection/3.		%existing connection w/ socket & RW streams

client_timeout(time(0,100)).		%timeout for checking if client is ready

linda(Hook) :-
	nonvar(Hook),
	Hook = (Host:Port)-Goal,
	create_server_socket(Socket, Host, Port),
	call(Goal),
	server1(Socket,[]).		%start with no existing connection

linda(Host,Port) :-
	linda(((Host:Port)-true)).

linda :-
	linda(	(Addr-	(
			write('Server address is '),
			write(Addr),
			nl,
			flush_output(user)
			)
		)
	).

%
% serve network connection to the server-socket or
% a list of existing clients
%
server1(Server_socket,Active_connections) :-
	wait_for_arrival(Server_socket,Active_connections,Ready_connections,New_active_connections),	%get connections ready for reading
	serve_all(Ready_connections,New_active_connections,Final_active_connections),			%serve all ready connections
	server1(Server_socket,Final_active_connections).						%loop

%
% wait for arrival from a server socket & a list of existing client streams
% return the new list of streams which has incoming traffic
%
wait_for_arrival(Server_socket,Active_connections,Ready_connections,New_active_connections):-
	findall(X,member(connection(_,X,_),Active_connections),Active_read_streams),		%get the list of active read streams to test
	a_sockets:a_socket_select([Server_socket],Active_read_streams,0,Ready_connections),	%check for readiness
	merge_connection(Ready_connections,Active_connections,New_active_connections).		%merge connection list

merge_connection([],Old,Old).
merge_connection([C|Cs],Old,New) :-
	member(C,Old),
	!,
	merge_connection(Cs,Old,New).
merge_connection([C|Cs],Old,[C|Temp]) :-
	merge_connection(Cs,Old,Temp).



%
% create server socket
%

create_server_socket(Socket,Host,Port) :-
	a_sockets:a_hostname(Host),
	a_sockets:a_socket_create('AF_INET',Socket),
	a_sockets:a_socket_bind(Socket,'AF_INET'(Port)),
	a_sockets:a_socket_listen(Socket,5).



%
% serve all connections ready for reading
%
serve_all([],Active,Active) :-				%finished!
	!.

serve_all([C|Cs],Active,Final_active) :-
	catch(serve_one(C,Active,New_active),_,handle_error(C,Active,New_active)),	%serve & trap exception
%	serve_one(C,Active,New_active),								%serve 1 socket
%	on_exception(_,serve_one(C,Active,New_active),handle_error(C,Active,New_active)),	%serve & trap exception
	serve_all(Cs,New_active,Final_active).	%serve the rest

handle_error(connection(S,R,W),Active,New_active) :-
%	skip_one_line(R),								%skip to eol
	delete_element:delete_element(Active,connection(S,R,W),New_active).		%remove it from active list
%	connection(Socket,R,_),		%get streams
%	skip_line(R),			%skip to eol
%	delete(Active,Socket,New_active).

%
% skip to EOL
%
skip_one_line(R) :-
	at_end_of_stream(R),
	!.
skip_one_line(R) :-
	at_end_of_line(R),
	!.
skip_one_line(R) :-
	get_code(R,_),
	skip_one_line(R).

%
% serve an active-&-ready connection
%

%
% existing connection
%
serve_one(connection(S,R,W),Active,New_active) :-
	connection(S,R,W),					%an existing client
	!,
	write('---------- Existing connection'),
	nl,
	flush_output(user),					%flushing needed for ECLiPSe
	get_code(R,Protocol),					%read protocol from client
%	get0(R,Protocol),					%read protocol from client
	serve_one(Protocol,connection(S,R,W),Active,New_active),%serve with protocol
	!.

%
% new connection
%
serve_one(Connection,Active,New_active) :-
	Connection=connection(_,R,_),
	assertz(Connection),							%store connection info
	write('---------- New connection'),
	nl,
	flush_output(user),							%flushing in ECLiPSe
	get_code(R,Protocol),
%	get0(R,Protocol),
	merge_connection([Connection],Active,Temp_active),
	serve_one(Protocol,Connection,Temp_active,New_active),			%serve with protocol
	!.


%
% serve an active socket with the given protocol
%
serve_one(Protocol,Connection,Active,New_active) :-
	Protocol < 0,							%EOF, broken connection
	!,
%	Connection=connection(S,R,W),
	close_connection(Connection),					%close connection
%	a_sockets:a_socket_destroy(Socket),						%close socket
	write('---------- Connection closed'),
	nl,
	flush_output(user),
	retractall(waiting_list(_,_,_,Connection,_)),			%remove all waiting entries of it
	retract(Connection),						%remove connection
	delete_element:delete_element(Active,Connection,New_active).	%remove it from list

serve_one(Protocol,Connection,Active,Active) :-
	Connection=connection(_,R,_),					%get the R & W streams
	get_code(R,Request),						%get client request
%	get0(R,Request),						%get client request
%	trace_linda(before,Request,Ti),
	perform_request(Request,Protocol,Connection).			%perform request
%	trace_linda(after,_,Ti).



%----------------------------------------
%
% serving different requests
%
%----------------------------------------

%
% rd_noblock
%
perform_request(0's,Protocol,connection(_,R,W)) :-  /* rd_noblock */ !,
	get_input(R,Protocol,Tuple),
	cond_to_client(W,Protocol,tuple(Tuple),Tuple).

%
% rd
%
perform_request(0'r,Protocol,connection(S,R,W)) :-  /* rd */ !,
	get_input(R,Protocol,Tuple),
	to_client_blocking(rd,Protocol,connection(S,R,W),Tuple).

%
% rd with a list of templates
%
perform_request(0'R,Protocol,connection(S,R,W)) :- /* rd, conjunction */ !,
	get_input(R,Protocol,Tlist),
	(   member(T,Tlist),tuple(T) ->		%if there is a matching member
	    to_client(W,Protocol,T)		%then that's the answer
	;   time_stamp(TimeStamp),		%otherwise store into waiting list
	    assert_waiting_lists(Tlist,Protocol,TimeStamp,connection(S,R,W),rd)
	).

%
% in
%
perform_request(0'i,Protocol,connection(S,R,W)) :-  /* in */ !,
	get_input(R,Protocol,Tuple),
	to_client_blocking(in,Protocol,connection(S,R,W),Tuple).
%	to_client_blocking(Write,in,Protocol,Socket,Tuple).

%
% in with a list of templates
%
perform_request(0'I,Protocol,Connection) :- /* in, conjunction */ !,
	Connection=connection(_,R,W),
	get_input(R,Protocol,Tlist),
	(   member(T,Tlist),tuple(T) ->		%if there is a template matching
	    retract(tuple(T)),			%remove tuple
	    to_client(W,Protocol,T)		%and send to client
	;   time_stamp(TimeStamp),		%otherwise save into waiting list
	    assert_waiting_lists(Tlist,Protocol,TimeStamp,Connection,in)
	).

%
% in_noblock
%
perform_request(0'j,Protocol,connection(_,R,W)) :-  /* in_noblock */ !,
	get_input(R,Protocol,Tuple),
	cond_to_client(W,Protocol,retract(tuple(Tuple)), Tuple).

%
% out
%
perform_request(0'o,Protocol,connection(_,R,_)) :- /* out */ !,
	get_input(R,Protocol,Tuple),
	perform_out(Tuple),
	make_duplicates(Tuple).		%make duplicates to monitors if needed

%
% ping
%
perform_request(0'p,Protocol,connection(_,R,W)) :-
	get_input(R,Protocol,In),	%get input from client
	ping_answer(In,Out),		%get answer to 'ping'
	to_client(W,Protocol,Out).	%send reply to client

%
% bagof_rd_noblock
%
perform_request(0'b,Protocol,connection(_,R,W)) :- /* bagof_rd_noblock */ !,
	get_input(R,Protocol,b(Template,Tuple,Bag)),
	tupleGoal(Tuple,Goal),
	(bagof(Template,Goal,Bag) ; Bag = []),
	to_client(W,Protocol,Bag).

%
% call
% i.e. execute a goal
%
perform_request(0'c,Protocol,connection(_,R,W)) :- /* call */ !,
	get_input(R,Protocol,Goal),
	cond_to_client(W,Protocol,call(Goal), Goal).

%
% trace(?)
%
perform_request(0't,Protocol,connection(_,R,W)) :- /* trace */ !,
	get_input(R,Protocol,Q),
	trace_answer(Q,Repl),
	to_client(W,Protocol,Repl).

%
% unknown request
%
perform_request(Request,_,connection(_,R,_)) :-
	write('*** Unknown request from '),
	write(R),
	write(':'),
	write(Request),
	nl,
	flush_output(user).

%
% the answer of ping
%
ping_answer(ping,pong) :- !.
ping_answer(X, illegal(X)).


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

%
% perform that 'out' request from a client
%

%
% somebody is waiting for this tuple
%
perform_out(Tuple) :-
	retract(waiting_list(Tuple,Protocol,TimeStamp,connection(_,_,W),Op)),	%find a waiting client

%	retract(waiting_list(Tuple,Protocol,TimeStamp,Socket,W,Op)),		%find a waiting client
%	client_timeout(Timeout),						%get timeout value
%	a_sockets:a_socket_select([],[Socket],[],Timeout,_,[Socket],_),		%check if it is ready for writing
	!,
	remove_disjunction(TimeStamp),						%remove disjunctions
	to_clientS(Protocol,W,Tuple),
	(	(	Op=rd,							%if it is 'rd', tuple is still here
			perform_out(Tuple)					%so check for other waiting clients
		);
		true
	).


%
% nobody is waiting, just store it
%
perform_out(Tuple) :-
	assertz(tuple(Tuple)).

%
% remove waiting entries from the same request (template list)
%
remove_disjunction(TS) :- TS>0, !,
	retractall(waiting_list(_,_,TS,_,_)).
remove_disjunction(_).	

%-----------------------------------------------------------------------------
trace_answer(Q,on) :- var(Q), linda_trace, !.
trace_answer(Q,off) :- var(Q), !.
trace_answer(on,on) :- linda_trace, !.
trace_answer(on,on) :- !, assert(linda_trace).
trace_answer(_,off) :- retractall(linda_trace).

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

%
% time-stamp for waiting list
%
time_stamp(T) :- retract(ts(T)), !, T1 is T+1, assert(ts(T1)).
time_stamp(1) :- assert(ts(2)).

%
% put a list of matching templates into waiting list
%
assert_waiting_lists([],_,_,_,_).

assert_waiting_lists([T|Ts],Protocol,TimeStamp,Connection,Op) :-
	assert_one_waiting(T,Protocol,TimeStamp,Connection,Op),
	assert_waiting_lists(Ts,Protocol,TimeStamp,Connection,Op).

%
% put 1 request into waiting list
%
assert_one_waiting(Tuple,Protocol,TimeStamp,Connection,Op) :-
	assertz(waiting_list(Tuple,Protocol,TimeStamp,Connection,Op)).

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

%
% answer request & reply to client
% may block client & put it into waiting list
%

%
% rd
%
to_client_blocking(rd,Protocol,connection(_,_,W),Tuple) :- 
	tuple(Tuple),				%match tuple
	!,
	to_client(W,Protocol,Tuple).		%reply to client

%
% in
%
to_client_blocking(in,Protocol,connection(_,_,W),Tuple) :- 
	retract(tuple(Tuple)),			%match & remove tuple
	!,
	to_client(W,Protocol,Tuple).		%reply to client

%
% ok, put into waiting list
%
to_client_blocking(Op,Protocol,Connection,Tuple) :- 
	assert_one_waiting(Tuple,Protocol,0,Connection,Op).	%put into waiting list


%-----------------------------------------------------------------------------
tupleGoal(T,tuple(T)) :- var(T).
tupleGoal(T,G) :- tupleGoal1(T,G).
tupleGoal1(V^T,V^G) :- !, tupleGoal1(T,G).
tupleGoal1(T,linda:tuple(T)).

%-----------------------------------------------------------------------------
get_input(R,0'f, Data) :-
	fast_read(R,Data).
get_input(R,0'p, Data) :-
	read(R,Data).
get_input(_,P,_) :-	write('*** unknown protocol '),
			write(P),
			nl,
			flush_output(user).
	
put_output(W,0'f, Data) :-
	fast_write(W,Data).

put_output(W,0'p, Data) :-
	write_canonical(W,Data),
	write(W,'.'),
	newline:newline(W).

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

%
% send to client according to protocol
%

%
% send command 'C' & data
%
to_client(W,Protocol,C,Data) :- 
	put(W,C),
	put_output(W,Protocol,Data),
	flush_output(W).

%
% send data only
%
to_client(W,Protocol,Data) :- 
	put_output(W,Protocol,Data),
	flush_output(W).

%
% send data to a particular stream
%
to_clientS(Protocol,Stream,Data) :- 
	put_output(Stream,Protocol,Data),
	flush_output(Stream).

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

%
% answer request & reply to client without blocking
% if succeed, return 's' then result tuple
% if fail, return 'f'
%

%
% goal succeeds
%
cond_to_client(W,Protocol,Cond,Data) :-
	call(Cond),
	!,					%condition succeeds
	to_client(W,Protocol,0's,Data).		%reply 's' & result tuple

%
% goal fails
%
cond_to_client(W,_,_,_) :-
	put(W,0'f),				%only 'f' for failure
	flush_output(W).

%-----------------------------------------------------------------------------
/*
trace_linda(before,R,Ti) :-
	linda_trace, !,
	put(user_output,R),
	tab(user_output,2),
	flush_output(user_output),
	statistics(runtime,[Ti,_]).
trace_linda(after,_,Ti) :-
	linda_trace, !,
	statistics(runtime,[Tf,_]),
	T is Tf-Ti,
	writet(user_output,'(~w ms)~n',[T]).
trace_linda(_,_,0).

*/


%
% close a connection
%
% NOTE:	SICStus is not quite happy if I simply close a socket
%	So I close the stream associated with it instead.
%	In other systems, you may be able to close the socket.
%
close_connection(Connection) :-
	prolog_detect:prolog(Prolog),
	close_connection(Prolog,Connection).
close_connection(sicstus,connection(_,R,_)) :-
	close(R),
	!.
close_connection(swi,connection(_,R,W)) :-
	close(R),
	close(W),
	!.
close_connection(eclipse,connection(_,R,_)) :-
	close(R),
	!.
close_connection(_,_).

%
% make duplicate for all monitors
%
make_duplicates(Tuple) :-
	findall(X,tuple(monitor(X)),Monitors),	%find all monitor/1 in the tuple space
	make_duplicates2(Monitors,Tuple).

%
% compose & out a duplicate to every monitor
%

make_duplicates2([],_).				%end of monitor list

make_duplicates2([M|Ms],Tuple) :-
	compose_duplicate(Tuple,M,Dup),		%compose a duplicate to a single monitor
	!,
	perform_out(Dup),			%perform an "out" of the dup
	make_duplicates2(Ms,Tuple).		%continue w/ the rest

make_duplicates2([_|Ms],Tuple) :-
	make_duplicates2(Ms,Tuple).		%duplicate not composed, proceeds with list

%
% given the name of the monitor and a tuple
% compose a copy of the msg/tuple to the monitor
%
compose_duplicate(Tuple,_,_) :-
	message_format:mf_decompose_message(_,Target,_,_,Tuple), 
	tuple(monitor(Target)),  % if message is to any monitor we know about
	!,
	fail.                    % then fail - do not compose duplicate.

compose_duplicate(Tuple,Monitor,Duplicate) :-
	message_format:mf_decompose_message(Context,_,Tuple),	%try decomposing into context & core
	message_format:mf_decompose_context(_,_,Sender,_,_,Context), %get original sender

      %adjustment I have made - T.Francis 1/7/1998
      %make the sender and receiver of forward msg match those of kraft_msg
	% i.e. message sender is original sender
	%      receiver is the Monitor

        % T.Francis - 2/7/98
        %content is original kraft_msg (not just the ccql part)

       message_format:mf_set_argument([sender:Sender,
	                               receiver:Monitor, 
				       language:'kraft_msg',
				       content:Tuple], Args),

       message_format:mf_compose_message(Sender,Monitor,forward,Args,Duplicate).	
	%compose forward message with those Args.


