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-2024, 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(sandbox),[safe_goal/1]).   91:- autoload(library(statistics),[thread_statistics/2]).   92:- autoload(library(term_to_json),[term_to_json/2]).   93:- autoload(library(thread_pool),
   94	    [thread_pool_create/3,thread_create_in_pool/4]).   95:- autoload(library(time),[alarm/4,call_with_time_limit/2]).   96:- autoload(library(uri),
   97	    [ uri_components/2,
   98	      uri_query_components/2,
   99	      uri_data/3,
  100	      uri_data/4,
  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:- autoload(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
  116:- if(exists_source(library(uuid))).  117:- autoload(library(uuid), [uuid/2]).  118:- endif.  119
  120
  121:- meta_predicate
  122    pengine_create(:),
  123    pengine_rpc(+, +, :),
  124    pengine_event_loop(1, +).  125
  126:- multifile
  127    write_result/3,                 % +Format, +Event, +Dict
  128    event_to_json/3,                % +Event, -JSON, +Format
  129    prepare_module/3,               % +Module, +Application, +Options
  130    prepare_goal/3,                 % +GoalIn, -GoalOut, +Options
  131    authentication_hook/3,          % +Request, +Application, -User
  132    not_sandboxed/2,                % +User, +App
  133    pengine_flush_output_hook/0.  134
  135:- predicate_options(pengine_create/1, 1,
  136                     [ id(-atom),
  137                       alias(atom),
  138                       application(atom),
  139                       destroy(boolean),
  140                       server(atom),
  141                       ask(compound),
  142                       template(compound),
  143                       chunk(integer;oneof([false])),
  144                       bindings(list),
  145                       src_list(list),
  146                       src_text(any),           % text
  147                       src_url(atom),
  148                       src_predicates(list)
  149                     ]).  150:- predicate_options(pengine_ask/3, 3,
  151                     [ template(any),
  152                       chunk(integer;oneof([false])),
  153                       bindings(list)
  154                     ]).  155:- predicate_options(pengine_next/2, 2,
  156                     [ chunk(integer),
  157                       pass_to(pengine_send/3, 3)
  158                     ]).  159:- predicate_options(pengine_stop/2, 2,
  160                     [ pass_to(pengine_send/3, 3)
  161                     ]).  162:- predicate_options(pengine_respond/3, 2,
  163                     [ pass_to(pengine_send/3, 3)
  164                     ]).  165:- predicate_options(pengine_rpc/3, 3,
  166                     [ chunk(integer;oneof([false])),
  167                       pass_to(pengine_create/1, 1)
  168                     ]).  169:- predicate_options(pengine_send/3, 3,
  170                     [ delay(number)
  171                     ]).  172:- predicate_options(pengine_event/2, 2,
  173                     [ listen(atom),
  174                       pass_to(system:thread_get_message/3, 3)
  175                     ]).  176:- predicate_options(pengine_pull_response/2, 2,
  177                     [ pass_to(http_open/3, 3)
  178                     ]).  179:- predicate_options(pengine_event_loop/2, 2,
  180                     []).                       % not yet implemented
  181
  182% :- debug(pengine(transition)).
  183:- debug(pengine(debug)).               % handle pengine_debug in pengine_rpc/3.
  184
  185goal_expansion(random_delay, Expanded) :-
  186    (   debugging(pengine(delay))
  187    ->  Expanded = do_random_delay
  188    ;   Expanded = true
  189    ).
  190
  191do_random_delay :-
  192    Delay is random(20)/1000,
  193    sleep(Delay).
  194
  195:- meta_predicate                       % internal meta predicates
  196    solve(+, ?, 0, +),
  197    findnsols_no_empty(+, ?, 0, -),
  198    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. */

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

 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.

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

 pengine_send(+NameOrID, +Term) is det

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

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

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

 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.

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

 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.

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

*/

  516pengine_ask(ID, Query, Options) :-
  517    partition(pengine_ask_option, Options, AskOptions, SendOptions),
  518    pengine_send(ID, ask(Query, AskOptions), SendOptions).
  519
  520
  521pengine_ask_option(template(_)).
  522pengine_ask_option(chunk(_)).
  523pengine_ask_option(bindings(_)).
  524pengine_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).

*/

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

*/

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

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

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

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

 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)

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

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

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

 pengine_self(-Id) is det

True if the current thread is a pengine with Id.

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

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

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

*/

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

 current_pengine_application(?Application) is nondet

True when Application is a currently defined application.

See also

- pengine_application/1

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

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

 pengine_output(+Term) is det

Sends Term to the parent pengine or thread. */

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

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

 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.

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

 thread_pool:create_pool(+Application) is det

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

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

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

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

 pengine_main(+Parent, +Options, +Application)

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

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

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

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

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

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

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

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

 1400set_projection(Goal, Bindings) :-
 1401    b_setval('$variable_names', Bindings),
 1402    call(Goal).
 1403
 1404projection(Projection) :-
 1405    nb_current('$variable_names', Bindings),
 1406    !,
 1407    maplist(var_name, Bindings, Projection).
 1408projection([]).

 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.

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

 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.

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

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

 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?

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

 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.

 1568pengine_not_sandboxed(ID) :-
 1569    pengine_user(ID, User),
 1570    pengine_property(ID, application(App)),
 1571    not_sandboxed(User, App),
 1572    !.

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

 1594pengine_pull_response(Pengine, Options) :-
 1595    pengine_remote(Pengine, Server),
 1596    !,
 1597    remote_pengine_pull_response(Server, Pengine, Options).
 1598pengine_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. */

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

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

*/

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

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

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

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

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

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

*/

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