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-2016, 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:- use_module(library(http/http_dispatch)).   71:- use_module(library(http/http_parameters)).   72:- use_module(library(http/http_client)).   73:- use_module(library(http/http_json)).   74:- use_module(library(http/http_open)).   75:- use_module(library(http/http_stream)).   76:- use_module(library(http/http_wrapper)).   77:- use_module(library(http/http_cors)).   78:- use_module(library(thread_pool)).   79:- use_module(library(broadcast)).   80:- use_module(library(uri)).   81:- use_module(library(filesex)).   82:- use_module(library(time)).   83:- use_module(library(lists)).   84:- use_module(library(charsio)).   85:- use_module(library(apply)).   86:- use_module(library(aggregate)).   87:- use_module(library(option)).   88:- use_module(library(settings)).   89:- use_module(library(debug)).   90:- use_module(library(error)).   91:- use_module(library(sandbox)).   92:- use_module(library(modules)).   93:- use_module(library(term_to_json)).   94:- if(exists_source(library(uuid))).   95:- use_module(library(uuid)).   96:- endif.   97
   98
   99:- meta_predicate
  100    pengine_create(:),
  101    pengine_rpc(+, +, :),
  102    pengine_event_loop(1, +).  103
  104:- multifile
  105    write_result/3,                 % +Format, +Event, +Dict
  106    event_to_json/3,                % +Event, -JSON, +Format
  107    prepare_module/3,               % +Module, +Application, +Options
  108    prepare_goal/3,                 % +GoalIn, -GoalOut, +Options
  109    authentication_hook/3,          % +Request, +Application, -User
  110    not_sandboxed/2.                % +User, +App
  111
  112:- predicate_options(pengine_create/1, 1,
  113                     [ id(-atom),
  114                       alias(atom),
  115                       application(atom),
  116                       destroy(boolean),
  117                       server(atom),
  118                       ask(compound),
  119                       template(compound),
  120                       chunk(integer),
  121                       bindings(list),
  122                       src_list(list),
  123                       src_text(any),           % text
  124                       src_url(atom),
  125                       src_predicates(list)
  126                     ]).  127:- predicate_options(pengine_ask/3, 3,
  128                     [ template(any),
  129                       chunk(integer),
  130                       bindings(list)
  131                     ]).  132:- predicate_options(pengine_next/2, 2,
  133                     [ chunk(integer),
  134                       pass_to(pengine_send/3, 3)
  135                     ]).  136:- predicate_options(pengine_stop/2, 2,
  137                     [ pass_to(pengine_send/3, 3)
  138                     ]).  139:- predicate_options(pengine_respond/3, 2,
  140                     [ pass_to(pengine_send/3, 3)
  141                     ]).  142:- predicate_options(pengine_rpc/3, 3,
  143                     [ chunk(integer),
  144                       pass_to(pengine_create/1, 1)
  145                     ]).  146:- predicate_options(pengine_send/3, 3,
  147                     [ delay(number)
  148                     ]).  149:- predicate_options(pengine_event/2, 2,
  150                     [ pass_to(thread_get_message/3, 3)
  151                     ]).  152:- predicate_options(pengine_pull_response/2, 2,
  153                     [ pass_to(http_open/3, 3)
  154                     ]).  155:- predicate_options(pengine_event_loop/2, 2,
  156                     []).                       % not yet implemented
  157
  158% :- debug(pengine(transition)).
  159:- debug(pengine(debug)).               % handle pengine_debug in pengine_rpc/3.
  160
  161goal_expansion(random_delay, Expanded) :-
  162    (   debugging(pengine(delay))
  163    ->  Expanded = do_random_delay
  164    ;   Expanded = true
  165    ).
  166
  167do_random_delay :-
  168    Delay is random(20)/1000,
  169    sleep(Delay).
  170
  171:- meta_predicate                       % internal meta predicates
  172    solve(+, ?, 0, +),
  173    findnsols_no_empty(+, ?, 0, -),
  174    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. */

  229pengine_create(M:Options0) :-
  230    translate_local_sources(Options0, Options, M),
  231    (   select_option(server(BaseURL), Options, RestOptions)
  232    ->  remote_pengine_create(BaseURL, RestOptions)
  233    ;   local_pengine_create(Options)
  234    ).

 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.

  248translate_local_sources(OptionsIn, Options, Module) :-
  249    translate_local_sources(OptionsIn, Sources, Options2, Module),
  250    (   Sources == []
  251    ->  Options = Options2
  252    ;   Sources = [Source]
  253    ->  Options = [src_text(Source)|Options2]
  254    ;   atomics_to_string(Sources, Source)
  255    ->  Options = [src_text(Source)|Options2]
  256    ).
  257
  258translate_local_sources([], [], [], _).
  259translate_local_sources([H0|T], [S0|S], Options, M) :-
  260    nonvar(H0),
  261    translate_local_source(H0, S0, M),
  262    !,
  263    translate_local_sources(T, S, Options, M).
  264translate_local_sources([H|T0], S, [H|T], M) :-
  265    translate_local_sources(T0, S, T, M).
  266
  267translate_local_source(src_predicates(PIs), Source, M) :-
  268    must_be(list, PIs),
  269    with_output_to(string(Source),
  270                   maplist(list_in_module(M), PIs)).
  271translate_local_source(src_list(Terms), Source, _) :-
  272    must_be(list, Terms),
  273    with_output_to(string(Source),
  274                   forall(member(Term, Terms),
  275                          format('~k .~n', [Term]))).
  276translate_local_source(src_text(Source), Source, _).
  277
  278list_in_module(M, PI) :-
  279    listing(M:PI).

 pengine_send(+NameOrID, +Term) is det

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

  286pengine_send(Target, Event) :-
  287    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. */

  302pengine_send(Target, Event, Options) :-
  303    must_be(atom, Target),
  304    pengine_send2(Target, Event, Options).
  305
  306pengine_send2(self, Event, Options) :-
  307    !,
  308    thread_self(Queue),
  309    delay_message(queue(Queue), Event, Options).
  310pengine_send2(Name, Event, Options) :-
  311    child(Name, Target),
  312    !,
  313    delay_message(pengine(Target), Event, Options).
  314pengine_send2(Target, Event, Options) :-
  315    delay_message(pengine(Target), Event, Options).
  316
  317delay_message(Target, Event, Options) :-
  318    option(delay(Delay), Options),
  319    !,
  320    alarm(Delay,
  321          send_message(Target, Event, Options),
  322          _AlarmID,
  323          [remove(true)]).
  324delay_message(Target, Event, Options) :-
  325    random_delay,
  326    send_message(Target, Event, Options).
  327
  328send_message(queue(Queue), Event, _) :-
  329    thread_send_message(Queue, pengine_request(Event)).
  330send_message(pengine(Pengine), Event, Options) :-
  331    (   pengine_remote(Pengine, Server)
  332    ->  remote_pengine_send(Server, Pengine, Event, Options)
  333    ;   pengine_thread(Pengine, Thread)
  334    ->  thread_send_message(Thread, pengine_request(Event))
  335    ;   existence_error(pengine, Pengine)
  336    ).

 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.

  343pengine_request(Request) :-
  344    pengine_self(Self),
  345    get_pengine_application(Self, Application),
  346    setting(Application:idle_limit, IdleLimit),
  347    thread_self(Me),
  348    (   thread_get_message(Me, pengine_request(Request), [timeout(IdleLimit)])
  349    ->  true
  350    ;   Request = destroy
  351    ).

 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.

  364pengine_reply(Event) :-
  365    pengine_parent(Queue),
  366    pengine_reply(Queue, Event).
  367
  368pengine_reply(_Queue, _Event0) :-
  369    nb_current(pengine_idle_limit_exceeded, true),
  370    !.
  371pengine_reply(Queue, Event0) :-
  372    arg(1, Event0, ID),
  373    wrap_first_answer(ID, Event0, Event),
  374    random_delay,
  375    debug(pengine(event), 'Reply to ~p: ~p', [Queue, Event]),
  376    (   pengine_self(ID)
  377    ->  get_pengine_application(ID, Application),
  378        setting(Application:idle_limit, IdleLimit),
  379        debug(pengine(reply), 'Sending ~p, timout: ~q', [Event, IdleLimit]),
  380        (   thread_send_message(Queue, pengine_event(ID, Event),
  381                                [ timeout(IdleLimit)
  382                                ])
  383        ->  true
  384        ;   thread_self(Me),
  385            debug(pengine(reply), 'pengine_reply: timeout for ~q (thread ~q)',
  386                  [ID, Me]),
  387            nb_setval(pengine_idle_limit_exceeded, true),
  388            thread_detach(Me),
  389            abort
  390        )
  391    ;   thread_send_message(Queue, pengine_event(ID, Event))
  392    ).
  393
  394wrap_first_answer(ID, Event0, CreateEvent) :-
  395    wrap_first_answer_in_create_event(CreateEvent, [answer(Event0)]),
  396    arg(1, CreateEvent, ID),
  397    !,
  398    retract(wrap_first_answer_in_create_event(CreateEvent, [answer(Event0)])).
  399wrap_first_answer(_ID, Event, Event).
  400
  401
  402empty_queue :-
  403    pengine_parent(Queue),
  404    empty_queue(Queue, 0, Discarded),
  405    debug(pengine(abort), 'Abort: discarded ~D messages', [Discarded]).
  406
  407empty_queue(Queue, C0, C) :-
  408    thread_get_message(Queue, _Term, [timeout(0)]),
  409    !,
  410    C1 is C0+1,
  411    empty_queue(Queue, C1, C).
  412empty_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).

