1:- encoding(utf8).
    2/*  Part of SWI-Prolog
    3
    4    Author:        Torbjörn Lager and Jan Wielemaker
    5    E-mail:        J.Wielemaker@vu.nl
    6    WWW:           http://www.swi-prolog.org
    7    Copyright (C): 2014-2020, Torbjörn Lager,
    8                              VU University Amsterdam
    9    All rights reserved.
   10
   11    Redistribution and use in source and binary forms, with or without
   12    modification, are permitted provided that the following conditions
   13    are met:
   14
   15    1. Redistributions of source code must retain the above copyright
   16       notice, this list of conditions and the following disclaimer.
   17
   18    2. Redistributions in binary form must reproduce the above copyright
   19       notice, this list of conditions and the following disclaimer in
   20       the documentation and/or other materials provided with the
   21       distribution.
   22
   23    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
   24    "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
   25    LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
   26    FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
   27    COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
   28    INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
   29    BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
   30    LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
   31    CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
   32    LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
   33    ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   34    POSSIBILITY OF SUCH DAMAGE.
   35*/
   36
   37:- module(pengines,
   38          [ pengine_create/1,                   % +Options
   39            pengine_ask/3,                      % +Pengine, :Query, +Options
   40            pengine_next/2,                     % +Pengine. +Options
   41            pengine_stop/2,                     % +Pengine. +Options
   42            pengine_event/2,                    % -Event, +Options
   43            pengine_input/2,                    % +Prompt, -Term
   44            pengine_output/1,                   % +Term
   45            pengine_respond/3,                  % +Pengine, +Input, +Options
   46            pengine_debug/2,                    % +Format, +Args
   47            pengine_self/1,                     % -Pengine
   48            pengine_pull_response/2,            % +Pengine, +Options
   49            pengine_destroy/1,                  % +Pengine
   50            pengine_destroy/2,                  % +Pengine, +Options
   51            pengine_abort/1,                    % +Pengine
   52            pengine_application/1,              % +Application
   53            current_pengine_application/1,      % ?Application
   54            pengine_property/2,                 % ?Pengine, ?Property
   55            pengine_user/1,                     % -User
   56            pengine_event_loop/2,               % :Closure, +Options
   57            pengine_rpc/2,                      % +Server, :Goal
   58            pengine_rpc/3                       % +Server, :Goal, +Options
   59          ]).

   70:- autoload(library(aggregate),[aggregate_all/3]).   71:- autoload(library(apply),[maplist/2,partition/4,exclude/3,maplist/3]).   72:- autoload(library(broadcast),[broadcast/1]).   73:- autoload(library(charsio),[open_chars_stream/2]).   74:- autoload(library(debug),[debug/1,debugging/1,debug/3,assertion/1]).   75:- autoload(library(error),
   76	    [ must_be/2,
   77	      existence_error/2,
   78	      permission_error/3,
   79	      domain_error/2
   80	    ]).   81:- autoload(library(filesex),[directory_file_path/3]).   82:- autoload(library(listing),[listing/1]).   83:- autoload(library(lists),[member/2,flatten/2,select/3,append/3]).   84:- autoload(library(modules),[in_temporary_module/3]).   85:- autoload(library(occurs),[sub_term/2]).   86:- autoload(library(option),
   87	    [select_option/3,option/2,option/3,select_option/4]).   88:- autoload(library(prolog_stack),[print_prolog_backtrace/2]).   89:- autoload(library(sandbox),[safe_goal/1]).   90:- autoload(library(statistics),[thread_statistics/2]).   91:- autoload(library(term_to_json),[term_to_json/2]).   92:- autoload(library(thread_pool),
   93	    [thread_pool_create/3,thread_create_in_pool/4]).   94:- autoload(library(time),[alarm/4,call_with_time_limit/2]).   95:- autoload(library(uri),
   96	    [ uri_components/2,
   97	      uri_query_components/2,
   98	      uri_data/3,
   99	      uri_data/4,
  100	      uri_encoded/3
  101	    ]).  102:- autoload(library(http/http_client),[http_read_data/3]).  103:- autoload(library(http/http_cors),[cors_enable/0,cors_enable/2]).  104:- autoload(library(http/http_dispatch),
  105	    [http_handler/3,http_404/2,http_reply_file/3]).  106:- autoload(library(http/http_open),[http_open/3]).  107:- autoload(library(http/http_parameters),[http_parameters/2]).  108:- autoload(library(http/http_stream),[is_cgi_stream/1]).  109:- autoload(library(http/http_wrapper),[http_peer/2]).  110
  111:- use_module(library(settings),[setting/2,setting/4]).  112:- use_module(library(http/http_json),
  113              [http_read_json_dict/2,reply_json/1]).  114
  115:- if(exists_source(library(uuid))).  116:- autoload(library(uuid), [uuid/2]).  117:- endif.  118
  119
  120:- meta_predicate
  121    pengine_create(:),
  122    pengine_rpc(+, +, :),
  123    pengine_event_loop(1, +).  124
  125:- multifile
  126    write_result/3,                 % +Format, +Event, +Dict
  127    event_to_json/3,                % +Event, -JSON, +Format
  128    prepare_module/3,               % +Module, +Application, +Options
  129    prepare_goal/3,                 % +GoalIn, -GoalOut, +Options
  130    authentication_hook/3,          % +Request, +Application, -User
  131    not_sandboxed/2.                % +User, +App
  132
  133:- predicate_options(pengine_create/1, 1,
  134                     [ id(-atom),
  135                       alias(atom),
  136                       application(atom),
  137                       destroy(boolean),
  138                       server(atom),
  139                       ask(compound),
  140                       template(compound),
  141                       chunk(integer),
  142                       bindings(list),
  143                       src_list(list),
  144                       src_text(any),           % text
  145                       src_url(atom),
  146                       src_predicates(list)
  147                     ]).  148:- predicate_options(pengine_ask/3, 3,
  149                     [ template(any),
  150                       chunk(integer),
  151                       bindings(list)
  152                     ]).  153:- predicate_options(pengine_next/2, 2,
  154                     [ chunk(integer),
  155                       pass_to(pengine_send/3, 3)
  156                     ]).  157:- predicate_options(pengine_stop/2, 2,
  158                     [ pass_to(pengine_send/3, 3)
  159                     ]).  160:- predicate_options(pengine_respond/3, 2,
  161                     [ pass_to(pengine_send/3, 3)
  162                     ]).  163:- predicate_options(pengine_rpc/3, 3,
  164                     [ chunk(integer),
  165                       pass_to(pengine_create/1, 1)
  166                     ]).  167:- predicate_options(pengine_send/3, 3,
  168                     [ delay(number)
  169                     ]).  170:- predicate_options(pengine_event/2, 2,
  171                     [ listen(atom),
  172                       pass_to(thread_get_message/3, 3)
  173                     ]).  174:- predicate_options(pengine_pull_response/2, 2,
  175                     [ pass_to(http_open/3, 3)
  176                     ]).  177:- predicate_options(pengine_event_loop/2, 2,
  178                     []).                       % not yet implemented
  179
  180% :- debug(pengine(transition)).
  181:- debug(pengine(debug)).               % handle pengine_debug in pengine_rpc/3.
  182
  183goal_expansion(random_delay, Expanded) :-
  184    (   debugging(pengine(delay))
  185    ->  Expanded = do_random_delay
  186    ;   Expanded = true
  187    ).
  188
  189do_random_delay :-
  190    Delay is random(20)/1000,
  191    sleep(Delay).
  192
  193:- meta_predicate                       % internal meta predicates
  194    solve(+, ?, 0, +),
  195    findnsols_no_empty(+, ?, 0, -),
  196    pengine_event_loop(+, 1, +).

 pengine_create(:Options) is det

Creates a new pengine. Valid options are:

id(-ID)

ID gets instantiated to the id of the created pengine. ID is atomic.

alias(+Name)

The pengine is named Name (an atom). A slave pengine (child) can subsequently be referred to by this name.

application(+Application)

Application in which the pengine runs. See pengine_application/1.

server(+URL)

The pengine will run in (and in the Prolog context of) the pengine server located at URL.

src_list(+List_of_clauses)

Inject a list of Prolog clauses into the pengine.

src_text(+Atom_or_string)

Inject the clauses specified by a source text into the pengine.

src_url(+URL)

Inject the clauses specified in the file located at URL into the pengine.

src_predicates(+List)

Send the local predicates denoted by List to the remote pengine. List is a list of predicate indicators.

Remaining options are passed to http_open/3 (meaningful only for non-local pengines) and thread_create/3. Note that for thread_create/3 only options changing the stack-sizes can be used. In particular, do not pass the detached or alias options..

Successful creation of a pengine will return an event term of the following form:

create(ID, Term)

ID is the id of the pengine that was created. Term is not used at the moment.

An error will be returned if the pengine could not be created:

error(ID, Term)

