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

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

  254pengine_create(M:Options0) :-
  255    translate_local_sources(Options0, Options, M),
  256    (   select_option(server(BaseURL), Options, RestOptions)
  257    ->  remote_pengine_create(BaseURL, RestOptions)
  258    ;   local_pengine_create(Options)
  259    ).

 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.

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

 pengine_send(+NameOrID, +Term) is det

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

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

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

 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.

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

 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.

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

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. If false, the Pengine goal is not executed using findall/3 and friends and we do not backtrack immediately over the goal. As a result, changes to backtrackable global state are retained. This is similar that using set_prolog_flag(toplevel_mode, recursive).

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

*/

  517pengine_ask(ID, Query, Options) :-
  518    partition(pengine_ask_option, Options, AskOptions, SendOptions),
  519    pengine_send(ID, ask(Query, AskOptions), SendOptions).
  520
  521
  522pengine_ask_option(template(_)).
  523pengine_ask_option(chunk(_)).
  524pengine_ask_option(bindings(_)).
  525pengine_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. This may not be used if the goal was started with chunk(false).

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

*/

  570pengine_next(ID, Options) :-
  571    select_option(chunk(Count), Options, Options1),
  572    !,
  573    pengine_send(ID, next(Count), Options1).
  574pengine_next(ID, Options) :-
  575    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).

*/

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

  602pengine_abort(Name) :-
  603    (   child(Name, Pengine)
  604    ->  true
  605    ;   Pengine = Name
  606    ),
  607    (   pengine_remote(Pengine, Server)
  608    ->  remote_pengine_abort(Server, Pengine, [])
  609    ;   pengine_thread(Pengine, Thread),
  610        debug(pengine(abort), 'Signalling thread ~p', [Thread]),
  611        catch(thread_signal(Thread, throw(abort_query)), _, true)
  612    ).

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

  622pengine_destroy(ID) :-
  623    pengine_destroy(ID, []).
  624
  625pengine_destroy(Name, Options) :-
  626    (   child(Name, ID)
  627    ->  true
  628    ;   ID = Name
  629    ),
  630    option(force(true), Options),
  631    !,
  632    (   pengine_thread(ID, Thread)
  633    ->  catch(thread_signal(Thread, abort),
  634              error(existence_error(thread, _), _), true)
  635    ;   true
  636    ).
  637pengine_destroy(ID, Options) :-
  638    catch(pengine_send(ID, destroy, Options),
  639          error(existence_error(pengine, ID), _),
  640          retractall(child(_,ID))).
  641
  642
  643/*================= pengines administration =======================
  644*/

 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)

  655:- dynamic
  656    current_pengine/6,              % Id, ParentId, Thread, URL, App, Destroy
  657    pengine_queue/4,                % Id, Queue, TimeOut, Time
  658    output_queue/3,                 % Id, Queue, Time
  659    pengine_user/2,                 % Id, User
  660    pengine_data/2,                 % Id, Data
  661    pengine_detached/2.             % Id, Data
  662:- volatile
  663    current_pengine/6,
  664    pengine_queue/4,
  665    output_queue/3,
  666    pengine_user/2,
  667    pengine_data/2,
  668    pengine_detached/2.  669
  670:- thread_local
  671    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

  677pengine_register_local(Id, Thread, Queue, URL, Application, Destroy) :-
  678    asserta(current_pengine(Id, Queue, Thread, URL, Application, Destroy)).
  679
  680pengine_register_remote(Id, URL, Application, Destroy) :-
  681    thread_self(Queue),
  682    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.

  690pengine_unregister(Id) :-
  691    thread_self(Me),
  692    (   current_pengine(Id, Queue, Me, http, _, _)
  693    ->  with_mutex(pengine, sync_destroy_queue_from_pengine(Id, Queue))
  694    ;   true
  695    ),
  696    retractall(current_pengine(Id, _, Me, _, _, _)),
  697    retractall(pengine_user(Id, _)),
  698    retractall(pengine_data(Id, _)).
  699
  700pengine_unregister_remote(Id) :-
  701    retractall(current_pengine(Id, _Parent, 0, _, _, _)).

 pengine_self(-Id) is det

