:- ensure_loaded(common).
cvs_id('$Id: harvest.pl,v 1.40 2005/04/14 00:01:29 sandro Exp $').
/*

  This implementation just uses the prolog database for the work
  queue.  In theory using message queues would be better, but it's
  a little less flexible, and the performance benefits are small here
  (given the work involved in really handling each queue entry).

*/

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

:- multi_dynamic(harvest_queue/2).
:- multi_dynamic(harvest_in_progress/1).
:- multi_dynamic(harvest_completed/2).
:- multi_dynamic(host_contact_started/3).
:- multi_dynamic(host_contact_stopped/3).
:- multi_dynamic(this_harvester/1).

:- multi_dynamic(harvest_thread/1).

% start_harvest_threads :- true.
start_harvest_threads :-
	genid(harvester, ThisHarvester),
	assert(this_harvester(ThisHarvester)),
	start_harvest_threads(100, 12), % background harvesting
	start_harvest_threads(2, 8),   % side-effect of browsing
	start_harvest_threads(1, 2).   % only for explicit loads

stop_harvest_threads :-
	retractall(harvest_thread(_)).  % ... does NOT reap them....

start_harvest_threads(MaxPri, Count) :-
	forall(between(1,Count,_),
	       (   flag(harvest_thread_count, TCount, TCount+1),
		   aformat(Name, 'harvest_~w', [TCount]),
		   assert(harvest_thread(Name)),
		   thread_create(harvest_loop(MaxPri), _, [alias(Name),
                         % these numbers for all threads must add up
                         % to less than the available address space
                         % (2G on 32-bit machines).   So keep 'em small!
							  global(1024),
							  local(1024),
							  trail(1024),
							  argument(1024),
							  stack(1024)
							  ])
	       )).



harvest_loop(MaxPri) :-
	thread_self(Me),
	thread_push_state(waiting(maxpri=MaxPri)),
	repeat,
	(   between(1, MaxPri, Priority),
	    harvest_dequeue(Priority, Ret),
	    assertz(harvest_in_progress(Ret), Handle)
	->  harvest_with_error_catching(Ret),
	    erase(Handle),
	    get_time(Now),
	    assertz(harvest_completed(Ret, Now)),
	    thread_high_water(HW),
	    log_term(HW),
	    thread_set_state(waiting(maxpri=MaxPri))  % restart the state timer
	;   sleep(0.05)
	),
	\+ harvest_thread(Me),   % fail => loop back  [unless stop called]
	thread_pop_state.

harvest_dequeue(Priority, Ret) :-
	with_mutex(harvest_queue_mutex,
		   (   retract(harvest_queue(Priority, Ret)),
		       (   ret_address_still_fresh(Ret)
		       ->  % ret_set_status(Ret, retr:previousRetrievalStillFresh),
			   fail    % ignore this one and try another
		       ;   true    % keep working with this one
		       ),
		       (  claim_host_slot(Ret)
		       ->  true   % this one's fine
		       ;   % ret_set_status(Ret, waitingForHostAvailability),
			   assertz(harvest_queue(Priority, Ret)), % back on the queue!
			   fail	% try another
		       )
		   )
		  ),
	debug(harvest, 'dequeued ~q', [[Priority, Ret]]).

claim_host_slot(Ret) :-
	once(ret_get(Ret, retr:host, literal(Host))),
	debug(host_slots, 'claim_host_slot(~q) for ~q', [Ret, Host]),
	with_mutex(host_mutex,
		   (
		     findall(x, host_contact_started(Host, _StartTime, _Ret), L),
		     length(L, Connections),
		     host_max_connections(Host, Max),
		     (   Connections >= Max
		     ->  debug(host_slots, 'claim_host_slot failing, Connection=~w Max=~w',
			       [Connections, Max]),
			 fail
		     ;   get_time(Start),
			 (   ret_host_still_warm(Ret)
			 ->  fail
			 ;   true
			 ),
			 assert(host_contact_started(Host, Start, Ret)),
			 debug(host_slots, 'claim_host_slot claimed!', [])
		     )
		   )
		  ).

