view cpack/dml/lib/async.pl @ 0:718306e29690 tip

commiting public release

/* Part of DML (Digital Music Laboratory)
	Copyright 2014-2015 Samer Abdallah, University of London
	 
	This program is free software; you can redistribute it and/or
	modify it under the terms of the GNU General Public License
	as published by the Free Software Foundation; either version 2
	of the License, or (at your option) any later version.

	This program is distributed in the hope that it will be useful,
	but WITHOUT ANY WARRANTY; without even the implied warranty of
	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
	GNU General Public License for more details.

	You should have received a copy of the GNU General Public
	License along with this library; if not, write to the Free Software
	Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/

:- module(async, 
      [  async_memo/4
      ,  async_current_job/5
      ,  async_thread_progress/4
      ,  async_get_progress_callback/1
      ,  async_set_progress_callback/1
      ,  async_unset_progress_callback/0
      ,  async_cancel/3
      ,  async_cancel/2
      ,  with_progress_stack/1
      ,  with_cont/4
      ,  simple_task/2
      ,  stepwise_task/4
      ,  updatable_status_task/3
      ,  map_with_progress/3
      ,  map_fold_with_progress/5
      ,  with_global/3
      ,  with_new_ref/3
      ]).

/** <module> Asynchronous memoised computations with progress and partial results

   This module provides services for managing memoised computations in a thread pool
   with the ability to check on the progress of running jobs and receive partial
   results if they are available. Completed computations are memoised using
   library(memo).

   ---++ Job control
   The first half of the module is concerned with managing memoised computations.
   Computations are requested using async_memo/4. If the result is already available in
   the memoised database, the result is returned. If the computation is already running,
   status information about the job is returned. Otherwise, a new job is started. Status
   information is returned as the type =|async_status(A)|=.
   ==
   async_status(A) ---> done(metadata)
                      ; initiated(job_id,natural)
                      ; waiting(natural)
                      ; running(job_id,A))
                      ; recompute(job_id,natural,metadata).

   job_id == atom.
   ==
   The type =|metadata|= is defined in library(memo).
   

   ---++ Helper meta-predicates
   The second half of the module provides a collection of predicates for computations
   to use to handle requests for progress information, in the form of a stack of tasks. 
   It recognises the following levels of progress information:
   ==
   summary :: progress_level(list(any)).
   partial_result :: progress_level(any).
   ==
*/
:- meta_predicate 
         async_set_progress_callback(2)
      ,  async_memo(+,0,-,+)
      ,  with_global(+,+,0)
      ,  with_new_ref(-,+,0)
      ,  with_progress_stack(0)
      ,  with_progress_handler(2,0)
      ,  with_cont(+,1,2,-)
      ,  simple_task(+,0)
      ,  stepwise_task(+,+,-,0)
      % ,  stepwise_partial_task(+,+,-,-,0)
      % ,  fold_with_progress(3,+,+,-)
      ,  map_with_progress(2,+,-)
      ,  map_fold_with_progress(2,3,+,+,-).

:- use_module(library(thread_pool)).
:- use_module(library(insist)).
:- use_module(library(memo)).
:- use_module(library(dcg_core),[seqmap/4]).

%% async_memo(+P:thread_pool, +G:memo_goal, -S:async_status(_), Opts:options) is det.
%
%  Start running a goal asynchronously in a thread pool if it is not
%  already running. 
%  In order for the goal to provide progress information, it must use nb_setval/2
%  to set the global variable =|progress|=.
%
%  Options:
%     * status_var(Var:status_var(A))
%       If computation is already running, information of type =|A|=
%       associated with job is returned in Stats. Default is =|progress(summary)|=.
%     * recompute(R:oneof([none,failed,test(Pred)]))
%       If =|none|=, then a previously failed computation is not retried, and the
%       status =|done(Comp-Result)|= is returned, where Comp includes
%       the computation time and host and Result is =|fail|= or =|ex(Exception)|=.
%       If =|failed|=, unsuccessful computations are retried and the status
%       =|recomputing(ID,Pos,Meta)|= is returned, where =|ID|= is the ID of the
%       new job, =|Pos|= its initial position in the queue, and =|Meta=Comp-Result|=
%       the status of the previous computation.
%       If =|test(Pred)|=, then the computation is retried if =|call(Pred,Meta)|=
%       succeeds, where Meta is the memoisied computation metadata and the variables
%       in Goal will be unified with their values resulting from that computation.
%       The status returned is =|initiated(ID,Pos)|=. Pred must include its module
%       specifier as a I am too lazy to worry about meta-predicate options right now.
%     * globals(list(pair(atom,any)))
%       Set global variables to given values in new thread.
%
%  The value of the =|status_var(_)|= option determines the information returned
%  if the status of the computation is =|running(ID,StatusVals)|=.
%  ==
%  progress(progress_level(A)) :: status_var(A).
%  stats(list(stats_key))      :: status_var(list(stats_vals)).
%  time                        :: status_var(pair(time,float)). % start and elapsed times
%  status_var(A)-status_var(B) :: status_var(pair(A,B)).
%
%  time==float. % as returned by get_time/1.
%  ==
%  The type =|stats_key|= denotes any atom accepted by thread_statistics/3, and
%  =|stats_vals|= the type of values returned.