ID is invalid, since no pengine was created. Term is the exception's error term. */

  251pengine_create(M:Options0) :-
  252    translate_local_sources(Options0, Options, M),
  253    (   select_option(server(BaseURL), Options, RestOptions)
  254    ->  remote_pengine_create(BaseURL, RestOptions)
  255    ;   local_pengine_create(Options)
  256    ).

 translate_local_sources(+OptionsIn, -Options, +Module) is det

Translate the src_predicates and src_list options into src_text. We need to do that anyway for remote pengines. For local pengines, we could avoid this step, but there is very little point in transferring source to a local pengine anyway as local pengines can access any Prolog predicate that you make visible to the application.

Multiple sources are concatenated to end up with a single src_text option.

  270translate_local_sources(OptionsIn, Options, Module) :-
  271    translate_local_sources(OptionsIn, Sources, Options2, Module),
  272    (   Sources == []
  273    ->  Options = Options2
  274    ;   Sources = [Source]
  275    ->  Options = [src_text(Source)|Options2]
  276    ;   atomics_to_string(Sources, Source)
  277    ->  Options = [src_text(Source)|Options2]
  278    ).
  279
  280translate_local_sources([], [], [], _).
  281translate_local_sources([H0|T], [S0|S], Options, M) :-
  282    nonvar(H0),
  283    translate_local_source(H0, S0, M),
  284    !,
  285    translate_local_sources(T, S, Options, M).
  286translate_local_sources([H|T0], S, [H|T], M) :-
  287    translate_local_sources(T0, S, T, M).
  288
  289translate_local_source(src_predicates(PIs), Source, M) :-
  290    must_be(list, PIs),
  291    with_output_to(string(Source),
  292                   maplist(list_in_module(M), PIs)).
  293translate_local_source(src_list(Terms), Source, _) :-
  294    must_be(list, Terms),
  295    with_output_to(string(Source),
  296                   forall(member(Term, Terms),
  297                          format('~k .~n', [Term]))).
  298translate_local_source(src_text(Source), Source, _).
  299
  300list_in_module(M, PI) :-
  301    listing(M:PI).

 pengine_send(+NameOrID, +Term) is det

Same as pengine_send(NameOrID, Term, []). */

  308pengine_send(Target, Event) :-
  309    pengine_send(Target, Event, []).

 pengine_send(+NameOrID, +Term, +Options) is det

Succeeds immediately and places Term in the queue of the pengine NameOrID. Options is a list of options:

delay(+Time)

The actual sending is delayed by Time seconds. Time is an integer or a float.

Any remaining options are passed to http_open/3. */

  324pengine_send(Target, Event, Options) :-
  325    must_be(atom, Target),
  326    pengine_send2(Target, Event, Options).
  327
  328pengine_send2(self, Event, Options) :-
  329    !,
  330    thread_self(Queue),
  331    delay_message(queue(Queue), Event, Options).
  332pengine_send2(Name, Event, Options) :-
  333    child(Name, Target),
  334    !,
  335    delay_message(pengine(Target), Event, Options).
  336pengine_send2(Target, Event, Options) :-
  337    delay_message(pengine(Target), Event, Options).
  338
  339delay_message(Target, Event, Options) :-
  340    option(delay(Delay), Options),
  341    !,
  342    alarm(Delay,
  343          send_message(Target, Event, Options),
  344          _AlarmID,
  345          [remove(true)]).
  346delay_message(Target, Event, Options) :-
  347    random_delay,
  348    send_message(Target, Event, Options).
  349
  350send_message(queue(Queue), Event, _) :-
  351    thread_send_message(Queue, pengine_request(Event)).
  352send_message(pengine(Pengine), Event, Options) :-
  353    (   pengine_remote(Pengine, Server)
  354    ->  remote_pengine_send(Server, Pengine, Event, Options)
  355    ;   pengine_thread(Pengine, Thread)
  356    ->  thread_send_message(Thread, pengine_request(Event))
  357    ;   existence_error(pengine, Pengine)
  358    ).

 pengine_request(-Request) is det

To be used by a pengine to wait for the next request. Such messages are placed in the queue by pengine_send/2. Keeps the thread in normal state if an event arrives within a second. Otherwise it waits for the idle_limit setting while using thread_idle/2 to minimis resources.

  368pengine_request(Request) :-
  369    thread_self(Me),
  370    thread_get_message(Me, pengine_request(Request), [timeout(1)]),
  371    !.
  372pengine_request(Request) :-
  373    pengine_self(Self),
  374    get_pengine_application(Self, Application),
  375    setting(Application:idle_limit, IdleLimit0),
  376    IdleLimit is IdleLimit0-1,
  377    thread_self(Me),
  378    (   thread_idle(thread_get_message(Me, pengine_request(Request),
  379                                       [timeout(IdleLimit)]),
  380                    long)
  381    ->  true
  382    ;   Request = destroy
  383    ).

 pengine_reply(+Event) is det

 pengine_reply(+Queue, +Event) is det

Reply Event to the parent of the current Pengine or the given Queue. Such events are read by the other side with pengine_event/1.

If the message cannot be sent within the idle_limit setting of the pengine, abort the pengine.

  396pengine_reply(Event) :-
  397    pengine_parent(Queue),
  398    pengine_reply(Queue, Event).
  399
  400pengine_reply(_Queue, _Event0) :-
  401    nb_current(pengine_idle_limit_exceeded, true),
  402    !.
  403pengine_reply(Queue, Event0) :-
  404    arg(1, Event0, ID),
  405    wrap_first_answer(ID, Event0, Event),
  406    random_delay,
  407    debug(pengine(event), 'Reply to ~p: ~p', [Queue, Event]),
  408    (   pengine_self(ID),
  409        \+ pengine_detached(ID, _)
  410    ->  get_pengine_application(ID, Application),
  411        setting(Application:idle_limit, IdleLimit),
  412        debug(pengine(reply), 'Sending ~p, timout: ~q', [Event, IdleLimit]),
  413        (   thread_send_message(Queue, pengine_event(ID, Event),
  414                                [ timeout(IdleLimit)
  415                                ])
  416        ->  true
  417        ;   thread_self(Me),
  418            debug(pengine(reply), 'pengine_reply: timeout for ~q (thread ~q)',
  419                  [ID, Me]),
  420            nb_setval(pengine_idle_limit_exceeded, true),
  421            thread_detach(Me),
  422            abort
  423        )
  424    ;   thread_send_message(Queue, pengine_event(ID, Event))
  425    ).
  426
  427wrap_first_answer(ID, Event0, CreateEvent) :-
  428    wrap_first_answer_in_create_event(CreateEvent, [answer(Event0)]),
  429    arg(1, CreateEvent, ID),
  430    !,
  431    retract(wrap_first_answer_in_create_event(CreateEvent, [answer(Event0)])).
  432wrap_first_answer(_ID, Event, Event).
  433
  434
  435empty_queue :-
  436    pengine_parent(Queue),
  437    empty_queue(Queue, 0, Discarded),
  438    debug(pengine(abort), 'Abort: discarded ~D messages', [Discarded]).
  439
  440empty_queue(Queue, C0, C) :-
  441    thread_get_message(Queue, _Term, [timeout(0)]),
  442    !,
  443    C1 is C0+1,
  444    empty_queue(Queue, C1, C).
  445empty_queue(_, C, C).

 pengine_ask(+NameOrID, @Query, +Options) is det

Asks pengine NameOrID a query Query.

Options is a list of options:

template(+Template)

Template is a variable (or a term containing variables) shared with the query. By default, the template is identical to the query.

chunk(+Integer)

Retrieve solutions in chunks of Integer rather than one by one. 1 means no chunking (default). Other integers indicate the maximum number of solutions to retrieve in one chunk.

bindings(+Bindings)

Sets the global variable '$variable_names' to a list of Name = Var terms, providing access to the actual variable names.

Any remaining options are passed to pengine_send/3.

Note that the predicate pengine_ask/3 is deterministic, even for queries that have more than one solution. Also, the variables in Query will not be bound. Instead, results will be returned in the form of event terms.

success(ID, Terms, Projection, Time, More)

ID is the id of the pengine that succeeded in solving the query. Terms is a list holding instantiations of Template. Projection is a list of variable names that should be displayed. Time is the CPU time used to produce the results and finally, More is either true or false, indicating whether we can expect the pengine to be able to return more solutions or not, would we call pengine_next/2.

failure(ID)

ID is the id of the pengine that failed for lack of a solutions.

error(ID, Term)

ID is the id of the pengine throwing the exception. Term is the exception's error term.

output(ID, Term)

ID is the id of a pengine running the query that called pengine_output/1. Term is the term that was passed in the first argument of pengine_output/1 when it was called.

prompt(ID, Term)

ID is the id of the pengine that called pengine_input/2 and Term is the prompt.

Defined in terms of pengine_send/3, like so:

pengine_ask(ID, Query, Options) :-
    partition(pengine_ask_option, Options, AskOptions, SendOptions),
    pengine_send(ID, ask(Query, AskOptions), SendOptions).