release_host_slot(Ret) :-
	once(ret_get(Ret, retr:host, literal(Host))),
	with_mutex(host_mutex,
		   (
		     retract(host_contact_started(Host, _, Ret)),
		     get_time(End),
		     assertz(host_contact_stopped(Host, End, Ret))
		   )
		  ).

ret_address_still_fresh(Ret) :-
	once(ret_get(Ret, retr:initialResource, Address)),
	% special case to avoid endless TAP junk, until they
	% start serving RDF instead of SOAP....
	atom_prefix(Address, 'http://tap.stanford.edu/data/').
ret_address_still_fresh(Ret) :-
	once(ret_get(Ret, retr:initialResource, Address)),
	ret_options(Ret, Options),
	Options = [],
	address_latest_successful_ret(Address, OldRet),
	stale(OldRet, SecondsStale, Options),
	SecondsStale < 0.

harvest_with_error_catching(Ret) :-
	catch((   thread_state_call(delegated(ret_get_status_label(Ret)),
				    harvest_with_host_accounting(Ret))
	      ->  true
	      ;   Error = failed
	      ),
	      % my(Error) when you want prolog errors to not be caught...
	      Error,   
	      true % print_message(error, harvest_hidden(Ret, Error)) % helps with debugging, yet clutters console.  Hrm.
	     ),
	(   var(Error)
	->  true
	;   ( Error = my(E2); Error = E2 ),
	    !, 
	    debug(harvest, 'harvest error: ~q ~q', [E2, Ret]),
	    assert(ret_error(Ret, E2)),
	    term_to_atom(E2, E2Atom),
	    ret_set(Ret, retr:error, literal(E2Atom))
	).

harvest_with_host_accounting(Ret) :-
	call_cleanup(harvest(Ret),
		     release_host_slot(Ret)).

:- dynamic(host_max_connections/2).
:- dynamic(host_idle/2).

host_max_connections('www.w3.org', 10).
host_max_connections('triplestore.aktors.org', 3).
host_max_connections('no_connections.example.org', 0) :- !.
host_max_connections(_Host, 1).

% this is the number of seconds a slot must sit idle before it
% can be re-used
host_idle('www.w3.org', 0).
host_idle(_Host, 10).

ret_host_is_full(Ret) :-
	once(rdfpage(Ret, retr:host, literal(Host))),
	findall(x, host_contact_started(Host, _StartTime, _Ret), L),
	length(L, Connections),
	host_max_connections(Host, Max),
	Connections >= Max.
	

ret_host_still_warm(Ret) :-
	\+ ret_host_idle_time_reached(Ret).

ret_host_idle_time_reached(Ret) :-	
	once(rdfpage(Ret, retr:host, literal(Host))),
	host_max_connections(Host, Max),
	get_time(Now),
	findall(t(EndTime, OldRet),
		host_contact_stopped(Host, EndTime, OldRet),
		StopTimeList),	% naturally sorted by assertz
	length(StopTimeList, HighWater),
	(   HighWater == Max
	->  debug(host_slots, 'claim_host_slot slots have once been used...', []),
	    [t(EarliestEndTime, EarliestRet) | _] = StopTimeList,
	    Idle = Now - EarliestEndTime,
	    host_idle(Host, MinIdle),
	    debug(host_slots, 'claim_host_slot Idle=~w MinIdle=~w', [Idle, MinIdle]),
	    (   Idle < MinIdle
	    ->  debug(host_slots, 'host ~q only idle ~w... failing', [Host, Idle]),
		fail
	    ;   true
	    ),
	    retract(host_contact_stopped(Host, EarliestEndTime, EarliestRet)),
	    debug(host_slots, 'retracted earliest stopped-record', [])
	;   debug(host_slots, 'host still has never-used slots', [])
	).
	
	
