Mercurial > hg > dml-open-cliopatria
diff cpack/dml/lib/async.pl @ 0:718306e29690 tip
commiting public release
| author | Daniel Wolff |
|---|---|
| date | Tue, 09 Feb 2016 21:05:06 +0100 |
| parents | |
| children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/cpack/dml/lib/async.pl Tue Feb 09 21:05:06 2016 +0100 @@ -0,0 +1,469 @@ +/* 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).