async_memo(_,Goal,done(Meta),Opts) :- 
   option(recompute(none),Opts,none),
   browse(Goal,Meta), !.

async_memo(Pool,Goal,Status,Opts) :- 
   with_mutex(async,unsafe_async_memo(Pool,Goal,Status,Opts)).

unsafe_async_memo(Pool,Goal,Status,Opts) :-
   option(globals(Globals),Opts,[]),
   (  browse(Goal,Meta)
   -> option(recompute(Recompute),Opts,none),
      (  recompute(Recompute,Meta) 
      -> clear_all(Goal,Meta),
         spawn(Pool,Goal,Globals,ID,Waiting),
         Status=recomputing(ID,Waiting,Meta)
      ;  Status=done(Meta)
      )
   ;  % !!! it's possible that another thread might be computing
      % this goal and complete at this point. If this happens, 
      % that job will not be detected and a new a job will be started
      % here. If that happens, the newly created job will find the result
      % and complete quickly.
      (  async_current_job(Pool,Goal,ID,T0,S) 
      -> option(progress_levels(Levels),Opts,[summary]),
         % !!! it's possible that the job will complete here
         % and so the progress request will create an error
         goal_status(ID,T0,Levels,S,Status)
      ;  spawn(Pool,Goal,Globals,ID,Waiting), 
         Status=spawned(ID,Waiting)
      )
   ).

recompute(failed,_-fail).
recompute(failed,_-ex(_)).
recompute(test(P),M) :- call(P,M).

goal_status(ID, TSubmit, _, waiting(Pos),           waiting(ID,TSubmit,Pos)).
goal_status(ID, _, Levels,  running(Thread,TStart), running(ID,TStart,Thread,Progress)) :-
   catch( ( async_thread_progress(Thread,Levels,T,Vals), Progress=just(T-Vals)), Ex, 
          ( print_message(warning,error_getting_job_progress(Thread,Ex)), Progress=nothing)).

spawn(Pool,Goal,Globals,ID,Waiting) :-
   uuid(ID),
   (thread_pool_property(Pool,backlog(Waiting)) -> true; Waiting=0),
   get_time(Time),
   setup_call_catcher_cleanup(
      recordz(Pool,job(ID,Time,Goal),Ref),
      thread_create_in_pool(Pool,spawnee(ID,Goal,Globals),_,
                            [ detached(true), at_exit(erase(Ref))]),
      exception(_),
      erase(Ref)).

spawnee(ID,Goal,Globals) :- 
   setup_call_cleanup(
      with_mutex(async, 
         (  recorded(ID,async:cancelled,Ref)
         -> debug(async,'Cancelled before starting: ~q',[Goal]),
            erase(Ref), fail
         ;  pairs_keys_values(Globals,GNames,GVals),
            maplist(nb_setval,GNames,GVals),
            get_time(T0),
            thread_self(Thread),
            recordz(ID,running(Thread,T0),Ref)
         )),
      ( freeze(Res,check_result(Res)), % prevent memoisation if abort(_) is caught
         debug(async,"Running asyc goal ~q",[Goal]),
         memo(Goal,_-Res)),
      erase(Ref)).

check_result(Ex) :-
   (  Ex=ex(abort(Reason)) 
   -> debug(async,'Computation aborted (~w)',[Reason]),
      throw(abort(Reason))
   ;  true
   ).