harvest(Ret) :-
	perform_retrieval(Ret),
	once(ret_get(Ret, retr:initialResource, Address)),
	log_term(retrieval_done, retrieval_done(Address, Ret)),
	(   ret_get(Ret, rdf:type, retr:'Validation')
	->  true
	;   ret_set_status(Ret, retr:parsing),
	    parse_retrieval(Ret),
	    request_indexing(Ret),
	    ret_set_status(Ret, retr:lookingForLinks),
				% ...?
	    ret_set_status(Ret, retr:forwardChaining),
	    maybe_mark_as_ontology(Ret),
				% ...?
	    ret_set_status(Ret, retr:done),
	    print_message(informational, format('harvest done, ~q', [Address])),
	    reset_home_page_cache
	).

record_redirect(Ret) :-
	ret_get(Ret, retr:initialResource, U1),
	ret_get(Ret, retr:finalResource, U2),
	(   U1 = U2
	->  true
	;   ret_assert_triple(Ret, U1, retr:redirectsTo, U2)
	).


/*
initialize_retrieval(_Item, Ret) :-
	gensym(ret, Ret).
download(_) :- sleep(2).
parse(_) :- sleep(2).
index_retrieval(_) :- sleep(2).
*/

/*
queue_from_precache ?
obsoleted?


add_to_precache(Address, Reason)
refresh(Address)
    selected_for_viewing
    entered_as_search
  
wait_for_harvest(Address, Timeout)


view?subject=http....
search?text=http...     <-  redirect
        http_header.pl   moved(To)
view?search=[....]

throw(http_reply(moved(NewAddress)))

throw(http_reply(cgi_stream(In, Len)))

need to replce http_wrapper, I think.....
      - with one that does streaming....
		 
search

   buffer....   until some time?



flush_to_client :-
	[ do second half of http_wrapper, sending headers,
	  sending memfile, ... ]
        & set a flag saying we did so and you should just exit.

  [ we can do that on a timeout....
      MAYBE even in another thread??? ]

    I doubt that.

*/


%   highest level user interface...
/*
selected_for_viewing(Address, Timeout) :-
	get_time(Now),
	add_to_precache(Address, selected_for_viewing(Now)),
	queue_for_harvest(Address, [wait(Timeout)]).

entered_as_search_text(Address, Timeout) :-
	get_time(Now),
	add_to_precache(Address, entered_as_search_text(Now)),
	queue_for_harvest(Address, [wait(Timeout)]).

refresh(Address, Timeout) :-
	queue_for_harvest(Address, [max_age(0), wait(Timeout)]).
*/


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%
%%  [ MAIN WEB API ]
%%
%%  webload(+Address, +Options)
%%
%%  Request that data from the given Address be retrieved, parsed,
%%  and loaded as soon as possible, within the given priority.
%%  Priority is an integer; 1 is the most-urgent.
%%
%%  Options:
%%
%%       max_age(+Sec)      as for stale/3, perform_retrieve/1
%%       priority(Pri)      which level of the queue to use; 1=first; def=3
%%
%%       wait(Seconds)      max seconds to wait before returning
%%        intime(Goal)      goal to call if it finished before timeout
%%       timeout(Goal)      goal to call if it did not finish in time
%%     retrieval(-Ret)      unifies with the Retrieval created/used
%%
%%  If the item is already on the queue.... we still make a new retrieval
%%  and put it on the queue.   This is our lazy way of handling any callbacks
%%  it might have.