*/

  477pengine_ask(ID, Query, Options) :-
  478    partition(pengine_ask_option, Options, AskOptions, SendOptions),
  479    pengine_send(ID, ask(Query, AskOptions), SendOptions).
  480
  481
  482pengine_ask_option(template(_)).
  483pengine_ask_option(chunk(_)).
  484pengine_ask_option(bindings(_)).
  485pengine_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).

*/

  529pengine_next(ID, Options) :-
  530    select_option(chunk(Count), Options, Options1),
  531    !,
  532    pengine_send(ID, next(Count), Options1).
  533pengine_next(ID, Options) :-
  534    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).

*/

  550pengine_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. */

  561pengine_abort(Name) :-
  562    (   child(Name, Pengine)
  563    ->  true
  564    ;   Pengine = Name
  565    ),
  566    (   pengine_remote(Pengine, Server)
  567    ->  remote_pengine_abort(Server, Pengine, [])
  568    ;   pengine_thread(Pengine, Thread),
  569        debug(pengine(abort), 'Signalling thread ~p', [Thread]),
  570        catch(thread_signal(Thread, throw(abort_query)), _, true)
  571    ).

 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. */

  581pengine_destroy(ID) :-
  582    pengine_destroy(ID, []).
  583
  584pengine_destroy(Name, Options) :-
  585    (   child(Name, ID)
  586    ->  true
  587    ;   ID = Name
  588    ),
  589    option(force(true), Options),
  590    !,
  591    (   pengine_thread(ID, Thread)
  592    ->  catch(thread_signal(Thread, abort),
  593              error(existence_error(thread, _), _), true)
  594    ;   true
  595    ).
  596pengine_destroy(ID, _) :-
  597    catch(pengine_send(ID, destroy),
  598          error(existence_error(pengine, ID), _),
  599          retractall(child(_,ID))).
  600
  601
  602/*================= pengines administration =======================
  603*/

 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)

  614:- dynamic
  615    current_pengine/6,              % Id, ParentId, Thread, URL, App, Destroy
  616    pengine_queue/4,                % Id, Queue, TimeOut, Time
  617    output_queue/3,                 % Id, Queue, Time
  618    pengine_user/2,                 % Id, User
  619    pengine_data/2.                 % Id, Data
  620:- volatile
  621    current_pengine/6,
  622    pengine_queue/4,
  623    output_queue/3,
  624    pengine_user/2,
  625    pengine_data/2.  626
  627:- thread_local
  628    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

  634pengine_register_local(Id, Thread, Queue, URL, Application, Destroy) :-
  635    asserta(current_pengine(Id, Queue, Thread, URL, Application, Destroy)).
  636
  637pengine_register_remote(Id, URL, Application, Destroy) :-
  638    thread_self(Queue),
  639    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.

  647pengine_unregister(Id) :-
  648    thread_self(Me),
  649    (   current_pengine(Id, Queue, Me, http, _, _)
  650    ->  with_mutex(pengine, sync_destroy_queue_from_pengine(Id, Queue))
  651    ;   true
  652    ),
  653    retractall(current_pengine(Id, _, Me, _, _, _)),
  654    retractall(pengine_user(Id, _)),
  655    retractall(pengine_data(Id, _)).
  656
  657pengine_unregister_remote(Id) :-
  658    retractall(current_pengine(Id, _Parent, 0, _, _, _)).

 pengine_self(-Id) is det

True if the current thread is a pengine with Id.

  664pengine_self(Id) :-
  665    thread_self(Thread),
  666    current_pengine(Id, _Parent, Thread, _URL, _Application, _Destroy).
  667
  668pengine_parent(Parent) :-
  669    nb_getval(pengine_parent, Parent).
  670
  671pengine_thread(Pengine, Thread) :-
  672    current_pengine(Pengine, _Parent, Thread, _URL, _Application, _Destroy),
  673    Thread \== 0,
  674    !.
  675
  676pengine_remote(Pengine, URL) :-
  677    current_pengine(Pengine, _Parent, 0, URL, _Application, _Destroy).
  678
  679get_pengine_application(Pengine, Application) :-
  680    current_pengine(Pengine, _Parent, _, _URL, Application, _Destroy),
  681    !.
  682
  683get_pengine_module(Pengine, Pengine).
  684
  685:- if(current_predicate(uuid/2)).  686pengine_uuid(Id) :-
  687    uuid(Id, [version(4)]).             % Version 4 is random.
  688:- else.  689:- use_module(library(random)).  690pengine_uuid(Id) :-
  691    Max is 1<<128,
  692    random_between(0, Max, Num),
  693    atom_number(Id, Num).
  694:- endif.

 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).