*/

  510pengine_ask(ID, Query, Options) :-
  511    partition(pengine_ask_option, Options, AskOptions, SendOptions),
  512    pengine_send(ID, ask(Query, AskOptions), SendOptions).
  513
  514
  515pengine_ask_option(template(_)).
  516pengine_ask_option(chunk(_)).
  517pengine_ask_option(bindings(_)).
  518pengine_ask_option(breakpoints(_)).

 pengine_next(+NameOrID, +Options) is det

Asks pengine NameOrID for the next solution to a query started by pengine_ask/3. Defined options are:

chunk(+Count)

Modify the chunk-size to Count before asking the next set of solutions.

Remaining options are passed to pengine_send/3. The result of re-executing the current goal is returned to the caller's message queue in the form of event terms.

success(ID, Terms, Projection, Time, More)

See pengine_ask/3.

failure(ID)

ID is the id of the pengine that failed for lack of more solutions.

error(ID, Term)

ID is the id of the pengine throwing the exception. Term is the exception's error term.

output(ID, Term)

ID is the id of a pengine running the query that called pengine_output/1. Term is the term that was passed in the first argument of pengine_output/1 when it was called.

prompt(ID, Term)

ID is the id of the pengine that called pengine_input/2 and Term is the prompt.

Defined in terms of pengine_send/3, as follows:

pengine_next(ID, Options) :-
    pengine_send(ID, next, Options).

*/

  562pengine_next(ID, Options) :-
  563    select_option(chunk(Count), Options, Options1),
  564    !,
  565    pengine_send(ID, next(Count), Options1).
  566pengine_next(ID, Options) :-
  567    pengine_send(ID, next, Options).

 pengine_stop(+NameOrID, +Options) is det

Tells pengine NameOrID to stop looking for more solutions to a query started by pengine_ask/3. Options are passed to pengine_send/3.

Defined in terms of pengine_send/3, like so:

pengine_stop(ID, Options) :-
    pengine_send(ID, stop, Options).

*/

  583pengine_stop(ID, Options) :- pengine_send(ID, stop, Options).

 pengine_abort(+NameOrID) is det

