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

/*
	Modified by K.Y. Hui to use the
	Aberdeen socket library for SICStus & Quintus Prolog

	Modified by:	K.Y. Hui
	Version:	1.21
	Date:		12 November 1997
*/

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

:- use_module('util/prolog_detect').
:- use_module('util/newline').

:- use_module('../messaging/message_format').	%message-format module needed for duplicating messages for monitor

:- prolog_detect:prolog(Prolog),
	(	(Prolog=sicstus,
		!,
		prolog_flag(character_escapes,_,on),		% allow '\n' notation
		use_module(library(lists)),
		use_module('fastrw_lib/fastrw')
		);
		(Prolog=quintus,
		!,
		use_module(library(lists)),
		use_module(library(basics)),
		use_module('fastrw_lib/fastrw')
		);
		(Prolog=eclipse,
		use_module(library(iso))
		);
		true
	).


:- use_module('socket_lib/sockets').

:- dynamic ts/1.
:- dynamic tuple/1.
:- dynamic waiting_list/6.
:- 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-	(
			format('Server address is ~q~n',[Addr]),
			flush_output(user)
			)
		)
	).

%
% serve network connection to the server-socket or
% a list of existing clients
%

server1(Server_socket,Active) :-
	wait_for_arrival(Server_socket,Active,Sockets_ready,New_active),	%get sockets ready for reading
	serve_all(Sockets_ready,New_active,Final_active),			%serve all ready sockets
	server1(Server_socket,Final_active).					%loop

%
% wait for arrival from a server socket & a list of existing clients
% return the a new socket or an existing socket which has incoming traffic
%

wait_for_arrival(Server_socket,Active,Sockets_ready,New_active) :-
	socket_select([Server_socket|Active],[],[],0,Sockets_ready,_,_),	%check for readiness
	new_sockets_in_list(Sockets_ready,Active,New_active).			%update active list


%
% given a list of new/existing incoming sockets
% merge them to give a new list
%

new_sockets_in_list([],Active,Active) :-
	!.

new_sockets_in_list([Socket|Sockets],Active,New_active) :-
	member(Socket,Active),				%already in list, skip it
	!,
	new_sockets_in_list(Sockets,Active,New_active).	%and proceed

new_sockets_in_list([Socket|Sockets],Active,New_active) :-
	new_sockets_in_list(Sockets,[Socket|Active],New_active).	%add into list



%
% create server socket
%

create_server_socket(Socket,Host,Port) :-
	hostname(Host),
	socket_create('AF_INET',Socket),
	socket_bind(Socket,'AF_INET'(Port)),
	socket_listen(Socket,5).



%
% serve all sockets ready for reading
%

serve_all([],Active,Active) :-				%finished!
	!.

serve_all([S|Ss],Active,Final_active) :-
%	serve_one(S,Active,New_active),			%serve 1 socket
	on_exception(_,serve_one(S,Active,New_active),handle_error(S,Active,New_active)),
	serve_all(Ss,New_active,Final_active).	%serve the rest

handle_error(Socket,Active,New_active) :-
	connection(Socket,R,_),		%get streams
	skip_line(R),			%skip to eol
	delete(Active,Socket,New_active).
%	set_input(user_input),
%	set_output(user_output).



%
% serve an active socket
%

%
% existing connection
%
serve_one(Socket,Active,New_active) :-
	connection(Socket,R,W),					%an existing client
	!,
	format(user,"---------- Existing connection at socket # ~d~n",[Socket]),
	flush_output(user),					%flushing needed for ECLiPSe
%	set_input(R),
%	set_output(W),
	get0(R,Protocol),					%read protocol from client
	serve_one(Protocol,Socket,R,W,Active,New_active),	%serve with protocol
%	set_input(user_input),
%	set_output(user_output),
	!.

%
% new connection
%
serve_one(Socket,Active,New_active) :-
	socket_accept(Socket,New_socket,R,W),					%accept from server-socket
	assertz(connection(New_socket,R,W)),					%store connection info
	format(user,"---------- New connection at socket # ~d~n",[New_socket]),
	flush_output(user),							%flushing in ELCiPSe
%	set_input(R),
%	set_output(W),
	get0(R,Protocol),
	serve_one(Protocol,New_socket,R,W,[New_socket|Active],New_active),	%serve with protocol
%	set_input(user_input),
%	set_output(user_output),
	!.


%
% serve an active socket with the given protocol
%
serve_one(Protocol,Socket,_,_,Active,New_active) :-
	Protocol < 0,							%EOF, broken connection
	!,
	socket_destroy(Socket),						%close socket
	format(user,"---------- Connection closed at socket # ~d~n",[Socket]),
	flush_output(user),
	retractall(waiting_list(_,_,_,Socket,_,_)),			%remove all waiting entries of it
	retract(connection(Socket,_,_)),				%remove connection
	delete(Active,Socket,New_active).				%remove it from list

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



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

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

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

%
% rd with a list of templates
%
perform_request(0'R,Protocol,Socket,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
%	    current_output(Stream),
	    assert_waiting_lists(Tlist,Protocol,TimeStamp,Socket,W,rd)
	).

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

%
% in with a list of templates
%
perform_request(0'I,Protocol,Socket,R,W) :- /* in, conjunction */ !,
	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
%	    current_output(Stream),
	    assert_waiting_lists(Tlist,Protocol,TimeStamp,Socket,W,in)
	).

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

%
% out
%
perform_request(0'o,Protocol,_,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,_,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,_,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,_,R,W) :- /* call */ !,
	get_input(R,Protocol,Goal),
	cond_to_client(W,Protocol,call(Goal), Goal).

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

%
% unknown request
%
perform_request(Request,_,_,R,_) :-
%	current_input(Stream),
	format(user_error,'*** Unknown request from ~q: ~q~n',[R,Request]),
	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,Socket,W,Op)),	%find a waiting client
	client_timeout(Timeout),					%get timeout value
	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,Socket,Stream,Op) :-
	assert_one_waiting(T,Protocol,TimeStamp,Socket,Stream,Op),
	assert_waiting_lists(Ts,Protocol,TimeStamp,Socket,Stream,Op).

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

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

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

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

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

%
% ok, put into waiting list
%
to_client_blocking(W,Op,Protocol,Socket,Tuple) :- 
%	current_output(Stream), 
	assert_one_waiting(Tuple,Protocol,0,Socket,W,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,_) :-	format(user,"*** unknown protocol ~a~n",[P]),
			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(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).
%	current_output(S),
%	flush_output(S).

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

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

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

%
% 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), !,				%tuple found
	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).
%	current_output(Stream),
%	flush_output(Stream).

%-----------------------------------------------------------------------------
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,
	format(user_output,'(~w ms)~n',[T]).
trace_linda(_,_,0).





%
% 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.
