:- ensure_loaded(common).
cvs_id('$Id: lockrw.pl,v 1.10 2004/10/15 18:46:31 sandro Exp $').
/*

    Multiple-Readers // Single Writer locking

    see test_lockrw.pl for an example...


    Use of with_read_lock/1 and with_write_lock/1 is recommended
    because they do call_cleanup/3 for you.

    ToDo: support failing after a timeout...?
              accept "fail" as well as "try_now"
              spawn another thread which sleeps and sends fail
              kill the other thread in its sleep if you finish first
  
*/

:- ensure_loaded(library(debug)).
:- ensure_loaded(thread_util).
:- ensure_loaded(schedule).

:- dynamic(read_lock_held/2).     %   (resource, thread_id) 
:- dynamic(read_lock_wait/2).     %   (resource, thread_id) 
:- dynamic(write_lock_held/2).    %   (resource, thread_id)
:- dynamic(write_lock_wait/2).    %   (resource, thread_id)

:- multifile(read_lock_held/2).
:- multifile(read_lock_wait/2). 
:- multifile(write_lock_held/2).
:- multifile(write_lock_wait/2).



xwith_read_lock(rdf, Goal) :- Goal.    % quick hack around ....
xwith_read_lock(Resource, Goal) :-
	obtain_read_lock(Resource),
	call_cleanup(Goal, _, release_read_lock(Resource)).

xwith_write_lock(rdf, Goal) :- Goal.
xwith_write_lock(Resource, Goal) :-
	obtain_write_lock(Resource),
	call_cleanup(Goal, _, (release_write_lock(Resource))).

xwith_write_lock(rdf, Goal, _) :- Goal.
xwith_write_lock(Resource, Goal, Timeout) :-
	thread_self(Me),
	schedule_in(Timeout, timeout_lock(Resource, Me), Tag),
	obtain_write_lock(Resource),
	call_cleanup(Goal, _,
		     (	 unschedule(_, _, Tag),
			 release_write_lock(Resource)
		     )
		    ).

%  nestable -- lock and unlock an equal number of times
obtain_read_lock(Resource) :-
	thread_self(Me),
	debug(lockrw, '~a: lockr ~a', [Me, Resource]),
	thread_push_state(lockrw_mutex),
	with_mutex(lockrw_mutex,
		   (   once(write_lock_held(Resource, _))
		   ->  assertz(read_lock_wait(Resource, Me)),
		       Waiting = true
		   ;   assert(read_lock_held(Resource, Me)),
		       Waiting = fail
		   )
		  ),
	(  Waiting
	-> debug(lockrw, '~a: waiting for ~a', [Me, Resource]),
	   thread_set_state(read_lock_wait(Resource)),
	   thread_get_message(Me, read_lock_granted),
	   assume(read_lock_held(Resource, Me))
	;  true
	),
	thread_pop_state,
	debug(lockrw, '~a: got lockr ~a', [Me, Resource]).

release_read_lock(Resource) :-
	thread_self(Me),
	thread_push_state(lockrw_mutex),
	with_mutex(lockrw_mutex,
		   (   assume(read_lock_held(Resource, Me)),
		       once(retract(read_lock_held(Resource, Me))),
		       (   % transfer lock?
			   \+ once(read_lock_held(Resource, _)),
			   once(retract(write_lock_wait(Resource, Waiter)))
		       ->  assert(write_lock_held(Resource, Waiter)),
			   thread_send_message(Waiter, write_lock_granted)
		       ;   true
		       )
		   )
		  ),
	thread_pop_state.
	

obtain_write_lock(Resource) :-
	thread_self(Me),
	debug(lockrw, '~a: lockw ~a', [Me, Resource]),
	thread_push_state(lockrw_mutex),
	with_mutex(lockrw_mutex,
		   (   (   write_lock_held(Resource, Me)
		       ->  throw(lockrw_nested_lockw(Resource, Me))
		       ;   true
		       ),
		       (   once((   write_lock_held(Resource, _)
				;   read_lock_held(Resource, _)
				))
		       ->  assertz(write_lock_wait(Resource, Me)),
			   Waiting = true
		       ;   assert(write_lock_held(Resource, Me)),
			   Waiting = fail
		       )
		   )
		  ),
	(   Waiting
	->  debug(lockrw, '~a: waiting for lockw ~a', [Me, Resource]),
	    thread_set_state(write_lock_wait(Resource)),
	    thread_get_message(Me, Msg),
	    thread_pop_state,
	    !,
	    (   Msg=write_lock_granted
	    ->  assume(write_lock_held(Resource,  Me)),
		debug(lockrw, '~a: got lockw ~a', [Me, Resource])
	    ;   (   Msg=timeout
		->  fail
		;   throw(unexpected_message(Msg))
		)
	    )
	;  true
	).


release_write_lock(Resource) :-
	thread_self(Me),
	thread_push_state(lockrw_safety_read),
	% force the write to cause whatever regeneration is needed
	% *now*, while we still have the lock....
	once(rdf(_,_,_,_)),
	thread_push_state(lockrw_mutex),
	with_mutex(lockrw_mutex,
		   (   retract(write_lock_held(Resource, Me)),
		       assume(\+write_lock_held(Resource, _)),
		       % let the writers go first
		       (   once(retract(write_lock_wait(Resource, Waiter)))
		       ->  assert(write_lock_held(Resource, Waiter)),
			   thread_send_message(Waiter, write_lock_granted)
		       ;   forall( retract(read_lock_wait(Resource, Waiter)),
				   (  assert(read_lock_held(Resource, Waiter)),
				      thread_send_message(Waiter, read_lock_granted)
				   )
				 )
		       )
		   )
		  ),
	thread_pop_state.

timeout_lock(Resource, Thread) :-
	thread_push_state(lockrw_mutex),
	with_mutex(lockrw_mutex,
		   (   retractall(write_lock_wait(Resource, Thread)),
		       retractall(read_lock_wait(Resource, Thread)),
		       thread_send_message(Thread, timeout)
		   )
		  ),
	thread_pop_state.

lock_report :-
	format('~n~n== Locks==~n~n', []),
	findall(R, ( read_lock_held(R, _)
		   ; read_lock_wait(R, _)
		   ; write_lock_held(R, _)
		   ; write_lock_wait(R, _) ), RDups),
	sort(RDups, Resources),
	forall(member(R, Resources),
	       (   format('For ~p: ', R),
		   (   write('read_lock_held: '),
		       forall(  read_lock_held(R, T),
				(write(T), write(' '))
			     ),
		       write('read_lock_wait: '),
		       forall(  read_lock_wait(R, T),
				(write(T), write(' '))
			     ),
		       write('write_lock_held: '),
		       forall(  write_lock_held(R, T),
				(write(T), write(' '))
			     ),
		       write('write_lock_wait: '),
		       forall(  write_lock_wait(R, T),
				(write(T), write(' '))
			     )
		   )
	       )),
	nl.
	       
		   