Aborts the running query. The pengine goes back to state `2', waiting for new queries.

See also

- pengine_destroy/1. */

  594pengine_abort(Name) :-
  595    (   child(Name, Pengine)
  596    ->  true
  597    ;   Pengine = Name
  598    ),
  599    (   pengine_remote(Pengine, Server)
  600    ->  remote_pengine_abort(Server, Pengine, [])
  601    ;   pengine_thread(Pengine, Thread),
  602        debug(pengine(abort), 'Signalling thread ~p', [Thread]),
  603        catch(thread_signal(Thread, throw(abort_query)), _, true)
  604    ).

 pengine_destroy(+NameOrID) is det

 pengine_destroy(+NameOrID, +Options) is det

Destroys the pengine NameOrID. With the option force(true), the pengine is killed using abort/0 and pengine_destroy/2 succeeds. */

  614pengine_destroy(ID) :-
  615    pengine_destroy(ID, []).
  616
  617pengine_destroy(Name, Options) :-
  618    (   child(Name, ID)
  619    ->  true
  620    ;   ID = Name
  621    ),
  622    option(force(true), Options),
  623    !,
  624    (   pengine_thread(ID, Thread)
  625    ->  catch(thread_signal(Thread, abort),
  626              error(existence_error(thread, _), _), true)
  627    ;   true
  628    ).
  629pengine_destroy(ID, _) :-
  630    catch(pengine_send(ID, destroy),
  631          error(existence_error(pengine, ID), _),
  632          retractall(child(_,ID))).
  633
  634
  635/*================= pengines administration =======================
  636*/

 current_pengine(?Id, ?Parent, ?Location)

Dynamic predicate that registers our known pengines. Id is an atomic unique datatype. Parent is the id of our parent pengine. Location is one of

• thread(ThreadId)
• remote(URL)

  647:- dynamic
  648    current_pengine/6,              % Id, ParentId, Thread, URL, App, Destroy
  649    pengine_queue/4,                % Id, Queue, TimeOut, Time
  650    output_queue/3,                 % Id, Queue, Time
  651    pengine_user/2,                 % Id, User
  652    pengine_data/2,                 % Id, Data
  653    pengine_detached/2.             % Id, Data
  654:- volatile
  655    current_pengine/6,
  656    pengine_queue/4,
  657    output_queue/3,
  658    pengine_user/2,
  659    pengine_data/2,
  660    pengine_detached/2.  661
  662:- thread_local
  663    child/2.                        % ?Name, ?Child

 pengine_register_local(+Id, +Thread, +Queue, +URL, +App, +Destroy) is det

 pengine_register_remote(+Id, +URL, +Queue, +App, +Destroy) is det

 pengine_unregister(+Id) is det

  669pengine_register_local(Id, Thread, Queue, URL, Application, Destroy) :-
  670    asserta(current_pengine(Id, Queue, Thread, URL, Application, Destroy)).
  671
  672pengine_register_remote(Id, URL, Application, Destroy) :-
  673    thread_self(Queue),
  674    asserta(current_pengine(Id, Queue, 0, URL, Application, Destroy)).

 pengine_unregister(+Id)

Called by the pengine thread destruction. If we are a remote pengine thread, our URL equals http and the queue is the message queue used to send events to the HTTP workers.

  682pengine_unregister(Id) :-
  683    thread_self(Me),
  684    (   current_pengine(Id, Queue, Me, http, _, _)
  685    ->  with_mutex(pengine, sync_destroy_queue_from_pengine(Id, Queue))
  686    ;   true
  687    ),
  688    retractall(current_pengine(Id, _, Me, _, _, _)),
  689    retractall(pengine_user(Id, _)),
  690    retractall(pengine_data(Id, _)).
  691
  692pengine_unregister_remote(Id) :-
  693    retractall(current_pengine(Id, _Parent, 0, _, _, _)).

 pengine_self(-Id) is det

True if the current thread is a pengine with Id.

  699pengine_self(Id) :-
  700    thread_self(Thread),
  701    current_pengine(Id, _Parent, Thread, _URL, _Application, _Destroy).
  702
  703pengine_parent(Parent) :-
  704    nb_getval(pengine_parent, Parent).
  705
  706pengine_thread(Pengine, Thread) :-
  707    current_pengine(Pengine, _Parent, Thread, _URL, _Application, _Destroy),
  708    Thread \== 0,
  709    !.
  710
  711pengine_remote(Pengine, URL) :-
  712    current_pengine(Pengine, _Parent, 0, URL, _Application, _Destroy).
  713
  714get_pengine_application(Pengine, Application) :-
  715    current_pengine(Pengine, _Parent, _, _URL, Application, _Destroy),
  716    !.
  717
  718get_pengine_module(Pengine, Pengine).
  719
  720:- if(current_predicate(uuid/2)).  721pengine_uuid(Id) :-
  722    uuid(Id, [version(4)]).             % Version 4 is random.
  723:- else.  724pengine_uuid(Id) :-
  725    (   current_prolog_flag(max_integer, Max1)
  726    ->  Max is Max1-1
  727    ;   Max is 1<<128
  728    ),
  729    random_between(0, Max, Num),
  730    atom_number(Id, Num).
  731:- endif.

 protect_pengine(+Id, :Goal) is semidet

Run Goal while protecting the Pengine Id from being destroyed. Used by the HTTP I/O routines to avoid that the Pengine's module disappears while I/O is in progress. We use a pool of locks because the lock may be held relatively long by output routines.

This also runs Goal if the Pengine no longer exists. This deals with Pengines terminated through destroy_or_continue/1.

bug

- After destroy_or_continue/1 takes the destroy route, the module may drop-out at any point in time, resulting in a possible crash. Seems the only safe way out is to do (de)serialization inside the Pengine.

  748:- meta_predicate protect_pengine(+, 0).  749
  750protect_pengine(Id, Goal) :-
  751    term_hash(Id, Hash),
  752    LockN is Hash mod 64,
  753    atom_concat(pengine_done_, LockN, Lock),
  754    with_mutex(Lock,
  755               (   pengine_thread(Id, _)
  756               ->  Goal
  757               ;   Goal
  758               )).

 pengine_application(+Application) is det

Directive that must be used to declare a pengine application module. The module must not be associated to any file. The default application is pengine_sandbox. The example below creates a new application address_book and imports the API defined in the module file adress_book_api.pl into the application.

:- pengine_application(address_book).
:- use_module(address_book:adress_book_api).

*/

  775pengine_application(Application) :-
  776    throw(error(context_error(nodirective,
  777                             pengine_application(Application)), _)).
  778
  779:- multifile
  780    system:term_expansion/2,
  781    current_application/1.

 current_pengine_application(?Application) is nondet

True when Application is a currently defined application.

See also

- pengine_application/1

  789current_pengine_application(Application) :-
  790    current_application(Application).
  791
  792
  793% Default settings for all applications
  794
  795:- setting(thread_pool_size, integer, 100,
  796           'Maximum number of pengines this application can run.').  797:- setting(thread_pool_stacks, list(compound), [],
  798           'Maximum stack sizes for pengines this application can run.').  799:- setting(slave_limit, integer, 3,
  800           'Maximum number of slave pengines a master pengine can create.').  801:- setting(time_limit, number, 300,
  802           'Maximum time to wait for output').  803:- setting(idle_limit, number, 300,
  804           'Pengine auto-destroys when idle for this time').  805:- setting(safe_goal_limit, number, 10,
  806           'Maximum time to try proving safety of the goal').  807:- setting(program_space, integer, 100_000_000,
  808           'Maximum memory used by predicates').  809:- setting(allow_from, list(atom), [*],
  810           'IP addresses from which remotes are allowed to connect').  811:- setting(deny_from, list(atom), [],
  812           'IP addresses from which remotes are NOT allowed to connect').  813:- setting(debug_info, boolean, false,
  814           'Keep information to support source-level debugging').  815
  816
  817system:term_expansion((:- pengine_application(Application)), Expanded) :-
  818    must_be(atom, Application),
  819    (   module_property(Application, file(_))
  820    ->  permission_error(create, pengine_application, Application)
  821    ;   true
  822    ),
  823    expand_term((:- setting(Application:thread_pool_size, integer,
  824                            setting(pengines:thread_pool_size),
  825                            'Maximum number of pengines this \c
  826                            application can run.')),
  827                ThreadPoolSizeSetting),
  828    expand_term((:- setting(Application:thread_pool_stacks, list(compound),
  829                            setting(pengines:thread_pool_stacks),
  830                            'Maximum stack sizes for pengines \c
  831                            this application can run.')),
  832                ThreadPoolStacksSetting),
  833    expand_term((:- setting(Application:slave_limit, integer,
  834                            setting(pengines:slave_limit),
  835                            'Maximum number of local slave pengines \c
  836                            a master pengine can create.')),
  837                SlaveLimitSetting),
  838    expand_term((:- setting(Application:time_limit, number,
  839                            setting(pengines:time_limit),
  840                            'Maximum time to wait for output')),
  841                TimeLimitSetting),
  842    expand_term((:- setting(Application:idle_limit, number,
  843                            setting(pengines:idle_limit),
  844                            'Pengine auto-destroys when idle for this time')),
  845                IdleLimitSetting),
  846    expand_term((:- setting(Application:safe_goal_limit, number,
  847                            setting(pengines:safe_goal_limit),
  848                            'Maximum time to try proving safety of the goal')),
  849                SafeGoalLimitSetting),
  850    expand_term((:- setting(Application:program_space, integer,
  851                            setting(pengines:program_space),
  852                            'Maximum memory used by predicates')),
  853                ProgramSpaceSetting),
  854    expand_term((:- setting(Application:allow_from, list(atom),
  855                            setting(pengines:allow_from),
  856                            'IP addresses from which remotes are allowed \c
  857                            to connect')),
  858                AllowFromSetting),
  859    expand_term((:- setting(Application:deny_from, list(atom),
  860                            setting(pengines:deny_from),
  861                            'IP addresses from which remotes are NOT \c
  862                            allowed to connect')),
  863                DenyFromSetting),
  864    expand_term((:- setting(Application:debug_info, boolean,
  865                            setting(pengines:debug_info),
  866                            'Keep information to support source-level \c
  867                            debugging')),
  868                DebugInfoSetting),
  869    flatten([ pengines:current_application(Application),
  870              ThreadPoolSizeSetting,
  871              ThreadPoolStacksSetting,
  872              SlaveLimitSetting,
  873              TimeLimitSetting,
  874              IdleLimitSetting,
  875              SafeGoalLimitSetting,
  876              ProgramSpaceSetting,
  877              AllowFromSetting,
  878              DenyFromSetting,
  879              DebugInfoSetting
  880            ], Expanded).
  881
  882% Register default application
  883
  884:- pengine_application(pengine_sandbox).

 pengine_property(?Pengine, ?Property) is nondet

True when Property is a property of the given Pengine. Enumerates all pengines that are known to the calling Prolog process. Defined properties are:

self(ID)

Identifier of the pengine. This is the same as the first argument, and can be used to enumerate all known pengines.

alias(Name)

Name is the alias name of the pengine, as provided through the alias option when creating the pengine.

thread(Thread)

If the pengine is a local pengine, Thread is the Prolog thread identifier of the pengine.

remote(Server)

If the pengine is remote, the URL of the server.

application(Application)

Pengine runs the given application

module(Module)

Temporary module used for running the Pengine.

destroy(Destroy)

Destroy is true if the pengines is destroyed automatically after completing the query.

parent(Queue)

Message queue to which the (local) pengine reports.

source(?SourceID, ?Source)

Source is the source code with the given SourceID. May be present if the setting debug_info is present.

detached(?Time)

Pengine was detached at Time. */

  921pengine_property(Id, Prop) :-
  922    nonvar(Id), nonvar(Prop),
  923    pengine_property2(Prop, Id),
  924    !.
  925pengine_property(Id, Prop) :-
  926    pengine_property2(Prop, Id).
  927
  928pengine_property2(self(Id), Id) :-
  929    current_pengine(Id, _Parent, _Thread, _URL, _Application, _Destroy).
  930pengine_property2(module(Id), Id) :-
  931    current_pengine(Id, _Parent, _Thread, _URL, _Application, _Destroy).
  932pengine_property2(alias(Alias), Id) :-
  933    child(Alias, Id),
  934    Alias \== Id.
  935pengine_property2(thread(Thread), Id) :-
  936    current_pengine(Id, _Parent, Thread, _URL, _Application, _Destroy),
  937    Thread \== 0.
  938pengine_property2(remote(Server), Id) :-
  939    current_pengine(Id, _Parent, 0, Server, _Application, _Destroy).
  940pengine_property2(application(Application), Id) :-
  941    current_pengine(Id, _Parent, _Thread, _Server, Application, _Destroy).
  942pengine_property2(destroy(Destroy), Id) :-
  943    current_pengine(Id, _Parent, _Thread, _Server, _Application, Destroy).
  944pengine_property2(parent(Parent), Id) :-
  945    current_pengine(Id, Parent, _Thread, _URL, _Application, _Destroy).
  946pengine_property2(source(SourceID, Source), Id) :-
  947    pengine_data(Id, source(SourceID, Source)).
  948pengine_property2(detached(When), Id) :-
  949    pengine_detached(Id, When).

 pengine_output(+Term) is det

Sends Term to the parent pengine or thread. */

  956pengine_output(Term) :-
  957    pengine_self(Me),
  958    pengine_reply(output(Me, Term)).

 pengine_debug(+Format, +Args) is det

Create a message using format/3 from Format and Args and send this to the client. The default JavaScript client will call console.log(Message) if there is a console. The predicate pengine_rpc/3 calls debug(pengine(debug), '~w', [Message]). The debug topic pengine(debug) is enabled by default.

See also

- debug/1 and nodebug/1 for controlling the pengine(debug) topic

- format/2 for format specifications */

  973pengine_debug(Format, Args) :-
  974    pengine_parent(Queue),
  975    pengine_self(Self),
  976    catch(safe_goal(format(atom(_), Format, Args)), E, true),
  977    (   var(E)
  978    ->  format(atom(Message), Format, Args)
  979    ;   message_to_string(E, Message)
  980    ),
  981    pengine_reply(Queue, debug(Self, Message)).
  982
  983
  984/*================= Local pengine =======================
  985*/

 local_pengine_create(+Options)

Creates a local Pengine, which is a thread running pengine_main/2. It maintains two predicates:

• The global dynamic predicate id/2 relates Pengines to their childs.
• The local predicate id/2 maps named childs to their ids.

  996local_pengine_create(Options) :-
  997    thread_self(Self),
  998    option(application(Application), Options, pengine_sandbox),
  999    create(Self, Child, Options, local, Application),
 1000    option(alias(Name), Options, Child),
 1001    assert(child(Name, Child)).

 thread_pool:create_pool(+Application) is det

On demand creation of a thread pool for a pengine application.

 1008:- multifile thread_pool:create_pool/1. 1009
 1010thread_pool:create_pool(Application) :-
 1011    current_application(Application),
 1012    setting(Application:thread_pool_size, Size),
 1013    setting(Application:thread_pool_stacks, Stacks),
 1014    thread_pool_create(Application, Size, Stacks).

 create(+Queue, -Child, +Options, +URL, +Application) is det

Create a new pengine thread.

Arguments:

Queue	- is the queue (or thread handle) to report to
Child	- is the identifier of the created pengine.
URL	- is one of local or http

 1024create(Queue, Child, Options, local, Application) :-
 1025    !,
 1026    pengine_child_id(Child),
 1027    create0(Queue, Child, Options, local, Application).
 1028create(Queue, Child, Options, URL, Application) :-
 1029    pengine_child_id(Child),
 1030    catch(create0(Queue, Child, Options, URL, Application),
 1031          Error,
 1032          create_error(Queue, Child, Error)).
 1033
 1034pengine_child_id(Child) :-
 1035    (   nonvar(Child)
 1036    ->  true
 1037    ;   pengine_uuid(Child)
 1038    ).
 1039
 1040create_error(Queue, Child, Error) :-
 1041    pengine_reply(Queue, error(Child, Error)).
 1042
 1043create0(Queue, Child, Options, URL, Application) :-
 1044    (  current_application(Application)
 1045    -> true
 1046    ;  existence_error(pengine_application, Application)
 1047    ),
 1048    (   URL \== http                    % pengine is _not_ a child of the
 1049                                        % HTTP server thread
 1050    ->  aggregate_all(count, child(_,_), Count),
 1051        setting(Application:slave_limit, Max),
 1052        (   Count >= Max
 1053        ->  throw(error(resource_error(max_pengines), _))
 1054        ;   true
 1055        )
 1056    ;   true
 1057    ),
 1058    partition(pengine_create_option, Options, PengineOptions, RestOptions),
 1059    thread_create_in_pool(
 1060        Application,
 1061        pengine_main(Queue, PengineOptions, Application), ChildThread,
 1062        [ at_exit(pengine_done)
 1063        | RestOptions
 1064        ]),
 1065    option(destroy(Destroy), PengineOptions, true),
 1066    pengine_register_local(Child, ChildThread, Queue, URL, Application, Destroy),
 1067    thread_send_message(ChildThread, pengine_registered(Child)),
 1068    (   option(id(Id), Options)
 1069    ->  Id = Child
 1070    ;   true
 1071    ).
 1072
 1073pengine_create_option(src_text(_)).
 1074pengine_create_option(src_url(_)).
 1075pengine_create_option(application(_)).
 1076pengine_create_option(destroy(_)).
 1077pengine_create_option(ask(_)).
 1078pengine_create_option(template(_)).
 1079pengine_create_option(bindings(_)).
 1080pengine_create_option(chunk(_)).
 1081pengine_create_option(alias(_)).
 1082pengine_create_option(user(_)).

 pengine_done is det

Called from the pengine thread at_exit option. Destroys child pengines using pengine_destroy/1. Cleaning up the Pengine is synchronised by the pengine_done mutex. See read_event/6.

 1091:- public
 1092    pengine_done/0. 1093
 1094pengine_done :-
 1095    thread_self(Me),
 1096    (   thread_property(Me, status(exception('$aborted'))),
 1097        thread_detach(Me),
 1098        pengine_self(Pengine)
 1099    ->  catch(pengine_reply(destroy(Pengine, abort(Pengine))),
 1100              error(_,_), true)
 1101    ;   true
 1102    ),
 1103    forall(child(_Name, Child),
 1104           pengine_destroy(Child)),
 1105    pengine_self(Id),
 1106    protect_pengine(Id, pengine_unregister(Id)).

 pengine_main(+Parent, +Options, +Application)

Run a pengine main loop. First acknowledges its creation and run pengine_main_loop/1.

 1114:- thread_local wrap_first_answer_in_create_event/2. 1115
 1116:- meta_predicate
 1117    pengine_prepare_source(:, +). 1118
 1119pengine_main(Parent, Options, Application) :-
 1120    fix_streams,
 1121    thread_get_message(pengine_registered(Self)),
 1122    nb_setval(pengine_parent, Parent),
 1123    pengine_register_user(Options),
 1124    set_prolog_flag(mitigate_spectre, true),
 1125    catch(in_temporary_module(
 1126              Self,
 1127              pengine_prepare_source(Application, Options),
 1128              pengine_create_and_loop(Self, Application, Options)),
 1129          prepare_source_failed,
 1130          pengine_terminate(Self)).
 1131
 1132pengine_create_and_loop(Self, Application, Options) :-
 1133    setting(Application:slave_limit, SlaveLimit),
 1134    CreateEvent = create(Self, [slave_limit(SlaveLimit)|Extra]),
 1135    (   option(ask(Query0), Options)
 1136    ->  asserta(wrap_first_answer_in_create_event(CreateEvent, Extra)),
 1137        (   string(Query0)                      % string is not callable
 1138        ->  (   option(template(TemplateS), Options)
 1139            ->  Ask2 = Query0-TemplateS
 1140            ;   Ask2 = Query0
 1141            ),
 1142            catch(ask_to_term(Ask2, Self, Query, Template, Bindings),
 1143                  Error, true),
 1144            (   var(Error)
 1145            ->  true
 1146            ;   send_error(Error),
 1147                throw(prepare_source_failed)
 1148            )
 1149        ;   Query = Query0,
 1150            option(template(Template), Options, Query),
 1151            option(bindings(Bindings), Options, [])
 1152        ),
 1153        option(chunk(Chunk), Options, 1),
 1154        pengine_ask(Self, Query,
 1155                    [ template(Template),
 1156                      chunk(Chunk),
 1157                      bindings(Bindings)
 1158                    ])
 1159    ;   Extra = [],
 1160        pengine_reply(CreateEvent)
 1161    ),
 1162    pengine_main_loop(Self).

 ask_to_term(+AskSpec, +Module, -Options, OptionsTail) is det

Translate the AskSpec into a query, template and bindings. The trick is that we must parse using the operator declarations of the source and we must make sure variable sharing between query and answer template are known.

 1172ask_to_term(Ask-Template, Module, Ask1, Template1, Bindings) :-
 1173    !,
 1174    format(string(AskTemplate), 't((~s),(~s))', [Template, Ask]),
 1175    term_string(t(Template1,Ask1), AskTemplate,
 1176                [ variable_names(Bindings0),
 1177                  module(Module)
 1178                ]),
 1179    phrase(template_bindings(Template1, Bindings0), Bindings).
 1180ask_to_term(Ask, Module, Ask1, Template, Bindings1) :-
 1181    term_string(Ask1, Ask,
 1182                [ variable_names(Bindings),
 1183                  module(Module)
 1184                ]),
 1185    exclude(anon, Bindings, Bindings1),
 1186    dict_create(Template, swish_default_template, Bindings1).
 1187
 1188template_bindings(Var, Bindings) -->
 1189    { var(Var) }, !,
 1190    (   { var_binding(Bindings, Var, Binding)
 1191        }
 1192    ->  [Binding]
 1193    ;   []
 1194    ).
 1195template_bindings([H|T], Bindings) -->
 1196    !,
 1197    template_bindings(H, Bindings),
 1198    template_bindings(T, Bindings).
 1199template_bindings(Compoound, Bindings) -->
 1200    { compound(Compoound), !,
 1201      compound_name_arguments(Compoound, _, Args)
 1202    },
 1203    template_bindings(Args, Bindings).
 1204template_bindings(_, _) --> [].
 1205
 1206var_binding(Bindings, Var, Binding) :-
 1207    member(Binding, Bindings),
 1208    arg(2, Binding, V),
 1209    V == Var, !.

 fix_streams is det

If we are a pengine that is created from a web server thread, the current output points to a CGI stream.

 1216fix_streams :-
 1217    fix_stream(current_output).
 1218
 1219fix_stream(Name) :-
 1220    is_cgi_stream(Name),
 1221    !,
 1222    debug(pengine(stream), '~w is a CGI stream!', [Name]),
 1223    set_stream(user_output, alias(Name)).
 1224fix_stream(_).

 pengine_prepare_source(:Application, +Options) is det

Load the source into the pengine's module.

throws

- prepare_source_failed if it failed to prepare the sources.

 1233pengine_prepare_source(Module:Application, Options) :-
 1234    setting(Application:program_space, SpaceLimit),
 1235    set_module(Module:program_space(SpaceLimit)),
 1236    delete_import_module(Module, user),
 1237    add_import_module(Module, Application, start),
 1238    catch(prep_module(Module, Application, Options), Error, true),
 1239    (   var(Error)
 1240    ->  true
 1241    ;   send_error(Error),
 1242        throw(prepare_source_failed)
 1243    ).
 1244
 1245prep_module(Module, Application, Options) :-
 1246    maplist(copy_flag(Module, Application), [var_prefix]),
 1247    forall(prepare_module(Module, Application, Options), true),
 1248    setup_call_cleanup(
 1249        '$set_source_module'(OldModule, Module),
 1250        maplist(process_create_option(Module), Options),
 1251        '$set_source_module'(OldModule)).
 1252
 1253copy_flag(Module, Application, Flag) :-
 1254    current_prolog_flag(Application:Flag, Value),
 1255    !,
 1256    set_prolog_flag(Module:Flag, Value).
 1257copy_flag(_, _, _).
 1258
 1259process_create_option(Application, src_text(Text)) :-
 1260    !,
 1261    pengine_src_text(Text, Application).
 1262process_create_option(Application, src_url(URL)) :-
 1263    !,
 1264    pengine_src_url(URL, Application).
 1265process_create_option(_, _).

 prepare_module(+Module, +Application, +Options) is semidet

Hook, called to initialize the temporary private module that provides the working context of a pengine. This hook is executed by the pengine's thread. Preparing the source consists of three steps:

1. Add Application as (first) default import module for Module
2. Call this hook
3. Compile the source provided by the the src_text and src_url options

Arguments:

Module	- is a new temporary module (see in_temporary_module/3) that may be (further) prepared by this hook.
Application	- (also a module) associated to the pengine.
Options	- is passed from the environment and should (currently) be ignored.

 1288pengine_main_loop(ID) :-
 1289    catch(guarded_main_loop(ID), abort_query, pengine_aborted(ID)).
 1290
 1291pengine_aborted(ID) :-
 1292    thread_self(Self),
 1293    debug(pengine(abort), 'Aborting ~p (thread ~p)', [ID, Self]),
 1294    empty_queue,
 1295    destroy_or_continue(abort(ID)).

 guarded_main_loop(+Pengine) is det

Executes state `2' of the pengine, where it waits for two events:

destroy

Terminate the pengine

ask(:Goal, +Options)

Solve Goal.

 1308guarded_main_loop(ID) :-
 1309    pengine_request(Request),
 1310    (   Request = destroy
 1311    ->  debug(pengine(transition), '~q: 2 = ~q => 1', [ID, destroy]),
 1312        pengine_terminate(ID)
 1313    ;   Request = ask(Goal, Options)
 1314    ->  debug(pengine(transition), '~q: 2 = ~q => 3', [ID, ask(Goal)]),
 1315        ask(ID, Goal, Options)
 1316    ;   debug(pengine(transition), '~q: 2 = ~q => 2', [ID, protocol_error]),
 1317        pengine_reply(error(ID, error(protocol_error, _))),
 1318        guarded_main_loop(ID)
 1319    ).
 1320
 1321
 1322pengine_terminate(ID) :-
 1323    pengine_reply(destroy(ID)),
 1324    thread_self(Me),            % Make the thread silently disappear
 1325    thread_detach(Me).

 solve(+Chunk, +Template, :Goal, +ID) is det

Solve Goal. Note that because we can ask for a new goal in state `6', we must provide for an ancesteral cut (prolog_cut_to/1). We need to be sure to have a choice point before we can call prolog_current_choice/1. This is the reason why this predicate has two clauses.

 1336solve(Chunk, Template, Goal, ID) :-
 1337    prolog_current_choice(Choice),
 1338    State = count(Chunk),
 1339    statistics(cputime, Epoch),
 1340    Time = time(Epoch),
 1341    nb_current('$variable_names', Bindings),
 1342    filter_template(Template, Bindings, Template2),
 1343    '$current_typein_module'(CurrTypeIn),
 1344    (   '$set_typein_module'(ID),
 1345        call_cleanup(catch(findnsols_no_empty(State, Template2,
 1346                                              set_projection(Goal, Bindings),
 1347                                              Result),
 1348                           Error, true),
 1349                     query_done(Det, CurrTypeIn)),
 1350        arg(1, Time, T0),
 1351        statistics(cputime, T1),
 1352        CPUTime is T1-T0,
 1353        (   var(Error)
 1354        ->  projection(Projection),
 1355            (   var(Det)
 1356            ->  pengine_reply(success(ID, Result, Projection,
 1357                                      CPUTime, true)),
 1358                more_solutions(ID, Choice, State, Time)
 1359            ;   !,                      % commit
 1360                destroy_or_continue(success(ID, Result, Projection,
 1361                                            CPUTime, false))
 1362            )
 1363        ;   !,                          % commit
 1364            (   Error == abort_query
 1365            ->  throw(Error)
 1366            ;   destroy_or_continue(error(ID, Error))
 1367            )
 1368        )
 1369    ;   !,                              % commit
 1370        arg(1, Time, T0),
 1371        statistics(cputime, T1),
 1372        CPUTime is T1-T0,
 1373        destroy_or_continue(failure(ID, CPUTime))
 1374    ).
 1375solve(_, _, _, _).                      % leave a choice point
 1376
 1377query_done(true, CurrTypeIn) :-
 1378    '$set_typein_module'(CurrTypeIn).

 set_projection(:Goal, +Bindings)

findnsols/4 copies its goal and template to avoid instantiation thereof when it stops after finding N solutions. Using this helper we can a renamed version of Bindings that we can set.

 1387set_projection(Goal, Bindings) :-
 1388    b_setval('$variable_names', Bindings),
 1389    call(Goal).
 1390
 1391projection(Projection) :-
 1392    nb_current('$variable_names', Bindings),
 1393    !,
 1394    maplist(var_name, Bindings, Projection).
 1395projection([]).

 filter_template(+Template0, +Bindings, -Template) is det

Establish the final template. This is there because hooks such as goal_expansion/2 and the SWISH query hooks can modify the set of bindings.

bug

- Projection and template handling is pretty messy.

 1405filter_template(Template0, Bindings, Template) :-
 1406    is_dict(Template0, swish_default_template),
 1407    !,
 1408    dict_create(Template, swish_default_template, Bindings).
 1409filter_template(Template, _Bindings, Template).
 1410
 1411findnsols_no_empty(N, Template, Goal, List) :-
 1412    findnsols(N, Template, Goal, List),
 1413    List \== [].
 1414
 1415destroy_or_continue(Event) :-
 1416    arg(1, Event, ID),
 1417    (   pengine_property(ID, destroy(true))
 1418    ->  thread_self(Me),
 1419        thread_detach(Me),
 1420        pengine_reply(destroy(ID, Event))
 1421    ;   pengine_reply(Event),
 1422        guarded_main_loop(ID)
 1423    ).

 more_solutions(+Pengine, +Choice, +State, +Time)

Called after a solution was found while there can be more. This is state `6' of the state machine. It processes these events:

