:- ensure_loaded(common).
cvs_id('$Id: thread_util.pl,v 1.11 2004/10/13 19:35:06 sandro Exp $').
/*

  Some general thread utilities

*/

%:- debug(thread_clear_message).

thread_clear_message :-
	(   thread_peek_message(Msg)
        ->  thread_get_message(Msg),
	    debug(thread_clear_message, 'removed message: ~q', [Msg])
	;   fail
	).
thread_clear_all_messages :-
	(   thread_clear_message,
	    thread_clear_all_messages
	;   true
	).
unit_test_thread_clear_message_0 :-
	thread_clear_all_messages,
	\+ thread_peek_message(_).
unit_test_thread_clear_message_1 :-
	thread_send_message(main, foo),
	thread_clear_all_messages,
	\+ thread_peek_message(_).
unit_test_thread_clear_message_2 :-
	thread_send_message(main, foo),
	thread_send_message(main, bar),
	thread_clear_all_messages,
	\+ thread_peek_message(_).
unit_test_thread_clear_message_10 :-
	(   between(1,10,_), 
	    thread_send_message(main, bar),
	    fail
	;   true
	),
	thread_clear_all_messages,
	\+ thread_peek_message(_).


%%
%%   return after we get some message (returned) or
%%   the time passes (Msg=timeout)
%%
%%   an alternative would be to have an awakener thread
%%   but I suspect that would be more expensive.
%%
interruptable_sleep(T, Msg) :-
	get_time(Start),
	EndTime is Start + T,
	(   repeat,
	    sleep(0.01),
	    % escape from this loop if either ...
	    (  thread_peek_message(Msg),
               thread_get_message(Msg)
	    % ... or ...
	    ;   get_time(Now),
		Msg = timeout,
		Now > EndTime
	    )
	;   true
	), !.

unit_test_interruptable_sleep_1 :-
	get_time(Start),
	interruptable_sleep(0.01, Msg),
	get_time(Stop),
	Dur is Stop - Start,
	Dur < 0.10,
	Msg = timeout.
unit_test_interruptable_sleep_2 :-
	get_time(Start),
	thread_send_message(main, stop_please),
	interruptable_sleep(2, Msg),
	get_time(Stop),
	Dur is Stop - Start,
	Dur < 0.10,
	Msg = stop_please.


thread_create_caught(Goal, Success, Error, Fail) :-
	thread_create(  (
			  (   catch(Goal, Error, true)
			  ->  true
			  ;   Error = failed
			  ),
			  (   var(Error)
			  ->  Success
			  ;   Fail
			  )
			),
			_,[detached(true)]
		     ).


qtest:-
	thread_create_caught(throw(foo), fail, E, E=foo).

qtest:-
	thread_create_caught(true, true, _E, fail).

qtest:-
	thread_create_caught(fail, fail, E, E=failed).





:- dynamic(thread_state/3).
:- multifile(thread_state/3).

% copied from swi "threads", but with State added, so I can tell
% what the threads are doing (if they choose to record their state).
%
% there are lots of design issues here -- alternatives include
% using the stack (how can we peek at another thread's stack?),
% and having a nestable state.   This is probably good enough
% for me for now.
%
% intended use:   eg for lockrw, so I can see which threads are
% hung waiting for what....
%

ps :-
	get_time(Now),
	format('~*t~60|~n', [45]),
        format('~t~w~20|  ~w~32| ~w~n', ['Thread', 'Status', 'State']),
        format('~*t~60|~n', [45]),
        current_thread(Id, Status),
	findall(holds(M, C),
		current_mutex(M, Id, C),
		Locks),
	(   Locks = []
	->  LocksText = ''
	;   LocksText = Locks
	),
	(   (   flag(Id, State, State),
		State \= 0,
		Time=Now
	    ; 	thread_state(Id, State, Time)
	    )
	->  (   State = delegated(Goal)
	    ->  call(Goal, State2, Time2)
	    ;   State2 = State,
		Time2 = Time
	    ),
	    Dur is round(Now - Time),
	    format('~t~w~20|  ~p~32| ~p  [for ~ds] ~w~n', [Id, Status, State2, Dur, LocksText])
	;   format('~t~w~20|  ~p ~w~n', [Id, Status, LocksText])
	),
	% thread_util:rip_thread(Status, Id),
        fail.
ps :-
        format('~*t~60|~n', [45]).

%   fails if there's nothing to pop
thread_pop_state :- thread_pop_state(_,_).
thread_pop_state(OldState, OldTime) :-
	thread_self(Id),
	once(retract(thread_state(Id, OldState, OldTime))).

thread_push_state(State) :-
	thread_self(Id),
	get_time(Now),
	asserta(thread_state(Id, State, Now)).
	
thread_set_state(State) :-
	thread_self(Id),
	assume(ground(State)),
	(   thread_pop_state(OldState, OldTime)
	->  thread_push_state(State),
	    get_time(Now),
	    Dur is (Now - OldTime) * 1000000,
	    debug(thread_state, 'thread ~a  ~p => ~p after ~w usec',
		  [Id, OldState, State, Dur])
	;   thread_push_state(State),
	    debug(thread_state, 'thread ~a   initial state: ~p', [Id, State])
	).
thread_state_call(Goal) :-
	copy_term(Goal, SafeGoal),
	numbervars(SafeGoal, 0, _),   % no vars in state, please
	thread_push_state(SafeGoal),
	call_cleanup(Goal, thread_pop_state).
	
thread_state_call(Label, Goal) :-
	thread_push_state(Label),
	Goal,
	thread_pop_state.

aformat(Atom, Format, Args) :-
	sformat(String, Format, Args),
	string_to_atom(String, Atom).

thread_stats(Thread) :-
	member(Measure, [cputime, inferences, heap,
		   local, localused, locallimit,
		   global, globalused, globallimit,
		   trail, trailused, traillimit]), 
	thread_statistics(Thread, Measure, Value),
	format('~a~15| ~a~30|~w~n', [Thread,Measure,Value]),
	fail.

thread_high_water(thread_high_water(Thread, [local(Local), global(Global), trail(Trail)])) :-
	(  var(Thread)
	-> thread_self(Thread)
	;  true
	),
	thread_statistics(Thread, local, Local),
	thread_statistics(Thread, global, Global),
	thread_statistics(Thread, trail, Trail).

thread_stats :-
	current_thread(Thread, _),
	thread_stats(Thread).



