view reactive.pl @ 2:b39711ae9035

Added explicit import of library(time) for older version of SWI Prolog.

:- module(reactive,
		[	run_process/1
		,	init_process/2
		,	step_event/3
		,	step_timeout/2
		,	step_timeout/3
		,	get_timeout/2
		,	req_message/2
		,	req_message_or_timeout/4
		,	(>)/3
		]).

/** <module> Tools for reactive programming

	This module provides a framework for doing
	reactive programming, where a reactive process is
	represented by one or two continuations  
	which define how it will react to received events or the 
	passage of time.

	Relevant types:
	==
	ptail    ==  pred(process).

	process ---> get(term,ptail)
	           ; gett(time,term,ptail,ptail).


	event   ---> event(time,_).

	event_handler ---> term^ptail.
	==

	The type =|ptail|= represents a slice of execution for a process.
	When called, code may execute until the process
	wishes to block waiting for an event or time-out to occur. 
	It does this by returning a value of type =|process|= 
	representing either a process waiting indefinitely for an event, 
	or a processing waiting until a certain for an event.

	Values of type time are either floats representing times as
	returned by get_time/1 or the atom 'inf' representing a point
	infinitely far in the future. 
*/

:- meta_predicate req_message(:,-).
:- meta_predicate req_message_or_timeout(+,:,:,-).
:- meta_predicate run_process(1).
:- meta_predicate init_process(1,?).
:- meta_predicate >(+,1,?).

:- use_module(qutils).

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

%% init_process( +Spec:ptail, -P:process) is det.
%
%  Initialise process specified by Spec, which is
%  simply a callable term taking one argument.
%  P is bound to representation of sleeping
%  process on exit. 
init_process(PS,P1) :- call(PS,P1).

%% >(+E:event,+Spec:ptail,-P:process) is det.
%
%  Initialise process and immediately supply an event.
%  This uses init_process/2 to initialise the process specified
%  by Spec and then use step_event/3 to process the given event.
%
%  This implies that if Spec is of type ptail, then 
%  Event>Spec is also of type ptail.
>(Event,Spec,Proc) :-
	init_process(Spec,P1),
	step_event(Event,P1,Proc).

%% run_process( +Spec:ptail) is det.
%
%	Runs process represented by Spec (see init_process/2).
%	After initialising, this predicate goes into a recursive procedure
%	waiting for events on the current thread's message queue and
%	passing them to the waiting process.
run_process(M:Spec) :- init_process(M:Spec,Proc), cont_process(Proc).


%% cont_process( +Proc:process) is det.
%
%  Takes a quiescent process, waits for and then handles a new event.
%  (Including timeout events.)
cont_process(Proc) :-
	get_timeout(Proc,T), !,
	garbage_collect_atoms,
	get_message_or_timeout(T,Event),
	event_process_cont(Event,Proc,T).


%% event_process_cont(+Event:evdesc,+P:process,+Deadline:time) is det.
%
%  Handle latest event Event on process Proc. The evdesc type is defined
%  as =|evdesc ---> quit; timeout; event(time,term)|= with the following
%  semantics:
%
%   * quit
%   The process will terminate: this procedure succeeds and returns immediately.
%
%   * timeout
%   The timeout continuation of the process is followed.
%
%   * event(Time,Data)
%   If the current time is less than 0.5s after the event time,
%   the process's event continuation is called. Otherwise,
%   the event is dropped and we continue as if it had never happened.
event_process_cont(quit,_,_) :- !.
event_process_cont(timeout,P1,_) :- step_timeout(P1,P2), !, cont_process(P2).
event_process_cont(event(ET,EData),P1,T1) :- !, 
	get_time(Now),
	(	Now>ET+0.5 -> writeln('#'), cont_process(P1)
	;	cont_event(T1,P1,ET,EData)).

%% cont_event(+Deadline:time, +P:process, +EvTime:time, +EvData:term) is det.
%
%  Handles process continuation after receiving a valid event.
%  If the event is later than the process timeout, the timeout
%  continuation is called recursively until the process is ready
%  to receive the current event. Then this event is handled and
%  the process continues from there.
cont_event(T1,P1,ET,EData) :- 
	(	T1\=inf, ET>T1 
	-> step_timeout(P1,P2),
		get_timeout(P2,T2), !,
		cont_event(T2,P2,ET,EData)
	;	step_event(event(ET,EData),P1,P2), !,
		cont_process(P2)
	).

cont_event_safely(T1,P1,ET,EData) :- 
	(	T1\=inf, ET>T1 
	-> step_timeout(P1,P2),
		get_timeout(P2,T2), !,
		cont_event_safely(T2,P2,ET,EData)
	;	(	step_event(event(ET,EData),P1,P2) ->	true
		;	writeln(failed(step_event(event(ET,EData),P1,P2))), P1=P2
		),
		cont_process(P2)
	).

%% req_message( +EventHandler:event_handler, -P:process) is det.
%
%  Returns process state representing quiescent process
%  waiting indefinitley for event. EventHandler must be
%  a term of the form Msg^PTail, where Msg is to be unified
%  with next event term before calling PTail.
req_message(Mod:Temp^OnEvent, get(Temp,Mod:OnEvent)).

%% req_message_or_timeout( +T:time, +TimeOutHandler:ptail, +EventHandler:event_handler, -P:process) is det.
%
%  Returns process state representing quiescent process
%  waiting for event. EventHandler must be
%  a term of the form Msg^PTail, where Msg is to be unified
%  with next event term  before calling PTail to get the next
%  continuation of the process.
%  TimeOutHandler will be called if time no event arrives
%  before time T.
req_message_or_timeout(T,OnTimeout,Mod:Temp^OnEvent,
	gett(T,Temp,Mod:OnEvent,OnTimeout)).

leq(inf,T2) :- !, T2=inf.
leq(T1,T2)  :- !, (T2=inf -> true; T1=<T2). 

%% get_timeout(+P:process,-Deadline:time) is det
%
% Get timeout time of quiescent process. May be inf.
get_timeout(get(_,_),inf).
get_timeout(gett(T,_,_,_),T).

%% step_timeout(+P1:process,-P2:process) is det.
%
%  Wake up process P1 with timeout event and run until
%  it blocks at P2.
step_timeout(gett(_,_,_,OnTimeout),P2) :- call(OnTimeout,P2).
step_timeout(get(E,OnEvent),get(E,OnEvent)).

%% step_timeout(+T:time, +P1:process,-P2:process) is det.
%
%  If time T is later than P1's timeout time, wake it up and
%  run timeout event until it blocks at P2. Otherwise, P2=P1,
%  ie process is still waiting.
step_timeout(T,P1,P2) :-
	get_timeout(P1,T1), 
	(leq(T1,T) -> step_timeout(P1,P2); P1=P2).

:- index(step_event(0,1,0)).

%% step_event(+E:event, +P1:process, -P2:process) is det.
%
%  Wake up process P1 with event E and run until blocking at P2.
step_event(E,get(E,OnEvent),P2) :- call(OnEvent,P2).
step_event(E,gett(_,E,OnEvent,_),P2) :- call(OnEvent,P2).