stop

Go back via state `7' to state `2' (guarded_main_loop/1)

next

Fail. This causes solve/3 to backtrack on the goal asked, providing at most the current chunk solutions.

next(Count)

As next, but sets the new chunk-size to Count.

ask(Goal, Options)

Ask another goal. Note that we must commit the choice point of the previous goal asked for.

 1441more_solutions(ID, Choice, State, Time) :-
 1442    pengine_request(Event),
 1443    more_solutions(Event, ID, Choice, State, Time).
 1444
 1445more_solutions(stop, ID, _Choice, _State, _Time) :-
 1446    !,
 1447    debug(pengine(transition), '~q: 6 = ~q => 7', [ID, stop]),
 1448    destroy_or_continue(stop(ID)).
 1449more_solutions(next, ID, _Choice, _State, Time) :-
 1450    !,
 1451    debug(pengine(transition), '~q: 6 = ~q => 3', [ID, next]),
 1452    statistics(cputime, T0),
 1453    nb_setarg(1, Time, T0),
 1454    fail.
 1455more_solutions(next(Count), ID, _Choice, State, Time) :-
 1456    Count > 0,
 1457    !,
 1458    debug(pengine(transition), '~q: 6 = ~q => 3', [ID, next(Count)]),
 1459    nb_setarg(1, State, Count),
 1460    statistics(cputime, T0),
 1461    nb_setarg(1, Time, T0),
 1462    fail.
 1463more_solutions(ask(Goal, Options), ID, Choice, _State, _Time) :-
 1464    !,
 1465    debug(pengine(transition), '~q: 6 = ~q => 3', [ID, ask(Goal)]),
 1466    prolog_cut_to(Choice),
 1467    ask(ID, Goal, Options).
 1468more_solutions(destroy, ID, _Choice, _State, _Time) :-
 1469    !,
 1470    debug(pengine(transition), '~q: 6 = ~q => 1', [ID, destroy]),
 1471    pengine_terminate(ID).
 1472more_solutions(Event, ID, Choice, State, Time) :-
 1473    debug(pengine(transition), '~q: 6 = ~q => 6', [ID, protocol_error(Event)]),
 1474    pengine_reply(error(ID, error(protocol_error, _))),
 1475    more_solutions(ID, Choice, State, Time).

 ask(+Pengine, :Goal, +Options)

Migrate from state `2' to `3'. This predicate validates that it is safe to call Goal using safe_goal/1 and then calls solve/3 to prove the goal. It takes care of the chunk(N) option.

 1483ask(ID, Goal, Options) :-
 1484    catch(prepare_goal(ID, Goal, Goal1, Options), Error, true),
 1485    !,
 1486    (   var(Error)
 1487    ->  option(template(Template), Options, Goal),
 1488        option(chunk(N), Options, 1),
 1489        solve(N, Template, Goal1, ID)
 1490    ;   pengine_reply(error(ID, Error)),
 1491        guarded_main_loop(ID)
 1492    ).

 prepare_goal(+Pengine, +GoalIn, -GoalOut, +Options) is det

Prepare GoalIn for execution in Pengine. This implies we must perform goal expansion and, if the system is sandboxed, check the sandbox.