webload(Address, Options) :-
	(   rdfpage_uri(Address)
	->  throw(loopback_harvest(Address))
	;   true
	),
	(   memberchk(priority(Priority), Options)
	->  true
	;   Priority=3
	),
	(   memberchk(retrieval(Ret), Options)
	->  true
	;   true
	),
	(   address_latest_successful_ret(Address, OldRet),
	    stale(OldRet, SecondsStale, Options),
	    SecondsStale < 0
	->  debug(webload, 'Still fresh from ~a', [Address]),
	    Ret=OldRet,
	    (   memberchk(intime(Done), Options)
	    ->  call(Done)
	    ;   true
	    )
	;   create_retrieval(Address, Ret, Options),
	    ret_set_status(Ret, retr:queued),
	    %ret_set(Ret, retr:queuePriorty, literal(Priority)),
	    %this_harvester(ThisHarvester),
	    %ret_set(Ret, retr:queueFor, ThisHarvester),
	    debug(webload, 'Created ret ~a', [Ret]),
	    (   memberchk(wait(Seconds), Options)
	    ->	thread_self(Me),
		thread_clear_all_messages,
		assertz(after_parse(Ret, thread_send_message(Me, done))),
		assertz(harvest_queue(Priority, Ret)),
		debug(webload, 'Ret enqueued; waiting ~w seconds', [Seconds]),
		interruptable_sleep(Seconds, Msg),
		debug(webload, 'Back from sleep; status=~w', [Msg]),
		(   memberchk(intime(Done), Options),
		    Msg = done
		->  call(Done)
		;   true
		),
		(   memberchk(timeout(NotDone), Options),
		    Msg = timeout
		->  call(NotDone)
		;   true
		),
		thread_clear_all_messages
	    ;   assertz(harvest_queue(Priority, Ret)),
		debug(webload, 'Ret enqueued; returning', [])
	    )
	).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

rdfq_provide(retr:inQueueFor, Ret, Harvester, _Q) :-
	% source?
	this_harvester(Harvester),
	harvest_queue(_Priority, Ret).
	
rdfq_provide(retr:queuePosition, Ret, literal(Pos), _Q) :-
	% source?
	harvest_queue_priority_position(Ret, _Priority, Pos).

rdfq_provide(retr:queuePriority, Ret, literal(Priority), _Q) :-
	% source?
	harvest_queue(Priority, Ret).

% Not retained after restart, which makes some sense.
rdfq_provide(retr:bytesTransferred, Ret, literal(Bytes), _Q) :-
	%  Bug WOrkaround.  Properly should make ret_file enumerable
	ground(Ret),
	ret_file(Ret, data, Filename),
	exists_file(Filename),
	size_file(Filename, Bytes).

rdfq_provide(retr:percentTransferred, Ret, literal(Percent), Q) :-
	rdfq(Q, Ret, retr:bytesTransferred, literal(Transferred)),
	rdfq(Q, Ret, retr:contentLength, literal(Length)),
	Length > 0,
	Percent is 100 * Transferred / Length.

rdfq_provide(retr:triplesSoFar, Ret, literal(Parsed), _Q) :-
	\+ ret_get(Ret, retr:triples, _),
	% see parse.pl
	total_retrieval_counter(Ret, Counter),
	flag(Counter, Parsed, Parsed).

rdfq_provide(retr:hostConnectionLimitWaitingFor, Ret, Other, _Q) :-
	harvest_queue(_Priority, Ret),
	ret_get(Ret, retr:host, literal(Host)),
	ret_host_is_full(Ret),
	host_contact_started(Host, _, Other).

rdfq_provide(retr:hostIdleWaitingAfter, Ret, Other, _Q) :-
	harvest_queue(_Priority, Ret),
	ret_get(Ret, retr:host, literal(Host)),
	ret_host_still_warm(Ret),
	host_contact_stopped(Host, _, Other).

rdfq_provide(retr:fractionOfTriplesIndexed, Ret, Fraction, _Q) :-
	ret_index_done_fraction(Ret, Fraction).

harvest_style_report(Q, Focus, Style, H) :-
	self_uri(harvester_status, Q, [focus=Focus], L1),
	concat_atom([L1, '#', Style], L),
	CountDone=0,
	CountToDo=0,
	Counted=fail,
	H=[Style, ': ', a(href=L, [int(CountDone), '/', int(CountToDo),
				  +if(\+ Counted, '+')])].

harvest_status(_Q, Focus, H) :-
	rdfpage_uri(Focus),
	H = 'Harvesting not applicable (this resource is internal)'.