True if the current thread is a pengine with Id.

  707pengine_self(Id) :-
  708    thread_self(Thread),
  709    current_pengine(Id, _Parent, Thread, _URL, _Application, _Destroy).
  710
  711pengine_parent(Parent) :-
  712    nb_getval(pengine_parent, Parent).
  713
  714pengine_thread(Pengine, Thread) :-
  715    current_pengine(Pengine, _Parent, Thread, _URL, _Application, _Destroy),
  716    Thread \== 0,
  717    !.
  718
  719pengine_remote(Pengine, URL) :-
  720    current_pengine(Pengine, _Parent, 0, URL, _Application, _Destroy).
  721
  722get_pengine_application(Pengine, Application) :-
  723    current_pengine(Pengine, _Parent, _, _URL, Application, _Destroy),
  724    !.
  725
  726get_pengine_module(Pengine, Pengine).
  727
  728:- if(current_predicate(uuid/2)).  729pengine_uuid(Id) :-
  730    uuid(Id, [version(4)]).             % Version 4 is random.
  731:- else.  732pengine_uuid(Id) :-
  733    (   current_prolog_flag(max_integer, Max1)
  734    ->  Max is Max1-1
  735    ;   Max is 1<<128
  736    ),
  737    random_between(0, Max, Num),
  738    atom_number(Id, Num).
  739:- 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.

  756:- meta_predicate protect_pengine(+, 0).  757
  758protect_pengine(Id, Goal) :-
  759    term_hash(Id, Hash),
  760    LockN is Hash mod 64,
  761    atom_concat(pengine_done_, LockN, Lock),
  762    with_mutex(Lock,
  763               (   pengine_thread(Id, _)
  764               ->  Goal
  765               ;   Goal
  766               )).

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

*/

  783pengine_application(Application) :-
  784    throw(error(context_error(nodirective,
  785                             pengine_application(Application)), _)).
  786
  787:- multifile
  788    system:term_expansion/2,
  789    current_application/1.

 current_pengine_application(?Application) is nondet

True when Application is a currently defined application.

See also

- pengine_application/1

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

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

 pengine_output(+Term) is det