Note that expand_goal(Module:GoalIn, GoalOut) is what we'd like to write, but this does not work correctly if the user wishes to expand X:Y while interpreting X not as the module in which to run Y. This happens in the CQL package. Possibly we should disallow this reinterpretation?

 1506prepare_goal(ID, Goal0, Module:Goal, Options) :-
 1507    option(bindings(Bindings), Options, []),
 1508    b_setval('$variable_names', Bindings),
 1509    (   prepare_goal(Goal0, Goal1, Options)
 1510    ->  true
 1511    ;   Goal1 = Goal0
 1512    ),
 1513    get_pengine_module(ID, Module),
 1514    setup_call_cleanup(
 1515        '$set_source_module'(Old, Module),
 1516        expand_goal(Goal1, Goal),
 1517        '$set_source_module'(_, Old)),
 1518    (   pengine_not_sandboxed(ID)
 1519    ->  true
 1520    ;   get_pengine_application(ID, App),
 1521        setting(App:safe_goal_limit, Limit),
 1522        catch(call_with_time_limit(
 1523                  Limit,
 1524                  safe_goal(Module:Goal)), E, true)
 1525    ->  (   var(E)
 1526        ->  true
 1527        ;   E = time_limit_exceeded
 1528        ->  throw(error(sandbox(time_limit_exceeded, Limit),_))
 1529        ;   throw(E)
 1530        )
 1531    ).

 prepare_goal(+Goal0, -Goal1, +Options) is semidet

Pre-preparation hook for running Goal0. The hook runs in the context of the pengine. Goal is the raw goal given to ask. The returned Goal1 is subject to goal expansion (expand_goal/2) and sandbox validation (safe_goal/1) prior to execution. If this goal fails, Goal0 is used for further processing.

Arguments:

Options

- provides the options as given to ask

 pengine_not_sandboxed(+Pengine) is semidet

True when pengine does not operate in sandboxed mode. This implies a user must be registered by authentication_hook/3 and the hook pengines:not_sandboxed(User, Application) must succeed.

 1551pengine_not_sandboxed(ID) :-
 1552    pengine_user(ID, User),
 1553    pengine_property(ID, application(App)),
 1554    not_sandboxed(User, App),
 1555    !.

 not_sandboxed(+User, +Application) is semidet

This hook is called to see whether the Pengine must be executed in a protected environment. It is only called after authentication_hook/3 has confirmed the authentity of the current user. If this hook succeeds, both loading the code and executing the query is executed without enforcing sandbox security. Typically, one should:

1. Provide a safe user authentication hook.
2. Enable HTTPS in the server or put it behind an HTTPS proxy and ensure that the network between the proxy and the pengine server can be trusted.

 pengine_pull_response(+Pengine, +Options) is det

Pulls a response (an event term) from the slave Pengine if Pengine is a remote process, else does nothing at all. */

 1577pengine_pull_response(Pengine, Options) :-
 1578    pengine_remote(Pengine, Server),
 1579    !,
 1580    remote_pengine_pull_response(Server, Pengine, Options).
 1581pengine_pull_response(_ID, _Options).

 pengine_input(+Prompt, -Term) is det

Sends Prompt to the master (parent) pengine and waits for input. Note that Prompt may be any term, compound as well as atomic. */

 1590pengine_input(Prompt, Term) :-
 1591    pengine_self(Self),
 1592    pengine_parent(Parent),
 1593    pengine_reply(Parent, prompt(Self, Prompt)),
 1594    pengine_request(Request),
 1595    (   Request = input(Input)
 1596    ->  Term = Input
 1597    ;   Request == destroy
 1598    ->  abort
 1599    ;   throw(error(protocol_error,_))
 1600    ).

 pengine_respond(+Pengine, +Input, +Options) is det

Sends a response in the form of the term Input to a slave (child) pengine that has prompted its master (parent) for input.

Defined in terms of pengine_send/3, as follows:

pengine_respond(Pengine, Input, Options) :-
    pengine_send(Pengine, input(Input), Options).

*/

 1617pengine_respond(Pengine, Input, Options) :-
 1618    pengine_send(Pengine, input(Input), Options).

 send_error(+Error) is det

Send an error to my parent. Remove non-readable blobs from the error term first using replace_blobs/2. If the error contains a stack-trace, this is resolved to a string before sending.

 1627send_error(error(Formal, context(prolog_stack(Frames), Message))) :-
 1628    is_list(Frames),
 1629    !,
 1630    with_output_to(string(Stack),
 1631                   print_prolog_backtrace(current_output, Frames)),
 1632    pengine_self(Self),
 1633    replace_blobs(Formal, Formal1),
 1634    replace_blobs(Message, Message1),
 1635    pengine_reply(error(Self, error(Formal1,
 1636                                    context(prolog_stack(Stack), Message1)))).
 1637send_error(Error) :-
 1638    pengine_self(Self),
 1639    replace_blobs(Error, Error1),
 1640    pengine_reply(error(Self, Error1)).

 replace_blobs(Term0, Term) is det

Copy Term0 to Term, replacing non-text blobs. This is required for error messages that may hold streams and other handles to non-readable objects.

 1648replace_blobs(Blob, Atom) :-
 1649    blob(Blob, Type), Type \== text,
 1650    !,
 1651    format(atom(Atom), '~p', [Blob]).
 1652replace_blobs(Term0, Term) :-
 1653    compound(Term0),
 1654    !,
 1655    compound_name_arguments(Term0, Name, Args0),
 1656    maplist(replace_blobs, Args0, Args),
 1657    compound_name_arguments(Term, Name, Args).
 1658replace_blobs(Term, Term).
 1659
 1660
 1661/*================= Remote pengines =======================
 1662*/
 1663
 1664
 1665remote_pengine_create(BaseURL, Options) :-
 1666    partition(pengine_create_option, Options, PengineOptions0, RestOptions),
 1667        (       option(ask(Query), PengineOptions0),
 1668                \+ option(template(_Template), PengineOptions0)
 1669        ->      PengineOptions = [template(Query)|PengineOptions0]
 1670        ;       PengineOptions = PengineOptions0
 1671        ),
 1672    options_to_dict(PengineOptions, PostData),
 1673    remote_post_rec(BaseURL, create, PostData, Reply, RestOptions),
 1674    arg(1, Reply, ID),
 1675    (   option(id(ID2), Options)
 1676    ->  ID = ID2
 1677    ;   true
 1678    ),
 1679    option(alias(Name), Options, ID),
 1680    assert(child(Name, ID)),
 1681    (   (   functor(Reply, create, _)   % actually created
 1682        ;   functor(Reply, output, _)   % compiler messages
 1683        )
 1684    ->  option(application(Application), PengineOptions, pengine_sandbox),
 1685        option(destroy(Destroy), PengineOptions, true),
 1686        pengine_register_remote(ID, BaseURL, Application, Destroy)
 1687    ;   true
 1688    ),
 1689    thread_self(Queue),
 1690    pengine_reply(Queue, Reply).
 1691
 1692options_to_dict(Options, Dict) :-
 1693    select_option(ask(Ask), Options, Options1),
 1694    select_option(template(Template), Options1, Options2),
 1695    !,
 1696    no_numbered_var_in(Ask+Template),
 1697    findall(AskString-TemplateString,
 1698            ask_template_to_strings(Ask, Template, AskString, TemplateString),
 1699            [ AskString-TemplateString ]),
 1700    options_to_dict(Options2, Dict0),
 1701    Dict = Dict0.put(_{ask:AskString,template:TemplateString}).
 1702options_to_dict(Options, Dict) :-
 1703    maplist(prolog_option, Options, Options1),
 1704    dict_create(Dict, _, Options1).
 1705
 1706no_numbered_var_in(Term) :-
 1707    sub_term(Sub, Term),
 1708    subsumes_term('$VAR'(_), Sub),
 1709    !,
 1710    domain_error(numbered_vars_free_term, Term).
 1711no_numbered_var_in(_).
 1712
 1713ask_template_to_strings(Ask, Template, AskString, TemplateString) :-
 1714    numbervars(Ask+Template, 0, _),
 1715    WOpts = [ numbervars(true), ignore_ops(true), quoted(true) ],
 1716    format(string(AskTemplate), '~W\n~W', [ Ask, WOpts,
 1717                                            Template, WOpts
 1718                                          ]),
 1719    split_string(AskTemplate, "\n", "", [AskString, TemplateString]).
 1720
 1721prolog_option(Option0, Option) :-
 1722    create_option_type(Option0, term),
 1723    !,
 1724    Option0 =.. [Name,Value],
 1725    format(string(String), '~k', [Value]),
 1726    Option =.. [Name,String].
 1727prolog_option(Option, Option).
 1728
 1729create_option_type(ask(_),         term).
 1730create_option_type(template(_),    term).
 1731create_option_type(application(_), atom).
 1732
 1733remote_pengine_send(BaseURL, ID, Event, Options) :-
 1734    remote_send_rec(BaseURL, send, ID, [event=Event], Reply, Options),
 1735    thread_self(Queue),
 1736    pengine_reply(Queue, Reply).
 1737
 1738remote_pengine_pull_response(BaseURL, ID, Options) :-
 1739    remote_send_rec(BaseURL, pull_response, ID, [], Reply, Options),
 1740    thread_self(Queue),
 1741    pengine_reply(Queue, Reply).
 1742
 1743remote_pengine_abort(BaseURL, ID, Options) :-
 1744    remote_send_rec(BaseURL, abort, ID, [], Reply, Options),
 1745    thread_self(Queue),
 1746    pengine_reply(Queue, Reply).

 remote_send_rec(+Server, +Action, +ID, +Params, -Reply, +Options)