%% async_current_job(+Pool:thread_pool, -Goal:callable, -ID:job_id, -T:time, -S:job_status) is nondet.
%  Retrieves information about current pending jobs. A job is pending if it is running
%  or waiting to run in the specified thread pool. T is the time the job was submitted.
%  Job status is:
%  ==
%  job_status ---> waiting(natural)     % queued at given postion
%                ; running(thread_id,time). % with start time
%  ==
async_current_job(Pool,Goal,ID,Time,Status) :-
   recorded(Pool,job(ID,Time,Goal)), % this finds ALL jobs
   (  recorded(ID,running(Thread,StartTime)) 
   -> Status=running(Thread,StartTime)
   ;  thread_pool_property(Pool,waiting(Waiting)),
      nth1(Pos,Waiting,create(Pool,async:spawnee(ID,_,_),_,_,_)) 
   -> Status=waiting(Pos)
   ;  print_message(warning, async_job_status_indeterminate(Pool,ID,Goal)),
      Status=indeterminate % could be finished, or transferring from waiting to running
   ).


%% async_thread_progress(+Thread:thread_id,+Levels:list(progress_level), -T:time, -Vals:list) is det.
%  Called in main thread to get information about job running in thread Thread.  
%  Will throw an exception if:
%     -  the thread has terminated
%     -  the thread does not respond with progress info in less than 3 seconds
%     -  the thread does not have a progress handler call back registered
%     -  the progress handler fails on any of the progress levels requested.
async_thread_progress(Thread,Levels,Time,Vals) :-
   thread_self(Me),
   gensym(prog,RID), % progress request ID
   flush_all_messages(Me,progress(_,_,_,_)),
   thread_signal(Thread,report_progress(RID,Levels,Me)),
   (  thread_get_message(Me,progress(RID,Time,Vals,St),[timeout(3)]) -> true
   ;  throw(get_thread_progress_timeout(Thread,Levels))
   ),
   debug(async,'Got progress ~w from thread ~w: ~w',[Levels,Thread,Vals]),
   memo:reflect(St).

flush_all_messages(Queue,Pattern) :-
   repeat, \+thread_get_message(Queue,Pattern,[timeout(0)]), !.

%% report_progress(+RID:atomic, +Var:status_var(A), +Thread:thread_id) is det.
%  Called from job thread in response to a signal, to send progress information to thread Thread.
report_progress(RID,Levels,Thread) :-
   get_time(Time),
   memo:reify(async:my_progress(Levels,Vals),St), !,
   debug(async,'reporting progress(~w): ~q',[Levels,Vals]),
   thread_send_message(Thread,progress(RID,Time,Vals,St)).

my_progress(Levels,Vals) :- nb_getval(async_progress,G), insist(maplist(G,Levels,Vals)).  


%% async_cancel(+Pool:thread_pool, +ID:job_id, +Ex:term) is det.
%% async_cancel(+Pool:thread_pool, +ID:job_id) is det.
%  Cancel job with given ID, if it is currently pending. If it is running, the
%  exception term Ex is thrown (=|abort(cancel)|= in async_cancel/2).
%  If not, the recorded database is used to mark the job for immediate failure 
%  as soon as it is started.
async_cancel(Pool,ID) :- async_cancel(Pool,ID,abort(cancel)).
async_cancel(Pool,ID,Ex) :-
   debug(async,'Cancelling ~w:~w with ~w',[Pool,ID,Ex]),
   with_mutex(async, (async_current_job(Pool,_,ID,_,S) -> cancel(ID,S,Ex); true)).

cancel(ID,waiting(_),_) :- recordz(ID,async:cancelled).
cancel(_, running(Thread,_),Ex) :- 
   catch( thread_signal(Thread,throw(Ex)), E,
          print_message(warning,error_signalling_job_thread(Thread,E))). 
cancel(ID,indeterminate,_) :- print_message(warning,indeterminate_job_state_on_cancel(ID)).

%% async_set_progress_callback(+P:pred(+Level:any,-Info:any)) is det.
%  Sets the progress handler for the current thread.
async_set_progress_callback(Pred) :- nb_setval(async_progress,Pred).

%% async_get_progress_callback(-P:pred(+Level:any,-Info:any)) is semidet.
%  Gets the progress handler for the current thread, if one has been set.
async_get_progress_callback(Pred) :- nb_current(async_progress,Pred).