Sends Term to the parent pengine or thread. */

  964pengine_output(Term) :-
  965    pengine_self(Me),
  966    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 */

  981pengine_debug(Format, Args) :-
  982    pengine_parent(Queue),
  983    pengine_self(Self),
  984    catch(safe_goal(format(atom(_), Format, Args)), E, true),
  985    (   var(E)
  986    ->  format(atom(Message), Format, Args)
  987    ;   message_to_string(E, Message)
  988    ),
  989    pengine_reply(Queue, debug(Self, Message)).
  990
  991
  992/*================= Local pengine =======================
  993*/

 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.

 1004local_pengine_create(Options) :-
 1005    thread_self(Self),
 1006    option(application(Application), Options, pengine_sandbox),
 1007    create(Self, Child, Options, local, Application),
 1008    option(alias(Name), Options, Child),
 1009    assert(child(Name, Child)).

 thread_pool:create_pool(+Application) is det

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

 1016:- multifile thread_pool:create_pool/1. 1017
 1018thread_pool:create_pool(Application) :-
 1019    current_application(Application),
 1020    setting(Application:thread_pool_size, Size),
 1021    setting(Application:thread_pool_stacks, Stacks),
 1022    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

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

 1099:- public
 1100    pengine_done/0. 1101
 1102pengine_done :-
 1103    thread_self(Me),
 1104    (   thread_property(Me, status(exception(Ex))),
 1105        abort_exception(Ex),
 1106        thread_detach(Me),
 1107        pengine_self(Pengine)
 1108    ->  catch(pengine_reply(destroy(Pengine, abort(Pengine))),
 1109              error(_,_), true)
 1110    ;   true
 1111    ),
 1112    forall(child(_Name, Child),
 1113           pengine_destroy(Child)),
 1114    pengine_self(Id),
 1115    protect_pengine(Id, pengine_unregister(Id)).
 1116
 1117abort_exception('$aborted').
 1118abort_exception(unwind(abort)).

 pengine_main(+Parent, +Options, +Application)

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

 1125:- thread_local wrap_first_answer_in_create_event/2. 1126
 1127:- meta_predicate
 1128    pengine_prepare_source(:, +). 1129
 1130pengine_main(Parent, Options, Application) :-
 1131    fix_streams,
 1132    thread_get_message(pengine_registered(Self)),
 1133    nb_setval(pengine_parent, Parent),
 1134    pengine_register_user(Options),
 1135    set_prolog_flag(mitigate_spectre, true),
 1136    catch(in_temporary_module(
 1137              Self,
 1138              pengine_prepare_source(Application, Options),
 1139              pengine_create_and_loop(Self, Application, Options)),
 1140          prepare_source_failed,
 1141          pengine_terminate(Self)).
 1142
 1143pengine_create_and_loop(Self, Application, Options) :-
 1144    setting(Application:slave_limit, SlaveLimit),
 1145    CreateEvent = create(Self, [slave_limit(SlaveLimit)|Extra]),
 1146    (   option(ask(Query0), Options)
 1147    ->  asserta(wrap_first_answer_in_create_event(CreateEvent, Extra)),
 1148        (   string(Query0)                      % string is not callable
 1149        ->  (   option(template(TemplateS), Options)
 1150            ->  Ask2 = Query0-TemplateS
 1151            ;   Ask2 = Query0
 1152            ),
 1153            catch(ask_to_term(Ask2, Self, Query, Template, Bindings),
 1154                  Error, true),
 1155            (   var(Error)
 1156            ->  true
 1157            ;   send_error(Error),
 1158                throw(prepare_source_failed)
 1159            )
 1160        ;   Query = Query0,
 1161            option(template(Template), Options, Query),
 1162            option(bindings(Bindings), Options, [])
 1163        ),
 1164        option(chunk(Chunk), Options, 1),
 1165        pengine_ask(Self, Query,
 1166                    [ template(Template),
 1167                      chunk(Chunk),
 1168                      bindings(Bindings)
 1169                    ])
 1170    ;   Extra = [],
 1171        pengine_reply(CreateEvent)
 1172    ),
 1173    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.

 1183ask_to_term(Ask-Template, Module, Ask1, Template1, Bindings) :-
 1184    !,
 1185    format(string(AskTemplate), 't((~s),(~s))', [Template, Ask]),
 1186    term_string(t(Template1,Ask1), AskTemplate,
 1187                [ variable_names(Bindings0),
 1188                  module(Module)
 1189                ]),
 1190    phrase(template_bindings(Template1, Bindings0), Bindings).
 1191ask_to_term(Ask, Module, Ask1, Template, Bindings1) :-
 1192    term_string(Ask1, Ask,
 1193                [ variable_names(Bindings),
 1194                  module(Module)
 1195                ]),
 1196    exclude(anon, Bindings, Bindings1),
 1197    dict_create(Template, swish_default_template, Bindings1).
 1198
 1199template_bindings(Var, Bindings) -->
 1200    { var(Var) }, !,
 1201    (   { var_binding(Bindings, Var, Binding)
 1202        }
 1203    ->  [Binding]
 1204    ;   []
 1205    ).
 1206template_bindings([H|T], Bindings) -->
 1207    !,
 1208    template_bindings(H, Bindings),
 1209    template_bindings(T, Bindings).
 1210template_bindings(Compoound, Bindings) -->
 1211    { compound(Compoound), !,
 1212      compound_name_arguments(Compoound, _, Args)
 1213    },
 1214    template_bindings(Args, Bindings).
 1215template_bindings(_, _) --> [].
 1216
 1217var_binding(Bindings, Var, Binding) :-
 1218    member(Binding, Bindings),
 1219    arg(2, Binding, V),
 1220    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.

 1227fix_streams :-
 1228    fix_stream(current_output).
 1229
 1230fix_stream(Name) :-
 1231    is_cgi_stream(Name),
 1232    !,
 1233    debug(pengine(stream), '~w is a CGI stream!', [Name]),
 1234    set_stream(user_output, alias(Name)).
 1235fix_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.

 1244pengine_prepare_source(Module:Application, Options) :-
 1245    setting(Application:program_space, SpaceLimit),
 1246    set_module(Module:program_space(SpaceLimit)),
 1247    delete_import_module(Module, user),
 1248    add_import_module(Module, Application, start),
 1249    catch(prep_module(Module, Application, Options), Error, true),
 1250    (   var(Error)
 1251    ->  true
 1252    ;   send_error(Error),
 1253        throw(prepare_source_failed)
 1254    ).
 1255
 1256prep_module(Module, Application, Options) :-
 1257    maplist(copy_flag(Module, Application), [var_prefix]),
 1258    forall(prepare_module(Module, Application, Options), true),
 1259    setup_call_cleanup(
 1260        '$set_source_module'(OldModule, Module),
 1261        maplist(process_create_option(Module), Options),
 1262        '$set_source_module'(OldModule)).
 1263
 1264copy_flag(Module, Application, Flag) :-
 1265    current_prolog_flag(Application:Flag, Value),
 1266    !,
 1267    set_prolog_flag(Module:Flag, Value).
 1268copy_flag(_, _, _).
 1269
 1270process_create_option(Application, src_text(Text)) :-
 1271    !,
 1272    pengine_src_text(Text, Application).
 1273process_create_option(Application, src_url(URL)) :-
 1274    !,
 1275    pengine_src_url(URL, Application).
 1276process_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.

 1299pengine_main_loop(ID) :-
 1300    catch(guarded_main_loop(ID), abort_query, pengine_aborted(ID)).
 1301
 1302pengine_aborted(ID) :-
 1303    thread_self(Self),
 1304    debug(pengine(abort), 'Aborting ~p (thread ~p)', [ID, Self]),
 1305    empty_queue,
 1306    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.

 1319guarded_main_loop(ID) :-
 1320    pengine_request(Request),
 1321    (   Request = destroy
 1322    ->  debug(pengine(transition), '~q: 2 = ~q => 1', [ID, destroy]),
 1323        pengine_terminate(ID)
 1324    ;   Request = ask(Goal, Options)
 1325    ->  debug(pengine(transition), '~q: 2 = ~q => 3', [ID, ask(Goal)]),
 1326        ask(ID, Goal, Options)
 1327    ;   debug(pengine(transition), '~q: 2 = ~q => 2', [ID, protocol_error]),
 1328        pengine_reply(error(ID, error(protocol_error, _))),
 1329        guarded_main_loop(ID)
 1330    ).
 1331
 1332
 1333pengine_terminate(ID) :-
 1334    pengine_reply(destroy(ID)),
 1335    thread_self(Me),            % Make the thread silently disappear
 1336    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.

 1347solve(Chunk, Template, Goal, ID) :-
 1348    prolog_current_choice(Choice),
 1349    (   integer(Chunk)
 1350    ->  State = count(Chunk)
 1351    ;   Chunk == false
 1352    ->  State = no_chunk
 1353    ;   domain_error(chunk, Chunk)
 1354    ),
 1355    statistics(cputime, Epoch),
 1356    Time = time(Epoch),
 1357    nb_current('$variable_names', Bindings),
 1358    filter_template(Template, Bindings, Template2),
 1359    '$current_typein_module'(CurrTypeIn),
 1360    (   '$set_typein_module'(ID),
 1361        call_cleanup(catch(findnsols_no_empty(State, Template2,
 1362                                              set_projection(Goal, Bindings),
 1363                                              Result),
 1364                           Error, true),
 1365                     query_done(Det, CurrTypeIn)),
 1366        arg(1, Time, T0),
 1367        statistics(cputime, T1),
 1368        CPUTime is T1-T0,
 1369        forall(pengine_flush_output_hook, true),
 1370        (   var(Error)
 1371        ->  projection(Projection),
 1372            (   var(Det)
 1373            ->  pengine_reply(success(ID, Result, Projection,
 1374                                      CPUTime, true)),
 1375                more_solutions(ID, Choice, State, Time)
 1376            ;   !,                      % commit
 1377                destroy_or_continue(success(ID, Result, Projection,
 1378                                            CPUTime, false))
 1379            )
 1380        ;   !,                          % commit
 1381            (   Error == abort_query
 1382            ->  throw(Error)
 1383            ;   destroy_or_continue(error(ID, Error))
 1384            )
 1385        )
 1386    ;   !,                              % commit
 1387        arg(1, Time, T0),
 1388        statistics(cputime, T1),
 1389        CPUTime is T1-T0,
 1390        destroy_or_continue(failure(ID, CPUTime))
 1391    ).
 1392solve(_, _, _, _).                      % leave a choice point
 1393
 1394query_done(true, CurrTypeIn) :-
 1395    '$set_typein_module'(CurrTypeIn).

 set_projection(:Goal, +Bindings)

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

 1404set_projection(Goal, Bindings) :-
 1405    b_setval('$variable_names', Bindings),
 1406    call(Goal).
 1407
 1408projection(Projection) :-
 1409    nb_current('$variable_names', Bindings),
 1410    !,
 1411    maplist(var_name, Bindings, Projection).
 1412projection([]).

 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.

 1422filter_template(Template0, Bindings, Template) :-
 1423    is_dict(Template0, swish_default_template),
 1424    !,
 1425    dict_create(Template, swish_default_template, Bindings).
 1426filter_template(Template, _Bindings, Template).
 1427
 1428findnsols_no_empty(no_chunk, Template, Goal, List) =>
 1429    List = [Template],
 1430    call(Goal).
 1431findnsols_no_empty(State, Template, Goal, List) =>
 1432    findnsols(State, Template, Goal, List),
 1433    List \== [].
 1434
 1435destroy_or_continue(Event) :-
 1436    arg(1, Event, ID),
 1437    (   pengine_property(ID, destroy(true))
 1438    ->  thread_self(Me),
 1439        thread_detach(Me),
 1440        pengine_reply(destroy(ID, Event))
 1441    ;   pengine_reply(Event),
 1442        guarded_main_loop(ID)
 1443    ).

 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.

 1461more_solutions(ID, Choice, State, Time) :-
 1462    pengine_request(Event),
 1463    more_solutions(Event, ID, Choice, State, Time).
 1464
 1465more_solutions(stop, ID, _Choice, _State, _Time) :-
 1466    !,
 1467    debug(pengine(transition), '~q: 6 = ~q => 7', [ID, stop]),
 1468    destroy_or_continue(stop(ID)).
 1469more_solutions(next, ID, _Choice, _State, Time) :-
 1470    !,
 1471    debug(pengine(transition), '~q: 6 = ~q => 3', [ID, next]),
 1472    statistics(cputime, T0),
 1473    nb_setarg(1, Time, T0),
 1474    fail.
 1475more_solutions(next(Count), ID, _Choice, State, Time) :-
 1476    Count > 0,
 1477    State = count(_),                   % else fallthrough to protocol error
 1478    !,
 1479    debug(pengine(transition), '~q: 6 = ~q => 3', [ID, next(Count)]),
 1480    nb_setarg(1, State, Count),
 1481    statistics(cputime, T0),
 1482    nb_setarg(1, Time, T0),
 1483    fail.
 1484more_solutions(ask(Goal, Options), ID, Choice, _State, _Time) :-
 1485    !,
 1486    debug(pengine(transition), '~q: 6 = ~q => 3', [ID, ask(Goal)]),
 1487    prolog_cut_to(Choice),
 1488    ask(ID, Goal, Options).
 1489more_solutions(destroy, ID, _Choice, _State, _Time) :-
 1490    !,
 1491    debug(pengine(transition), '~q: 6 = ~q => 1', [ID, destroy]),
 1492    pengine_terminate(ID).
 1493more_solutions(Event, ID, Choice, State, Time) :-
 1494    debug(pengine(transition), '~q: 6 = ~q => 6', [ID, protocol_error(Event)]),
 1495    pengine_reply(error(ID, error(protocol_error, _))),
 1496    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.

 1504ask(ID, Goal, Options) :-
 1505    catch(prepare_goal(ID, Goal, Goal1, Options), Error, true),
 1506    !,
 1507    (   var(Error)
 1508    ->  option(template(Template), Options, Goal),
 1509        option(chunk(N), Options, 1),
 1510        solve(N, Template, Goal1, ID)
 1511    ;   pengine_reply(error(ID, Error)),
 1512        guarded_main_loop(ID)
 1513    ).

 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?

 1527prepare_goal(ID, Goal0, Module:Goal, Options) :-
 1528    option(bindings(Bindings), Options, []),
 1529    b_setval('$variable_names', Bindings),
 1530    (   prepare_goal(Goal0, Goal1, Options)
 1531    ->  true
 1532    ;   Goal1 = Goal0
 1533    ),
 1534    get_pengine_module(ID, Module),
 1535    setup_call_cleanup(
 1536        '$set_source_module'(Old, Module),
 1537        expand_goal(Goal1, Goal),
 1538        '$set_source_module'(_, Old)),
 1539    (   pengine_not_sandboxed(ID)
 1540    ->  true
 1541    ;   get_pengine_application(ID, App),
 1542        setting(App:safe_goal_limit, Limit),
 1543        catch(call_with_time_limit(
 1544                  Limit,
 1545                  safe_goal(Module:Goal)), E, true)
 1546    ->  (   var(E)
 1547        ->  true
 1548        ;   E = time_limit_exceeded
 1549        ->  throw(error(sandbox(time_limit_exceeded, Limit),_))
 1550        ;   throw(E)
 1551        )
 1552    ).

 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.

 1572pengine_not_sandboxed(ID) :-
 1573    pengine_user(ID, User),
 1574    pengine_property(ID, application(App)),
 1575    not_sandboxed(User, App),
 1576    !.

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

 1598pengine_pull_response(Pengine, Options) :-
 1599    pengine_remote(Pengine, Server),
 1600    !,
 1601    remote_pengine_pull_response(Server, Pengine, Options).
 1602pengine_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. */

 1611pengine_input(Prompt, Term) :-
 1612    pengine_self(Self),
 1613    pengine_parent(Parent),
 1614    pengine_reply(Parent, prompt(Self, Prompt)),
 1615    pengine_request(Request),
 1616    (   Request = input(Input)
 1617    ->  Term = Input
 1618    ;   Request == destroy
 1619    ->  abort
 1620    ;   throw(error(protocol_error,_))
 1621    ).

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