harvest_status(Q, Focus, H) :-
	uri_stem(Focus, Stem),
	H = form(method=post, action=refresh,
		 input(type=hidden, name=focus, value=Focus),
		 input(type=hidden, name=address, value=Stem),
		 +harvester_controls(Q, Stem),
		'Harvester status: ',
		 +harvest_style_report(Q, Focus, deref), ', ',
		 +harvest_style_report(Q, Focus, seeAlso), ', ',
		 +harvest_style_report(Q, Focus, owl), ', ',
		 +harvest_style_report(Q, Focus, radius1), ', ',
		 +harvest_style_report(Q, Focus, radius2)
		 % input(style='float:right', type=submit, value='Initiate Refresh'),

		 /*
		 +if(address_latest_successful_ret(Stem, LSRet),
		     [ 'Latest successful retrieval: ', +ret_line(Q, LSRet), br([])]),

		 +if((address_latest_ret(Stem, LRet), LRet \= LSRet),
		     [ 'Latest retrieval attempt: ', +ret_line(Q, LRet)])
		   */
		).


ret_line(Q, Ret, H) :-
	ret_get_status(Ret, Status, Time),
	H = [ +word(Q, Ret), ', status: ', +word(Q,Status), ', ', +reltime(Time), '. ' ].


web_server_page(harvester_status).

harvester_status(Q, H) :-
	member(search(Search), Q),
	member(focus=Focus, Search),
	\+ Focus='',
	!,
	H = [ h2('Harvester status for "', Focus, '"'),
 	      p(a(href= \self_uri(harvester_status, Q, [focus='']), 'General harvester status')),
	      h3('Single-URI Status -- Not Implemented -- sorry!')
	    ].
harvester_status(Q, H) :-

	H = [ h2('Overall Harvester Status'),
	      p('Current information about the web harvester running as part of this server.'),
	      h3('In Progess'),
	      +doelse(+harvester_in_progress(Q), 'None'),
	      h3('Completed in last 60 seconds'),
	      +doelse(+harvester_recently_completed(Q), 'None'),
	      h3('On Queue'),
	      +doelse(+harvester_on_queue(Q), 'None')
	    ].


harvester_in_progress(Q, H) :-
	findall(Ret,
		harvest_in_progress(Ret),
		Rets
	       ),
	Rets \= [],
	H=  +forall(member(Ret, Rets),
		    +ret_on_queue(Q, Ret)).


harvester_recently_completed(Q, H) :-
	get_time(Now),
	findall(Ret,
		(   harvest_completed(Ret, Time),
		    Ago is Now - Time,
		    Ago < 60
		),
		Rets
	       ),
	Rets \= [],
	H=  +forall(member(Ret, Rets),
		    +ret_on_queue(Q, Ret)).

harvester_on_queue(Q, H) :-
	once(harvest_queue(_,_)),
	findall(Priority,
		harvest_queue(Priority, _),
	        UnsortedPriorities
	       ),
	sort(UnsortedPriorities, Priorities),
	H= +forall(member(Priority, Priorities),
		  +queue_table(Q, Priority)
		 ).
queue_table(Q, Priority, H) :-
	findall(Ret,
		harvest_queue(Priority, Ret),
		Rets),
	length(Rets, Count),
	H=[ h4(int(Count), ' Retrievals At Priority ', int(Priority), ':'),
	    +forall(member(Ret, Rets),
		    +ret_on_queue(Q, Ret))
	  ].

ret_on_queue(Q, Ret, H) :-
	ret_get(Ret, retr:initialResource, Address),
	H=[+word(Q, Ret),
	   ', address: ',
	   a(href=Address, Address), br([])
	  ].


	
 
harvester_controls(Q, Address, div(class=harvesterControls, H)) :-
	(   address_latest_successful_ret(Address, LSR)
	->  (   address_latest_ret(Address, LR)
	    ->  (  LR = LSR
		->  H=[ input(type=submit, value='Refresh Data') ]
		;   H=['Refresh in progress [',
		       a(href= \self_uri(basic, Q, [focus=LR]), details),
		       ']']
		)
	    ;   H='(error234234)'
	    )
	;   (   address_latest_ret(Address, LR)
	    ->  (   ret_done(LR)
		->  H=[ input(type=submit, value='Try Again') ]
		;   H=['Retrieval attempt in progress [',
		       a(href= \self_uri(basic, Q, [focus=LR]), details),
		       ']']
		)
	    ;   H=[ input(type=submit, value='Load Data') ]
	    )
	).
	    
		
	