*/

  710pengine_application(Application) :-
  711    throw(error(context_error(nodirective,
  712                             pengine_application(Application)), _)).
  713
  714:- multifile
  715    system:term_expansion/2,
  716    current_application/1.

 current_pengine_application(?Application) is nondet

True when Application is a currently defined application.

See also

- pengine_application/1

  724current_pengine_application(Application) :-
  725    current_application(Application).
  726
  727
  728% Default settings for all applications
  729
  730:- setting(thread_pool_size, integer, 100,
  731           'Maximum number of pengines this application can run.').  732:- setting(thread_pool_stacks, list(compound), [],
  733           'Maximum stack sizes for pengines this application can run.').  734:- setting(slave_limit, integer, 3,
  735           'Maximum number of slave pengines a master pengine can create.').  736:- setting(time_limit, number, 300,
  737           'Maximum time to wait for output').  738:- setting(idle_limit, number, 300,
  739           'Pengine auto-destroys when idle for this time').  740:- setting(safe_goal_limit, number, 10,
  741           'Maximum time to try proving safety of the goal').  742:- setting(program_space, integer, 100_000_000,
  743           'Maximum memory used by predicates').  744:- setting(allow_from, list(atom), [*],
  745           'IP addresses from which remotes are allowed to connect').  746:- setting(deny_from, list(atom), [],
  747           'IP addresses from which remotes are NOT allowed to connect').  748:- setting(debug_info, boolean, false,
  749           'Keep information to support source-level debugging').  750
  751
  752system:term_expansion((:- pengine_application(Application)), Expanded) :-
  753    must_be(atom, Application),
  754    (   module_property(Application, file(_))
  755    ->  permission_error(create, pengine_application, Application)
  756    ;   true
  757    ),
  758    expand_term((:- setting(Application:thread_pool_size, integer,
  759                            setting(pengines:thread_pool_size),
  760                            'Maximum number of pengines this \c
  761                            application can run.')),
  762                ThreadPoolSizeSetting),
  763    expand_term((:- setting(Application:thread_pool_stacks, list(compound),
  764                            setting(pengines:thread_pool_stacks),
  765                            'Maximum stack sizes for pengines \c
  766                            this application can run.')),
  767                ThreadPoolStacksSetting),
  768    expand_term((:- setting(Application:slave_limit, integer,
  769                            setting(pengines:slave_limit),
  770                            'Maximum number of local slave pengines \c
  771                            a master pengine can create.')),
  772                SlaveLimitSetting),
  773    expand_term((:- setting(Application:time_limit, number,
  774                            setting(pengines:time_limit),
  775                            'Maximum time to wait for output')),
  776                TimeLimitSetting),
  777    expand_term((:- setting(Application:idle_limit, number,
  778                            setting(pengines:idle_limit),
  779                            'Pengine auto-destroys when idle for this time')),
  780                IdleLimitSetting),
  781    expand_term((:- setting(Application:safe_goal_limit, number,
  782                            setting(pengines:safe_goal_limit),
  783                            'Maximum time to try proving safety of the goal')),
  784                SafeGoalLimitSetting),
  785    expand_term((:- setting(Application:program_space, integer,
  786                            setting(pengines:program_space),
  787                            'Maximum memory used by predicates')),
  788                ProgramSpaceSetting),
  789    expand_term((:- setting(Application:allow_from, list(atom),
  790                            setting(pengines:allow_from),
  791                            'IP addresses from which remotes are allowed \c
  792                            to connect')),
  793                AllowFromSetting),
  794    expand_term((:- setting(Application:deny_from, list(atom),
  795                            setting(pengines:deny_from),
  796                            'IP addresses from which remotes are NOT \c
  797                            allowed to connect')),
  798                DenyFromSetting),
  799    expand_term((:- setting(Application:debug_info, boolean,
  800                            setting(pengines:debug_info),
  801                            'Keep information to support source-level \c
  802                            debugging')),
  803                DebugInfoSetting),
  804    flatten([ pengines:current_application(Application),
  805              ThreadPoolSizeSetting,
  806              ThreadPoolStacksSetting,
  807              SlaveLimitSetting,
  808              TimeLimitSetting,
  809              IdleLimitSetting,
  810              SafeGoalLimitSetting,
  811              ProgramSpaceSetting,
  812              AllowFromSetting,
  813              DenyFromSetting,
  814              DebugInfoSetting
  815            ], Expanded).
  816
  817% Register default application
  818
  819:- 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. */

  854pengine_property(Id, Prop) :-
  855    nonvar(Id), nonvar(Prop),
  856    pengine_property2(Id, Prop),
  857    !.
  858pengine_property(Id, Prop) :-
  859    pengine_property2(Prop, Id).
  860
  861pengine_property2(self(Id), Id) :-
  862    current_pengine(Id, _Parent, _Thread, _URL, _Application, _Destroy).
  863pengine_property2(module(Id), Id) :-
  864    current_pengine(Id, _Parent, _Thread, _URL, _Application, _Destroy).
  865pengine_property2(alias(Alias), Id) :-
  866    child(Alias, Id),
  867    Alias \== Id.
  868pengine_property2(thread(Thread), Id) :-
  869    current_pengine(Id, _Parent, Thread, _URL, _Application, _Destroy),
  870    Thread \== 0.
  871pengine_property2(remote(Server), Id) :-
  872    current_pengine(Id, _Parent, 0, Server, _Application, _Destroy).
  873pengine_property2(application(Application), Id) :-
  874    current_pengine(Id, _Parent, _Thread, _Server, Application, _Destroy).
  875pengine_property2(destroy(Destroy), Id) :-
  876    current_pengine(Id, _Parent, _Thread, _Server, _Application, Destroy).
  877pengine_property2(parent(Parent), Id) :-
  878    current_pengine(Id, Parent, _Thread, _URL, _Application, _Destroy).
  879pengine_property2(source(SourceID, Source), Id) :-
  880    pengine_data(Id, source(SourceID, Source)).

 pengine_output(+Term) is det

Sends Term to the parent pengine or thread. */

  887pengine_output(Term) :-
  888    pengine_self(Me),
  889    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 */

  904pengine_debug(Format, Args) :-
  905    pengine_parent(Queue),
  906    pengine_self(Self),
  907    catch(safe_goal(format(atom(_), Format, Args)), E, true),
  908    (   var(E)
  909    ->  format(atom(Message), Format, Args)
  910    ;   message_to_string(E, Message)
  911    ),
  912    pengine_reply(Queue, debug(Self, Message)).
  913
  914
  915/*================= Local pengine =======================
  916*/

 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.

  927local_pengine_create(Options) :-
  928    thread_self(Self),
  929    option(application(Application), Options, pengine_sandbox),
  930    create(Self, Child, Options, local, Application),
  931    option(alias(Name), Options, Child),
  932    assert(child(Name, Child)).

 thread_pool:create_pool(+Application) is det

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

  939thread_pool:create_pool(Application) :-
  940    current_application(Application),
  941    setting(Application:thread_pool_size, Size),
  942    setting(Application:thread_pool_stacks, Stacks),
  943    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

  953create(Queue, Child, Options, local, Application) :-
  954    !,
  955    pengine_child_id(Child),
  956    create0(Queue, Child, Options, local, Application).
  957create(Queue, Child, Options, URL, Application) :-
  958    pengine_child_id(Child),
  959    catch(create0(Queue, Child, Options, URL, Application),
  960          Error,
  961          create_error(Queue, Child, Error)).
  962
  963pengine_child_id(Child) :-
  964    (   nonvar(Child)
  965    ->  true
  966    ;   pengine_uuid(Child)
  967    ).
  968
  969create_error(Queue, Child, Error) :-
  970    pengine_reply(Queue, error(Child, Error)).
  971
  972create0(Queue, Child, Options, URL, Application) :-
  973    (  current_application(Application)
  974    -> true
  975    ;  existence_error(pengine_application, Application)
  976    ),
  977    (   URL \== http                    % pengine is _not_ a child of the
  978                                        % HTTP server thread
  979    ->  aggregate_all(count, child(_,_), Count),
  980        setting(Application:slave_limit, Max),
  981        (   Count >= Max
  982        ->  throw(error(resource_error(max_pengines), _))
  983        ;   true
  984        )
  985    ;   true
  986    ),
  987    partition(pengine_create_option, Options, PengineOptions, RestOptions),
  988    thread_create_in_pool(
  989        Application,
  990        pengine_main(Queue, PengineOptions, Application), ChildThread,
  991        [ at_exit(pengine_done)
  992        | RestOptions
  993        ]),
  994    option(destroy(Destroy), PengineOptions, true),
  995    pengine_register_local(Child, ChildThread, Queue, URL, Application, Destroy),
  996    thread_send_message(ChildThread, pengine_registered(Child)),
  997    (   option(id(Id), Options)
  998    ->  Id = Child
  999    ;   true
 1000    ).
 1001
 1002pengine_create_option(src_text(_)).
 1003pengine_create_option(src_url(_)).
 1004pengine_create_option(application(_)).
 1005pengine_create_option(destroy(_)).
 1006pengine_create_option(ask(_)).
 1007pengine_create_option(template(_)).
 1008pengine_create_option(bindings(_)).
 1009pengine_create_option(chunk(_)).
 1010pengine_create_option(alias(_)).
 1011pengine_create_option(user(_)).

 pengine_done is det

Called from the pengine thread at_exit option. Destroys child pengines using pengine_destroy/1.

 1019:- public
 1020    pengine_done/0. 1021
 1022pengine_done :-
 1023    thread_self(Me),
 1024    (   thread_property(Me, status(exception('$aborted'))),
 1025        thread_detach(Me),
 1026        pengine_self(Pengine)
 1027    ->  catch(pengine_reply(destroy(Pengine, abort(Pengine))),
 1028              error(_,_), true)
 1029    ;   true
 1030    ),
 1031    forall(child(_Name, Child),
 1032           pengine_destroy(Child)),
 1033    pengine_self(Id),
 1034    pengine_unregister(Id).

 pengine_main(+Parent, +Options, +Application)

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

 1042:- thread_local wrap_first_answer_in_create_event/2. 1043
 1044:- meta_predicate
 1045    pengine_prepare_source(:, +). 1046
 1047pengine_main(Parent, Options, Application) :-
 1048    fix_streams,
 1049    thread_get_message(pengine_registered(Self)),
 1050    nb_setval(pengine_parent, Parent),
 1051    pengine_register_user(Options),
 1052    set_prolog_flag(mitigate_spectre, true),
 1053    catch(in_temporary_module(
 1054              Self,
 1055              pengine_prepare_source(Application, Options),
 1056              pengine_create_and_loop(Self, Application, Options)),
 1057          prepare_source_failed,
 1058          pengine_terminate(Self)).
 1059
 1060pengine_create_and_loop(Self, Application, Options) :-
 1061    setting(Application:slave_limit, SlaveLimit),
 1062    CreateEvent = create(Self, [slave_limit(SlaveLimit)|Extra]),
 1063    (   option(ask(Query0), Options)
 1064    ->  asserta(wrap_first_answer_in_create_event(CreateEvent, Extra)),
 1065        (   string(Query0)                      % string is not callable
 1066        ->  (   option(template(TemplateS), Options)
 1067            ->  Ask2 = Query0-TemplateS
 1068            ;   Ask2 = Query0
 1069            ),
 1070            catch(ask_to_term(Ask2, Self, Query, Template, Bindings),
 1071                  Error, true),
 1072            (   var(Error)
 1073            ->  true
 1074            ;   send_error(Error),
 1075                throw(prepare_source_failed)
 1076            )
 1077        ;   Query = Query0,
 1078            option(template(Template), Options, Query),
 1079            option(bindings(Bindings), Options, [])
 1080        ),
 1081        option(chunk(Chunk), Options, 1),
 1082        pengine_ask(Self, Query,
 1083                    [ template(Template),
 1084                      chunk(Chunk),
 1085                      bindings(Bindings)
 1086                    ])
 1087    ;   Extra = [],
 1088        pengine_reply(CreateEvent)
 1089    ),
 1090    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.

 1100ask_to_term(Ask-Template, Module, Ask1, Template1, Bindings) :-
 1101    !,
 1102    format(string(AskTemplate), 't((~s),(~s))', [Template, Ask]),
 1103    term_string(t(Template1,Ask1), AskTemplate,
 1104                [ variable_names(Bindings0),
 1105                  module(Module)
 1106                ]),
 1107    phrase(template_bindings(Template1, Bindings0), Bindings).
 1108ask_to_term(Ask, Module, Ask1, Template, Bindings1) :-
 1109    term_string(Ask1, Ask,
 1110                [ variable_names(Bindings),
 1111                  module(Module)
 1112                ]),
 1113    exclude(anon, Bindings, Bindings1),
 1114    dict_create(Template, swish_default_template, Bindings1).
 1115
 1116template_bindings(Var, Bindings) -->
 1117    { var(Var) }, !,
 1118    (   { var_binding(Bindings, Var, Binding)
 1119        }
 1120    ->  [Binding]
 1121    ;   []
 1122    ).
 1123template_bindings([H|T], Bindings) -->
 1124    !,
 1125    template_bindings(H, Bindings),
 1126    template_bindings(T, Bindings).
 1127template_bindings(Compoound, Bindings) -->
 1128    { compound(Compoound), !,
 1129      compound_name_arguments(Compoound, _, Args)
 1130    },
 1131    template_bindings(Args, Bindings).
 1132template_bindings(_, _) --> [].
 1133
 1134var_binding(Bindings, Var, Binding) :-
 1135    member(Binding, Bindings),
 1136    arg(2, Binding, V),
 1137    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.

 1144fix_streams :-
 1145    fix_stream(current_output).
 1146
 1147fix_stream(Name) :-
 1148    is_cgi_stream(Name),
 1149    !,
 1150    debug(pengine(stream), '~w is a CGI stream!', [Name]),
 1151    set_stream(user_output, alias(Name)).
 1152fix_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.

 1161pengine_prepare_source(Module:Application, Options) :-
 1162    setting(Application:program_space, SpaceLimit),
 1163    set_module(Module:program_space(SpaceLimit)),
 1164    delete_import_module(Module, user),
 1165    add_import_module(Module, Application, start),
 1166    catch(prep_module(Module, Application, Options), Error, true),
 1167    (   var(Error)
 1168    ->  true
 1169    ;   send_error(Error),
 1170        throw(prepare_source_failed)
 1171    ).
 1172
 1173prep_module(Module, Application, Options) :-
 1174    maplist(copy_flag(Module, Application), [var_prefix]),
 1175    forall(prepare_module(Module, Application, Options), true),
 1176    setup_call_cleanup(
 1177        '$set_source_module'(OldModule, Module),
 1178        maplist(process_create_option(Module), Options),
 1179        '$set_source_module'(OldModule)).
 1180
 1181copy_flag(Module, Application, Flag) :-
 1182    current_prolog_flag(Application:Flag, Value),
 1183    !,
 1184    set_prolog_flag(Module:Flag, Value).
 1185copy_flag(_, _, _).
 1186
 1187process_create_option(Application, src_text(Text)) :-
 1188    !,
 1189    pengine_src_text(Text, Application).
 1190process_create_option(Application, src_url(URL)) :-
 1191    !,
 1192    pengine_src_url(URL, Application).
 1193process_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.

 1216pengine_main_loop(ID) :-
 1217    catch(guarded_main_loop(ID), abort_query, pengine_aborted(ID)).
 1218
 1219pengine_aborted(ID) :-
 1220    thread_self(Self),
 1221    debug(pengine(abort), 'Aborting ~p (thread ~p)', [ID, Self]),
 1222    empty_queue,
 1223    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.

 1236guarded_main_loop(ID) :-
 1237    pengine_request(Request),
 1238    (   Request = destroy
 1239    ->  debug(pengine(transition), '~q: 2 = ~q => 1', [ID, destroy]),
 1240        pengine_terminate(ID)
 1241    ;   Request = ask(Goal, Options)
 1242    ->  debug(pengine(transition), '~q: 2 = ~q => 3', [ID, ask(Goal)]),
 1243        ask(ID, Goal, Options)
 1244    ;   debug(pengine(transition), '~q: 2 = ~q => 2', [ID, protocol_error]),
 1245        pengine_reply(error(ID, error(protocol_error, _))),
 1246        guarded_main_loop(ID)
 1247    ).
 1248
 1249
 1250pengine_terminate(ID) :-
 1251    pengine_reply(destroy(ID)),
 1252    thread_self(Me),            % Make the thread silently disappear
 1253    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.

 1264solve(Chunk, Template, Goal, ID) :-
 1265    prolog_current_choice(Choice),
 1266    State = count(Chunk),
 1267    statistics(cputime, Epoch),
 1268    Time = time(Epoch),
 1269    nb_current('$variable_names', Bindings),
 1270    filter_template(Template, Bindings, Template2),
 1271    '$current_typein_module'(CurrTypeIn),
 1272    (   '$set_typein_module'(ID),
 1273        call_cleanup(catch(findnsols_no_empty(State, Template2,
 1274                                              set_projection(Goal, Bindings),
 1275                                              Result),
 1276                           Error, true),
 1277                     query_done(Det, CurrTypeIn)),
 1278        arg(1, Time, T0),
 1279        statistics(cputime, T1),
 1280        CPUTime is T1-T0,
 1281        (   var(Error)
 1282        ->  projection(Projection),
 1283            (   var(Det)
 1284            ->  pengine_reply(success(ID, Result, Projection,
 1285                                      CPUTime, true)),
 1286                more_solutions(ID, Choice, State, Time)
 1287            ;   !,                      % commit
 1288                destroy_or_continue(success(ID, Result, Projection,
 1289                                            CPUTime, false))
 1290            )
 1291        ;   !,                          % commit
 1292            (   Error == abort_query
 1293            ->  throw(Error)
 1294            ;   destroy_or_continue(error(ID, Error))
 1295            )
 1296        )
 1297    ;   !,                              % commit
 1298        arg(1, Time, T0),
 1299        statistics(cputime, T1),
 1300        CPUTime is T1-T0,
 1301        destroy_or_continue(failure(ID, CPUTime))
 1302    ).
 1303solve(_, _, _, _).                      % leave a choice point
 1304
 1305query_done(true, CurrTypeIn) :-
 1306    '$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.

 1315set_projection(Goal, Bindings) :-
 1316    b_setval('$variable_names', Bindings),
 1317    call(Goal).
 1318
 1319projection(Projection) :-
 1320    nb_current('$variable_names', Bindings),
 1321    !,
 1322    maplist(var_name, Bindings, Projection).
 1323projection([]).

 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.

 1333filter_template(Template0, Bindings, Template) :-
 1334    is_dict(Template0, swish_default_template),
 1335    !,
 1336    dict_create(Template, swish_default_template, Bindings).
 1337filter_template(Template, _Bindings, Template).
 1338
 1339findnsols_no_empty(N, Template, Goal, List) :-
 1340    findnsols(N, Template, Goal, List),
 1341    List \== [].
 1342
 1343destroy_or_continue(Event) :-
 1344    arg(1, Event, ID),
 1345    (   pengine_property(ID, destroy(true))
 1346    ->  thread_self(Me),
 1347        thread_detach(Me),
 1348        pengine_reply(destroy(ID, Event))
 1349    ;   pengine_reply(Event),
 1350        garbage_collect,                % minimise our footprint
 1351        trim_stacks,
 1352        guarded_main_loop(ID)
 1353    ).

 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.

 1371more_solutions(ID, Choice, State, Time) :-
 1372    pengine_request(Event),
 1373    more_solutions(Event, ID, Choice, State, Time).
 1374
 1375more_solutions(stop, ID, _Choice, _State, _Time) :-
 1376    !,
 1377    debug(pengine(transition), '~q: 6 = ~q => 7', [ID, stop]),
 1378    destroy_or_continue(stop(ID)).
 1379more_solutions(next, ID, _Choice, _State, Time) :-
 1380    !,
 1381    debug(pengine(transition), '~q: 6 = ~q => 3', [ID, next]),
 1382    statistics(cputime, T0),
 1383    nb_setarg(1, Time, T0),
 1384    fail.
 1385more_solutions(next(Count), ID, _Choice, State, Time) :-
 1386    Count > 0,
 1387    !,
 1388    debug(pengine(transition), '~q: 6 = ~q => 3', [ID, next(Count)]),
 1389    nb_setarg(1, State, Count),
 1390    statistics(cputime, T0),
 1391    nb_setarg(1, Time, T0),
 1392    fail.
 1393more_solutions(ask(Goal, Options), ID, Choice, _State, _Time) :-
 1394    !,
 1395    debug(pengine(transition), '~q: 6 = ~q => 3', [ID, ask(Goal)]),
 1396    prolog_cut_to(Choice),
 1397    ask(ID, Goal, Options).
 1398more_solutions(destroy, ID, _Choice, _State, _Time) :-
 1399    !,
 1400    debug(pengine(transition), '~q: 6 = ~q => 1', [ID, destroy]),
 1401    pengine_terminate(ID).
 1402more_solutions(Event, ID, Choice, State, Time) :-
 1403    debug(pengine(transition), '~q: 6 = ~q => 6', [ID, protocol_error(Event)]),
 1404    pengine_reply(error(ID, error(protocol_error, _))),
 1405    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.

 1413ask(ID, Goal, Options) :-
 1414    catch(prepare_goal(ID, Goal, Goal1, Options), Error, true),
 1415    !,
 1416    (   var(Error)
 1417    ->  option(template(Template), Options, Goal),
 1418        option(chunk(N), Options, 1),
 1419        solve(N, Template, Goal1, ID)
 1420    ;   pengine_reply(error(ID, Error)),
 1421        guarded_main_loop(ID)
 1422    ).

 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?

 1436prepare_goal(ID, Goal0, Module:Goal, Options) :-
 1437    option(bindings(Bindings), Options, []),
 1438    b_setval('$variable_names', Bindings),
 1439    (   prepare_goal(Goal0, Goal1, Options)
 1440    ->  true
 1441    ;   Goal1 = Goal0
 1442    ),
 1443    get_pengine_module(ID, Module),
 1444    setup_call_cleanup(
 1445        '$set_source_module'(Old, Module),
 1446        expand_goal(Goal1, Goal),
 1447        '$set_source_module'(_, Old)),
 1448    (   pengine_not_sandboxed(ID)
 1449    ->  true
 1450    ;   get_pengine_application(ID, App),
 1451        setting(App:safe_goal_limit, Limit),
 1452        catch(call_with_time_limit(
 1453                  Limit,
 1454                  safe_goal(Module:Goal)), E, true)
 1455    ->  (   var(E)
 1456        ->  true
 1457        ;   E = time_limit_exceeded
 1458        ->  throw(error(sandbox(time_limit_exceeded, Limit),_))
 1459        ;   throw(E)
 1460        )
 1461    ).

 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.

 1481pengine_not_sandboxed(ID) :-
 1482    pengine_user(ID, User),
 1483    pengine_property(ID, application(App)),
 1484    not_sandboxed(User, App),
 1485    !.

 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. */

 1507pengine_pull_response(Pengine, Options) :-
 1508    pengine_remote(Pengine, Server),
 1509    !,
 1510    remote_pengine_pull_response(Server, Pengine, Options).
 1511pengine_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. */

 1520pengine_input(Prompt, Term) :-
 1521    pengine_self(Self),
 1522    pengine_parent(Parent),
 1523    pengine_reply(Parent, prompt(Self, Prompt)),
 1524    pengine_request(Request),
 1525    (   Request = input(Input)
 1526    ->  Term = Input
 1527    ;   Request == destroy
 1528    ->  abort
 1529    ;   throw(error(protocol_error,_))
 1530    ).

 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).

*/

 1547pengine_respond(Pengine, Input, Options) :-
 1548    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.

 1557send_error(error(Formal, context(prolog_stack(Frames), Message))) :-
 1558    is_list(Frames),
 1559    !,
 1560    with_output_to(string(Stack),
 1561                   print_prolog_backtrace(current_output, Frames)),
 1562    pengine_self(Self),
 1563    replace_blobs(Formal, Formal1),
 1564    replace_blobs(Message, Message1),
 1565    pengine_reply(error(Self, error(Formal1,
 1566                                    context(prolog_stack(Stack), Message1)))).
 1567send_error(Error) :-
 1568    pengine_self(Self),
 1569    replace_blobs(Error, Error1),
 1570    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.

 1578replace_blobs(Blob, Atom) :-
 1579    blob(Blob, Type), Type \== text,
 1580    !,
 1581    format(atom(Atom), '~p', [Blob]).
 1582replace_blobs(Term0, Term) :-
 1583    compound(Term0),
 1584    !,
 1585    compound_name_arguments(Term0, Name, Args0),
 1586    maplist(replace_blobs, Args0, Args),
 1587    compound_name_arguments(Term, Name, Args).
 1588replace_blobs(Term, Term).
 1589
 1590
 1591/*================= Remote pengines =======================
 1592*/
 1593
 1594
 1595remote_pengine_create(BaseURL, Options) :-
 1596    partition(pengine_create_option, Options, PengineOptions0, RestOptions),
 1597        (       option(ask(Query), PengineOptions0),
 1598                \+ option(template(_Template), PengineOptions0)
 1599        ->      PengineOptions = [template(Query)|PengineOptions0]
 1600        ;       PengineOptions = PengineOptions0
 1601        ),
 1602    options_to_dict(PengineOptions, PostData),
 1603    remote_post_rec(BaseURL, create, PostData, Reply, RestOptions),
 1604    arg(1, Reply, ID),
 1605    (   option(id(ID2), Options)
 1606    ->  ID = ID2
 1607    ;   true
 1608    ),
 1609    option(alias(Name), Options, ID),
 1610    assert(child(Name, ID)),
 1611    (   (   functor(Reply, create, _)   % actually created
 1612        ;   functor(Reply, output, _)   % compiler messages
 1613        )
 1614    ->  option(application(Application), PengineOptions, pengine_sandbox),
 1615        option(destroy(Destroy), PengineOptions, true),
 1616        pengine_register_remote(ID, BaseURL, Application, Destroy)
 1617    ;   true
 1618    ),
 1619    thread_self(Queue),
 1620    pengine_reply(Queue, Reply).
 1621
 1622options_to_dict(Options, Dict) :-
 1623    select_option(ask(Ask), Options, Options1),
 1624    select_option(template(Template), Options1, Options2),
 1625    !,
 1626    no_numbered_var_in(Ask+Template),
 1627    findall(AskString-TemplateString,
 1628            ask_template_to_strings(Ask, Template, AskString, TemplateString),
 1629            [ AskString-TemplateString ]),
 1630    options_to_dict(Options2, Dict0),
 1631    Dict = Dict0.put(_{ask:AskString,template:TemplateString}).
 1632options_to_dict(Options, Dict) :-
 1633    maplist(prolog_option, Options, Options1),
 1634    dict_create(Dict, _, Options1).
 1635
 1636no_numbered_var_in(Term) :-
 1637    sub_term(Sub, Term),
 1638    subsumes_term('$VAR'(_), Sub),
 1639    !,
 1640    domain_error(numbered_vars_free_term, Term).
 1641no_numbered_var_in(_).
 1642
 1643ask_template_to_strings(Ask, Template, AskString, TemplateString) :-
 1644    numbervars(Ask+Template, 0, _),
 1645    WOpts = [ numbervars(true), ignore_ops(true), quoted(true) ],
 1646    format(string(AskTemplate), '~W\n~W', [ Ask, WOpts,
 1647                                            Template, WOpts
 1648                                          ]),
 1649    split_string(AskTemplate, "\n", "", [AskString, TemplateString]).
 1650
 1651prolog_option(Option0, Option) :-
 1652    create_option_type(Option0, term),
 1653    !,
 1654    Option0 =.. [Name,Value],
 1655    format(string(String), '~k', [Value]),
 1656    Option =.. [Name,String].
 1657prolog_option(Option, Option).
 1658
 1659create_option_type(ask(_),         term).
 1660create_option_type(template(_),    term).
 1661create_option_type(application(_), atom).
 1662
 1663remote_pengine_send(BaseURL, ID, Event, Options) :-
 1664    remote_send_rec(BaseURL, send, ID, [event=Event], Reply, Options),
 1665    thread_self(Queue),
 1666    pengine_reply(Queue, Reply).
 1667
 1668remote_pengine_pull_response(BaseURL, ID, Options) :-
 1669    remote_send_rec(BaseURL, pull_response, ID, [], Reply, Options),
 1670    thread_self(Queue),
 1671    pengine_reply(Queue, Reply).
 1672
 1673remote_pengine_abort(BaseURL, ID, Options) :-
 1674    remote_send_rec(BaseURL, abort, ID, [], Reply, Options),
 1675    thread_self(Queue),
 1676    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.

 1683remote_send_rec(Server, Action, ID, [event=Event], Reply, Options) :-
 1684    !,
 1685    server_url(Server, Action, [id=ID], URL),
 1686    http_open(URL, Stream,              % putting this in setup_call_cleanup/3
 1687              [ post(prolog(Event))     % makes it impossible to interrupt.
 1688              | Options
 1689              ]),
 1690    call_cleanup(
 1691        read_prolog_reply(Stream, Reply),
 1692        close(Stream)).
 1693remote_send_rec(Server, Action, ID, Params, Reply, Options) :-
 1694    server_url(Server, Action, [id=ID|Params], URL),
 1695    http_open(URL, Stream, Options),
 1696    call_cleanup(
 1697        read_prolog_reply(Stream, Reply),
 1698        close(Stream)).
 1699
 1700remote_post_rec(Server, Action, Data, Reply, Options) :-
 1701    server_url(Server, Action, [], URL),
 1702    probe(Action, URL),
 1703    http_open(URL, Stream,
 1704              [ post(json(Data))
 1705              | Options
 1706              ]),
 1707    call_cleanup(
 1708        read_prolog_reply(Stream, Reply),
 1709        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.

 1717probe(create, URL) :-
 1718    !,
 1719    http_open(URL, Stream, [method(options)]),
 1720    close(Stream).
 1721probe(_, _).
 1722
 1723read_prolog_reply(In, Reply) :-
 1724    set_stream(In, encoding(utf8)),
 1725    read(In, Reply0),
 1726    rebind_cycles(Reply0, Reply).
 1727
 1728rebind_cycles(@(Reply, Bindings), Reply) :-
 1729    is_list(Bindings),
 1730    !,
 1731    maplist(bind, Bindings).
 1732rebind_cycles(Reply, Reply).
 1733
 1734bind(Var = Value) :-
 1735    Var = Value.
 1736
 1737server_url(Server, Action, Params, URL) :-
 1738    uri_components(Server, Components0),
 1739    uri_query_components(Query, Params),
 1740    uri_data(path, Components0, Path0),
 1741    atom_concat('pengine/', Action, PAction),
 1742    directory_file_path(Path0, PAction, Path),
 1743    uri_data(path, Components0, Path, Components),
 1744    uri_data(search, Components, Query),
 1745    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. */

 1766pengine_event(Event) :-
 1767    pengine_event(Event, []).
 1768
 1769pengine_event(Event, Options) :-
 1770    thread_self(Self),
 1771    option(listen(Id), Options, _),
 1772    (   thread_get_message(Self, pengine_event(Id, Event), Options)
 1773    ->  true
 1774    ;   Event = timeout
 1775    ),
 1776    update_remote_destroy(Event).
 1777
 1778update_remote_destroy(Event) :-
 1779    destroy_event(Event),
 1780    arg(1, Event, Id),
 1781    pengine_remote(Id, _Server),
 1782    !,
 1783    pengine_unregister_remote(Id).
 1784update_remote_destroy(_).
 1785
 1786destroy_event(destroy(_)).
 1787destroy_event(destroy(_,_)).
 1788destroy_event(create(_,Features)) :-
 1789    memberchk(answer(Answer), Features),
 1790    !,
 1791    nonvar(Answer),
 1792    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]

*/

 1821pengine_event_loop(Closure, Options) :-
 1822    child(_,_),
 1823    !,
 1824    pengine_event(Event),
 1825    (   option(autoforward(all), Options) % TODO: Implement all_but_sender and list of IDs
 1826    ->  forall(child(_,ID), pengine_send(ID, Event))
 1827    ;   true
 1828    ),
 1829    pengine_event_loop(Event, Closure, Options).
 1830pengine_event_loop(_, _).
 1831
 1832:- meta_predicate
 1833    pengine_process_event(+, 1, -, +). 1834
 1835pengine_event_loop(Event, Closure, Options) :-
 1836    pengine_process_event(Event, Closure, Continue, Options),
 1837    (   Continue == true
 1838    ->  pengine_event_loop(Closure, Options)
 1839    ;   true
 1840    ).
 1841
 1842pengine_process_event(create(ID, T), Closure, Continue, Options) :-
 1843    debug(pengine(transition), '~q: 1 = /~q => 2', [ID, create(T)]),
 1844    (   select(answer(First), T, T1)
 1845    ->  ignore(call(Closure, create(ID, T1))),
 1846        pengine_process_event(First, Closure, Continue, Options)
 1847    ;   ignore(call(Closure, create(ID, T))),
 1848        Continue = true
 1849    ).
 1850pengine_process_event(output(ID, Msg), Closure, true, _Options) :-
 1851    debug(pengine(transition), '~q: 3 = /~q => 4', [ID, output(Msg)]),
 1852    ignore(call(Closure, output(ID, Msg))),
 1853    pengine_pull_response(ID, []).
 1854pengine_process_event(debug(ID, Msg), Closure, true, _Options) :-
 1855    debug(pengine(transition), '~q: 3 = /~q => 4', [ID, debug(Msg)]),
 1856    ignore(call(Closure, debug(ID, Msg))),
 1857    pengine_pull_response(ID, []).
 1858pengine_process_event(prompt(ID, Term), Closure, true, _Options) :-
 1859    debug(pengine(transition), '~q: 3 = /~q => 5', [ID, prompt(Term)]),
 1860    ignore(call(Closure, prompt(ID, Term))).
 1861pengine_process_event(success(ID, Sol, _Proj, _Time, More), Closure, true, _) :-
 1862    debug(pengine(transition), '~q: 3 = /~q => 6/2', [ID, success(Sol, More)]),
 1863    ignore(call(Closure, success(ID, Sol, More))).
 1864pengine_process_event(failure(ID, _Time), Closure, true, _Options) :-
 1865    debug(pengine(transition), '~q: 3 = /~q => 2', [ID, failure]),
 1866    ignore(call(Closure, failure(ID))).
 1867pengine_process_event(error(ID, Error), Closure, Continue, _Options) :-
 1868    debug(pengine(transition), '~q: 3 = /~q => 2', [ID, error(Error)]),
 1869    (   call(Closure, error(ID, Error))
 1870    ->  Continue = true
 1871    ;   forall(child(_,Child), pengine_destroy(Child)),
 1872        throw(Error)
 1873    ).
 1874pengine_process_event(stop(ID), Closure, true, _Options) :-
 1875    debug(pengine(transition), '~q: 7 = /~q => 2', [ID, stop]),
 1876    ignore(call(Closure, stop(ID))).
 1877pengine_process_event(destroy(ID, Event), Closure, Continue, Options) :-
 1878    pengine_process_event(Event, Closure, _, Options),
 1879    pengine_process_event(destroy(ID), Closure, Continue, Options).
 1880pengine_process_event(destroy(ID), Closure, true, _Options) :-
 1881    retractall(child(_,ID)),
 1882    debug(pengine(transition), '~q: 1 = /~q => 0', [ID, destroy]),
 1883    ignore(call(Closure, destroy(ID))).

 pengine_rpc(+URL, +Query) is nondet

 pengine_rpc(+URL, +Query, +Options) is nondet

Semantically equivalent to the sequence below, except that the query is executed in (and in the Prolog context of) the pengine server referred to by URL, rather than locally.

  copy_term_nat(Query, Copy),  % attributes are not copied to the server
  call(Copy),			 % executed on server at URL
  Query = Copy.

Valid options are:

chunk(+Integer)

Can be used to reduce the number of network roundtrips being made. See pengine_ask/3.

timeout(+Time)

Wait at most Time seconds for the next event from the server. The default is defined by the setting pengines:time_limit.

Remaining options (except the server option) are passed to pengine_create/1. */

 1912pengine_rpc(URL, Query) :-
 1913    pengine_rpc(URL, Query, []).
 1914
 1915pengine_rpc(URL, Query, M:Options0) :-
 1916    translate_local_sources(Options0, Options1, M),
 1917    (  option(timeout(_), Options1)
 1918    -> Options = Options1
 1919    ;  setting(time_limit, Limit),
 1920       Options = [timeout(Limit)|Options1]
 1921    ),
 1922    term_variables(Query, Vars),
 1923    Template =.. [v|Vars],
 1924    State = destroy(true),              % modified by process_event/4
 1925    setup_call_catcher_cleanup(
 1926        pengine_create([ ask(Query),
 1927                         template(Template),
 1928                         server(URL),
 1929                         id(Id)
 1930                       | Options
 1931                       ]),
 1932        wait_event(Template, State, [listen(Id)|Options]),
 1933        Why,
 1934        pengine_destroy_and_wait(State, Id, Why)).
 1935
 1936pengine_destroy_and_wait(destroy(true), Id, Why) :-
 1937    !,
 1938    debug(pengine(rpc), 'Destroying RPC because of ~p', [Why]),
 1939    pengine_destroy(Id),
 1940    wait_destroy(Id, 10).
 1941pengine_destroy_and_wait(_, _, Why) :-
 1942    debug(pengine(rpc), 'Not destroying RPC (~p)', [Why]).
 1943
 1944wait_destroy(Id, _) :-
 1945    \+ child(_, Id),
 1946    !.
 1947wait_destroy(Id, N) :-
 1948    pengine_event(Event, [listen(Id),timeout(10)]),
 1949    !,
 1950    (   destroy_event(Event)
 1951    ->  retractall(child(_,Id))
 1952    ;   succ(N1, N)
 1953    ->  wait_destroy(Id, N1)
 1954    ;   debug(pengine(rpc), 'RPC did not answer to destroy ~p', [Id]),
 1955        pengine_unregister_remote(Id),
 1956        retractall(child(_,Id))
 1957    ).
 1958
 1959wait_event(Template, State, Options) :-
 1960    pengine_event(Event, Options),
 1961    debug(pengine(event), 'Received ~p', [Event]),
 1962    process_event(Event, Template, State, Options).
 1963
 1964process_event(create(_ID, Features), Template, State, Options) :-
 1965    memberchk(answer(First), Features),
 1966    process_event(First, Template, State, Options).
 1967process_event(error(_ID, Error), _Template, _, _Options) :-
 1968    throw(Error).
 1969process_event(failure(_ID, _Time), _Template, _, _Options) :-
 1970    fail.
 1971process_event(prompt(ID, Prompt), Template, State, Options) :-
 1972    pengine_rpc_prompt(ID, Prompt, Reply),
 1973    pengine_send(ID, input(Reply)),
 1974    wait_event(Template, State, Options).
 1975process_event(output(ID, Term), Template, State, Options) :-
 1976    pengine_rpc_output(ID, Term),
 1977    pengine_pull_response(ID, Options),
 1978    wait_event(Template, State, Options).
 1979process_event(debug(ID, Message), Template, State, Options) :-
 1980    debug(pengine(debug), '~w', [Message]),
 1981    pengine_pull_response(ID, Options),
 1982    wait_event(Template, State, Options).
 1983process_event(success(_ID, Solutions, _Proj, _Time, false),
 1984              Template, _, _Options) :-
 1985    !,
 1986    member(Template, Solutions).
 1987process_event(success(ID, Solutions, _Proj, _Time, true),
 1988              Template, State, Options) :-
 1989    (   member(Template, Solutions)
 1990    ;   pengine_next(ID, Options),
 1991        wait_event(Template, State, Options)
 1992    ).
 1993process_event(destroy(ID, Event), Template, State, Options) :-
 1994    !,
 1995    retractall(child(_,ID)),
 1996    nb_setarg(1, State, false),
 1997    debug(pengine(destroy), 'State: ~p~n', [State]),
 1998    process_event(Event, Template, State, Options).
 1999% compatibility with older versions of the protocol.
 2000process_event(success(ID, Solutions, Time, More),
 2001              Template, State, Options) :-
 2002    process_event(success(ID, Solutions, _Proj, Time, More),
 2003                  Template, State, Options).
 2004
 2005
 2006pengine_rpc_prompt(ID, Prompt, Term) :-
 2007    prompt(ID, Prompt, Term0),
 2008    !,
 2009    Term = Term0.
 2010pengine_rpc_prompt(_ID, Prompt, Term) :-
 2011    setup_call_cleanup(
 2012        prompt(Old, Prompt),
 2013        read(Term),
 2014        prompt(_, Old)).
 2015
 2016pengine_rpc_output(ID, Term) :-
 2017    output(ID, Term),
 2018    !.
 2019pengine_rpc_output(_ID, Term) :-
 2020    print(Term).