*/

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

 1648send_error(error(Formal, context(prolog_stack(Frames), Message))) :-
 1649    is_list(Frames),
 1650    !,
 1651    with_output_to(string(Stack),
 1652                   print_prolog_backtrace(current_output, Frames)),
 1653    pengine_self(Self),
 1654    replace_blobs(Formal, Formal1),
 1655    replace_blobs(Message, Message1),
 1656    pengine_reply(error(Self, error(Formal1,
 1657                                    context(prolog_stack(Stack), Message1)))).
 1658send_error(Error) :-
 1659    pengine_self(Self),
 1660    replace_blobs(Error, Error1),
 1661    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.

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

 1774remote_send_rec(Server, Action, ID, [event=Event], Reply, Options) :-
 1775    !,
 1776    server_url(Server, Action, [id=ID], URL),
 1777    http_open(URL, Stream,              % putting this in setup_call_cleanup/3
 1778              [ post(prolog(Event))     % makes it impossible to interrupt.
 1779              | Options
 1780              ]),
 1781    call_cleanup(
 1782        read_prolog_reply(Stream, Reply),
 1783        close(Stream)).
 1784remote_send_rec(Server, Action, ID, Params, Reply, Options) :-
 1785    server_url(Server, Action, [id=ID|Params], URL),
 1786    http_open(URL, Stream, Options),
 1787    call_cleanup(
 1788        read_prolog_reply(Stream, Reply),
 1789        close(Stream)).
 1790
 1791remote_post_rec(Server, Action, Data, Reply, Options) :-
 1792    server_url(Server, Action, [], URL),
 1793    probe(Action, URL, Options),
 1794    http_open(URL, Stream,
 1795              [ post(json(Data))
 1796              | Options
 1797              ]),
 1798    call_cleanup(
 1799        read_prolog_reply(Stream, Reply),
 1800        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.

 1808probe(create, URL, Options) :-
 1809    !,
 1810    http_open(URL, Stream, [method(options)|Options]),
 1811    close(Stream).
 1812probe(_, _, _).
 1813
 1814read_prolog_reply(In, Reply) :-
 1815    set_stream(In, encoding(utf8)),
 1816    read(In, Reply0),
 1817    rebind_cycles(Reply0, Reply).
 1818
 1819rebind_cycles(@(Reply, Bindings), Reply) :-
 1820    is_list(Bindings),
 1821    !,
 1822    maplist(bind, Bindings).
 1823rebind_cycles(Reply, Reply).
 1824
 1825bind(Var = Value) :-
 1826    Var = Value.
 1827
 1828server_url(Server, Action, Params, URL) :-
 1829    atom_concat('pengine/', Action, PAction),
 1830    uri_edit([ path(PAction),
 1831               search(Params)
 1832             ], Server, URL).

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

 1853pengine_event(Event) :-
 1854    pengine_event(Event, []).
 1855
 1856pengine_event(Event, Options) :-
 1857    thread_self(Self),
 1858    option(listen(Id), Options, _),
 1859    (   thread_get_message(Self, pengine_event(Id, Event), Options)
 1860    ->  true
 1861    ;   Event = timeout
 1862    ),
 1863    update_remote_destroy(Event).
 1864
 1865update_remote_destroy(Event) :-
 1866    destroy_event(Event),
 1867    arg(1, Event, Id),
 1868    pengine_remote(Id, _Server),
 1869    !,
 1870    pengine_unregister_remote(Id).
 1871update_remote_destroy(_).
 1872
 1873destroy_event(destroy(_)).
 1874destroy_event(destroy(_,_)).
 1875destroy_event(create(_,Features)) :-
 1876    memberchk(answer(Answer), Features),
 1877    !,
 1878    nonvar(Answer),
 1879    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]