Issue a GET request on Server and unify Reply with the replied term.

 1753remote_send_rec(Server, Action, ID, [event=Event], Reply, Options) :-
 1754    !,
 1755    server_url(Server, Action, [id=ID], URL),
 1756    http_open(URL, Stream,              % putting this in setup_call_cleanup/3
 1757              [ post(prolog(Event))     % makes it impossible to interrupt.
 1758              | Options
 1759              ]),
 1760    call_cleanup(
 1761        read_prolog_reply(Stream, Reply),
 1762        close(Stream)).
 1763remote_send_rec(Server, Action, ID, Params, Reply, Options) :-
 1764    server_url(Server, Action, [id=ID|Params], URL),
 1765    http_open(URL, Stream, Options),
 1766    call_cleanup(
 1767        read_prolog_reply(Stream, Reply),
 1768        close(Stream)).
 1769
 1770remote_post_rec(Server, Action, Data, Reply, Options) :-
 1771    server_url(Server, Action, [], URL),
 1772    probe(Action, URL),
 1773    http_open(URL, Stream,
 1774              [ post(json(Data))
 1775              | Options
 1776              ]),
 1777    call_cleanup(
 1778        read_prolog_reply(Stream, Reply),
 1779        close(Stream)).

 probe(+Action, +URL) is det

Probe the target. This is a good idea before posting a large document and be faced with an authentication challenge. Possibly we should make this an option for simpler scenarios.

 1787probe(create, URL) :-
 1788    !,
 1789    http_open(URL, Stream, [method(options)]),
 1790    close(Stream).
 1791probe(_, _).
 1792
 1793read_prolog_reply(In, Reply) :-
 1794    set_stream(In, encoding(utf8)),
 1795    read(In, Reply0),
 1796    rebind_cycles(Reply0, Reply).
 1797
 1798rebind_cycles(@(Reply, Bindings), Reply) :-
 1799    is_list(Bindings),
 1800    !,
 1801    maplist(bind, Bindings).
 1802rebind_cycles(Reply, Reply).
 1803
 1804bind(Var = Value) :-
 1805    Var = Value.
 1806
 1807server_url(Server, Action, Params, URL) :-
 1808    uri_components(Server, Components0),
 1809    uri_query_components(Query, Params),
 1810    uri_data(path, Components0, Path0),
 1811    atom_concat('pengine/', Action, PAction),
 1812    directory_file_path(Path0, PAction, Path),
 1813    uri_data(path, Components0, Path, Components),
 1814    uri_data(search, Components, Query),
 1815    uri_components(URL, Components).

 pengine_event(?EventTerm) is det

 pengine_event(?EventTerm, +Options) is det

Examines the pengine's event queue and if necessary blocks execution until a term that unifies to Term arrives in the queue. After a term from the queue has been unified to Term, the term is deleted from the queue.

Valid options are:

timeout(+Time)

Time is a float or integer and specifies the maximum time to wait in seconds. If no event has arrived before the time is up EventTerm is bound to the atom timeout.

listen(+Id)

Only listen to events from the pengine identified by Id. */

 1836pengine_event(Event) :-
 1837    pengine_event(Event, []).
 1838
 1839pengine_event(Event, Options) :-
 1840    thread_self(Self),
 1841    option(listen(Id), Options, _),
 1842    (   thread_get_message(Self, pengine_event(Id, Event), Options)
 1843    ->  true
 1844    ;   Event = timeout
 1845    ),
 1846    update_remote_destroy(Event).
 1847
 1848update_remote_destroy(Event) :-
 1849    destroy_event(Event),
 1850    arg(1, Event, Id),
 1851    pengine_remote(Id, _Server),
 1852    !,
 1853    pengine_unregister_remote(Id).
 1854update_remote_destroy(_).
 1855
 1856destroy_event(destroy(_)).
 1857destroy_event(destroy(_,_)).
 1858destroy_event(create(_,Features)) :-
 1859    memberchk(answer(Answer), Features),
 1860    !,
 1861    nonvar(Answer),
 1862    destroy_event(Answer).

 pengine_event_loop(:Closure, +Options) is det

Starts an event loop accepting event terms sent to the current pengine or thread. For each such event E, calls ignore(call(Closure, E)). A closure thus acts as a handler for the event. Some events are also treated specially:

create(ID, Term)

The ID is placed in a list of active pengines.

destroy(ID)

The ID is removed from the list of active pengines. When the last pengine ID is removed, the loop terminates.

output(ID, Term)

The predicate pengine_pull_response/2 is called.

Valid options are:

autoforward(+To)

Forwards received event terms to slaves. To is either all, all_but_sender or a Prolog list of NameOrIDs. [not yet implemented]

*/

 1891pengine_event_loop(Closure, Options) :-
 1892    child(_,_),
 1893    !,
 1894    pengine_event(Event),
 1895    (   option(autoforward(all), Options) % TODO: Implement all_but_sender and list of IDs
 1896    ->  forall(child(_,ID), pengine_send(ID, Event))
 1897    ;   true
 1898    ),
 1899    pengine_event_loop(Event, Closure, Options).
 1900pengine_event_loop(_, _).
 1901
 1902:- meta_predicate
 1903    pengine_process_event(+, 1, -, +). 1904
 1905pengine_event_loop(Event, Closure, Options) :-
 1906    pengine_process_event(Event, Closure, Continue, Options),
 1907    (   Continue == true
 1908    ->  pengine_event_loop(Closure, Options)
 1909    ;   true
 1910    ).
 1911
 1912pengine_process_event(create(ID, T), Closure, Continue, Options) :-
 1913    debug(pengine(transition), '~q: 1 = /~q => 2', [ID, create(T)]),
 1914    (   select(answer(First), T, T1)
 1915    ->  ignore(call(Closure, create(ID, T1))),
 1916        pengine_process_event(First, Closure, Continue, Options)
 1917    ;   ignore(call(Closure, create(ID, T))),
 1918        Continue = true
 1919    ).
 1920pengine_process_event(output(ID, Msg), Closure, true, _Options) :-
 1921    debug(pengine(transition), '~q: 3 = /~q => 4', [ID, output(Msg)]),
 1922    ignore(call(Closure, output(ID, Msg))),
 1923    pengine_pull_response(ID, []).
 1924pengine_process_event(debug(ID, Msg), Closure, true, _Options) :-
 1925    debug(pengine(transition), '~q: 3 = /~q => 4', [ID, debug(Msg)]),
 1926    ignore(call(Closure, debug(ID, Msg))),
 1927    pengine_pull_response(ID, []).
 1928pengine_process_event(prompt(ID, Term), Closure, true, _Options) :-
 1929    debug(pengine(transition), '~q: 3 = /~q => 5', [ID, prompt(Term)]),
 1930    ignore(call(Closure, prompt(ID, Term))).
 1931pengine_process_event(success(ID, Sol, _Proj, _Time, More), Closure, true, _) :-
 1932    debug(pengine(transition), '~q: 3 = /~q => 6/2', [ID, success(Sol, More)]),
 1933    ignore(call(Closure, success(ID, Sol, More))).
 1934pengine_process_event(failure(ID, _Time), Closure, true, _Options) :-
 1935    debug(pengine(transition), '~q: 3 = /~q => 2', [ID, failure]),
 1936    ignore(call(Closure, failure(ID))).
 1937pengine_process_event(error(ID, Error), Closure, Continue, _Options) :-
 1938    debug(pengine(transition), '~q: 3 = /~q => 2', [ID, error(Error)]),
 1939    (   call(Closure, error(ID, Error))
 1940    ->  Continue = true
 1941    ;   forall(child(_,Child), pengine_destroy(Child)),
 1942        throw(Error)
 1943    ).
 1944pengine_process_event(stop(ID), Closure, true, _Options) :-
 1945    debug(pengine(transition), '~q: 7 = /~q => 2', [ID, stop]),
 1946    ignore(call(Closure, stop(ID))).
 1947pengine_process_event(destroy(ID, Event), Closure, Continue, Options) :-
 1948    pengine_process_event(Event, Closure, _, Options),
 1949    pengine_process_event(destroy(ID), Closure, Continue, Options).
 1950pengine_process_event(destroy(ID), Closure, true, _Options) :-
 1951    retractall(child(_,ID)),
 1952    debug(pengine(transition), '~q: 1 = /~q => 0', [ID, destroy]),
 1953    ignore(call(Closure, destroy(ID))).