%% async_unset_progress_callback is det.
%  Unsets the progress handler for the current thread. Subsequent calls to 
%  async_get_progress_callback/1 will fail.
async_unset_progress_callback :- nb_delete(async_progress).

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


%% with_progress_stack(+Goal:callable) is nondet.
%
%  Runs Goal with the current thread's progress handler (which must be initially
%  unset) set to an empty task stack. Tasks (goals which run in cooperation with a
%  a progress handler) can be pushed on to the task stack using with_progress_handler/2.
%  Fails if progress callback has already been set to something else.
%
%  The progress handler recognises progress level terms:
%     *  summary :: progress_level(list(any))
%        Yields a list of progress terms, one for each entry in the task stack.
%     *  partial_result :: progress_level(any)
%        Yields partial results of the computation if they are available.
with_progress_stack(Goal) :-
   async_get_progress_callback(CB), !,
   CB=async:get_stack_progress,
   call(Goal).
with_progress_stack(Goal) :-
   with_global(async_progress_stack, nil,
      setup_call_cleanup( 
         async_set_progress_callback(get_stack_progress), Goal,
         async_unset_progress_callback)).

%% get_stack_progress(+Level:progress_level(A),-Info:A) is semidet.
%  Gets the request level of progress information from the currently installed
%  progress stack.
get_stack_progress(summary,P) :-
   nb_getval(async_progress_stack,H),
   call(H,progress(summary),P).

get_stack_progress(partial_result,R) :-
   nb_getval(async_progress_stack,H),
   call(H,partial_result,R).


%% with_progress_handler(+H:progress_handler, +Goal:callable) is nondet.
%  Pushes the progress handler H to the top of the task stack associated with
%  the current thread and calls Goal. The handler is removed on exit.
with_progress_handler(H,Goal) :- 
   nb_getval(async_progress_stack,H0), call(H0,push(H),H1),
   setup_call_cleanup( nb_setval(async_progress_stack,H1), Goal,
                       nb_setval(async_progress_stack,H0)).

nil(progress(_),[]).
nil(push(H),chain(H,nil)).
nil(partial_result,nothing).
nil(partial_result(Cont),R) :- call(Cont,nothing,R).

chain(Head,Tail1,push(H),chain(Head,Tail2)) :- call(Tail1,push(H),Tail2).
chain(Head,Tail,progress(L),[PH|PT]) :- 
   call(Head,progress(L),PH), 
   call(Tail,progress(L),PT).
chain(Head,Tail,partial_result,MR) :- 
   call(Tail,partial_result,R1),
   (call(Head,partial_result(R1),R) -> MR=just(R); MR=nothing).


%% simple_task(+Desc,+Goal:callable) is det.
%
%  Calls goal with a progress handler that reports Desc when progress
%  information is requested. No partial results are available.
simple_task(Desc,Goal) :- with_progress_handler(simple_progress(Desc),Goal).

simple_progress(Desc,progress(_),Desc).


%% with_cont(+Desc,+Pred:pred(-A),+Cont:pred(+A,-B),-Result:B) is det.
%
%  Calls Pred and then passes the result to Cont yielding Result.
%  A progress handler is installed such that any partial results yielded by
%  subtasks in Pred are passed to Cont to yield a partial result. Progress requests
%  yield Desc.
with_cont(Desc,Pred,Cont,Result) :- 
   with_progress_handler(cont(Desc,Cont),
      (call(Pred,R),call(Cont,R,Result))).

cont(_,Cont,Tail1,push(H),cont(Cont,Tail2)) :- call(Tail1,push(H),Tail2).
cont(Desc,_,progress(_),Desc).
cont(_,Cont,partial_result(just(R1)),R) :- call(Cont,R1,R).

%% stepwise_task(+Desc,+Total:nonneg,-Step:callable,+Goal:callable) is det.
%  
%  Calls Goal with a progress handler that is able to report progress as a term
%  =|stepwise(Desc,Done/Total)|=. Goal is permitted to call Step in order to increment
%  the =|Done|= counter. Partial results are not available.
stepwise_task(Desc,Total,async:increment(Counter),Goal) :-
   with_new_ref(Counter,0, with_progress_handler(stepwise_progress(Desc,Total,Counter), Goal)).

stepwise_progress(Desc,Total,Counter,progress(_),stepwise(Desc,Done/Total)) :- b_getval(Counter,Done).