*/

 1908pengine_event_loop(Closure, Options) :-
 1909    child(_,_),
 1910    !,
 1911    pengine_event(Event),
 1912    (   option(autoforward(all), Options) % TODO: Implement all_but_sender and list of IDs
 1913    ->  forall(child(_,ID), pengine_send(ID, Event))
 1914    ;   true
 1915    ),
 1916    pengine_event_loop(Event, Closure, Options).
 1917pengine_event_loop(_, _).
 1918
 1919:- meta_predicate
 1920    pengine_process_event(+, 1, -, +). 1921
 1922pengine_event_loop(Event, Closure, Options) :-
 1923    pengine_process_event(Event, Closure, Continue, Options),
 1924    (   Continue == true
 1925    ->  pengine_event_loop(Closure, Options)
 1926    ;   true
 1927    ).
 1928
 1929pengine_process_event(create(ID, T), Closure, Continue, Options) :-
 1930    debug(pengine(transition), '~q: 1 = /~q => 2', [ID, create(T)]),
 1931    (   select(answer(First), T, T1)
 1932    ->  ignore(call(Closure, create(ID, T1))),
 1933        pengine_process_event(First, Closure, Continue, Options)
 1934    ;   ignore(call(Closure, create(ID, T))),
 1935        Continue = true
 1936    ).
 1937pengine_process_event(output(ID, Msg), Closure, true, _Options) :-
 1938    debug(pengine(transition), '~q: 3 = /~q => 4', [ID, output(Msg)]),
 1939    ignore(call(Closure, output(ID, Msg))),
 1940    pengine_pull_response(ID, []).
 1941pengine_process_event(debug(ID, Msg), Closure, true, _Options) :-
 1942    debug(pengine(transition), '~q: 3 = /~q => 4', [ID, debug(Msg)]),
 1943    ignore(call(Closure, debug(ID, Msg))),
 1944    pengine_pull_response(ID, []).
 1945pengine_process_event(prompt(ID, Term), Closure, true, _Options) :-
 1946    debug(pengine(transition), '~q: 3 = /~q => 5', [ID, prompt(Term)]),
 1947    ignore(call(Closure, prompt(ID, Term))).
 1948pengine_process_event(success(ID, Sol, _Proj, _Time, More), Closure, true, _) :-
 1949    debug(pengine(transition), '~q: 3 = /~q => 6/2', [ID, success(Sol, More)]),
 1950    ignore(call(Closure, success(ID, Sol, More))).
 1951pengine_process_event(failure(ID, _Time), Closure, true, _Options) :-
 1952    debug(pengine(transition), '~q: 3 = /~q => 2', [ID, failure]),
 1953    ignore(call(Closure, failure(ID))).
 1954pengine_process_event(error(ID, Error), Closure, Continue, _Options) :-
 1955    debug(pengine(transition), '~q: 3 = /~q => 2', [ID, error(Error)]),
 1956    (   call(Closure, error(ID, Error))
 1957    ->  Continue = true
 1958    ;   forall(child(_,Child), pengine_destroy(Child)),
 1959        throw(Error)
 1960    ).
 1961pengine_process_event(stop(ID), Closure, true, _Options) :-
 1962    debug(pengine(transition), '~q: 7 = /~q => 2', [ID, stop]),
 1963    ignore(call(Closure, stop(ID))).
 1964pengine_process_event(destroy(ID, Event), Closure, Continue, Options) :-
 1965    pengine_process_event(Event, Closure, _, Options),
 1966    pengine_process_event(destroy(ID), Closure, Continue, Options).
 1967pengine_process_event(destroy(ID), Closure, true, _Options) :-
 1968    retractall(child(_,ID)),
 1969    debug(pengine(transition), '~q: 1 = /~q => 0', [ID, destroy]),
 1970    ignore(call(Closure, destroy(ID))).