%% updatable_status_task(+Initial,-Update:pred(any),+Goal:callable) is det.
%
%  Calls goal with Update set to a predicate which can be used to set the
%  current status of the task. This will be reported if progress is requested.
%  No partial results are available. Initial is the status before Update is called.
:- meta_predicate updatable_status_task(+,-,0).
updatable_status_task(Initial,async:set_ref(Status),Goal) :-
   with_new_ref(Status,Initial, with_progress_handler(status_progress(Status),Goal)).

status_progress(Status,progress(_),S) :- b_getval(Status,S).


map_with_progress(P,Xs,Ys) :-
   length(Xs,Total),
   stepwise_task(map(P),Total,Step, maplist(call_and_step(Step,P),Xs,Ys)).

call_and_step(Step,P,X,Y) :- call(P,X,Y), call(Step).

/*
fold_with_progress(Folder,Items,S0,S1) :-
   length(Items,N),
   stepwise_partial_task(folding(Folder),N,Step,Update,
      seqmap(step_update(Step,Update,Folder),Items,S0,S1)).

step_update(Step,Update,P,X,S0,S1) :-
   call(P,X,S0,S1),
   call(Update,S1),
   call(Step).
*/

:- setting(map_fold_batch,nonneg,512,"Batch size for flushing fold queue.").

%% map_fold_with_progress(+Mapper:pred(A,B), +Folder:pred(list(B),S,S), +Items:list(A), +S1:S, -S2:S) is det.
%
%  Map Mapper over Items and fold over the results with Folder, with progress information
%  partial results available. Note that Folder operates on a _list_ of items, not just one
%  item at a time. Mapper is used to accumulate up to N items before folding them batchwise
%  into the accumulator, where N is the value of the setting =|map_fold_batch|=.
% 
%  Progress information is returned as a term =|stepwise(Desc,Done/Total)|=, where
%  =|Desc=map_fold(Mapper,Folder)|=. If partial results are requested, the current batch
%  of mapped items is first folded in before returning the value of the accumulator.
map_fold_with_progress(Mapper,Folder,Items,S1,S2) :-
   length(Items,Total),
   setting(map_fold_batch,M1),
   with_new_ref(StateRef,s(Items,0,M1,Head,Head,S1),
      (  Handler=map_fold_progress(map_fold(Mapper,Folder),Total,StateRef),
         with_progress_handler(Handler,
            (  map_fold_loop(Mapper,Folder,StateRef),
               call(Handler,partial_result(nothing),S2))))).

map_fold_loop(Mapper,Folder,StateRef) :-
   b_getval(StateRef,s(Xs1,N1,M1,Head,Ys1,S)), 
   (  Xs1=[] -> Ys1=[]
   ;  Xs1=[X|Xs2] 
   -> call(Mapper,X,Y), !, 
      Ys1=[Y|Ys2], succ(N1,N2), 
      State1=s(Xs2,N2,M2,Head,Ys2,S), 
      (  succ(M2,M1) -> State2=State1
      ;  simple_task(folding,flush_fold(Folder,State1,State2))
      ),
      b_setval(StateRef,State2), !,
      map_fold_loop(Mapper,Folder,StateRef)
   ).

map_fold_progress(Desc,Total,StateRef,progress(_),stepwise(Desc,Done/Total)) :-
   b_getval(StateRef,s(_,Done,_,_,_,_)).
map_fold_progress(map_fold(_,Folder),_,StateRef,partial_result(_),S2) :-
   b_getval(StateRef,State1),
   flush_fold(Folder,State1,State2),
   b_setval(StateRef,State2),
   State2=s(_,_,_,_,_,S2).

flush_fold(Folder,s(Xs,Done,_,Head,[],S1),s(Xs,Done,M1,H2,H2,S2)) :-
   setting(map_fold_batch,M1),
   call(Folder,Head,S1,S2), !.

% for managing global variable
new_ref(Ref) :- gensym('$ref',Ref).
new_ref(Ref,Val) :- new_ref(Ref), b_setval(Ref,Val).

with_global(Name,Val,Goal) :- setup_call_cleanup(b_setval(Name,Val), Goal, nb_delete(Name)).
with_new_ref(Name,Val,Goal) :- new_ref(Name), with_global(Name,Val,Goal).

set_ref(Ref,Val) :- nb_setval(Ref,Val).
increment(Counter) :- b_getval(Counter,N), succ(N,M), b_setval(Counter,M).