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-2023, 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/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
  134:- predicate_options(pengine_create/1, 1,
  135                     [ id(-atom),
  136                       alias(atom),
  137                       application(atom),
  138                       destroy(boolean),
  139                       server(atom),
  140                       ask(compound),
  141                       template(compound),
  142                       chunk(integer;oneof([false])),
  143                       bindings(list),
  144                       src_list(list),
  145                       src_text(any),           % text
  146                       src_url(atom),
  147                       src_predicates(list)
  148                     ]).  149:- predicate_options(pengine_ask/3, 3,
  150                     [ template(any),
  151                       chunk(integer;oneof([false])),
  152                       bindings(list)
  153                     ]).  154:- predicate_options(pengine_next/2, 2,
  155                     [ chunk(integer),
  156                       pass_to(pengine_send/3, 3)
  157                     ]).  158:- predicate_options(pengine_stop/2, 2,
  159                     [ pass_to(pengine_send/3, 3)
  160                     ]).  161:- predicate_options(pengine_respond/3, 2,
  162                     [ pass_to(pengine_send/3, 3)
  163                     ]).  164:- predicate_options(pengine_rpc/3, 3,
  165                     [ chunk(integer;oneof([false])),
  166                       pass_to(pengine_create/1, 1)
  167                     ]).  168:- predicate_options(pengine_send/3, 3,
  169                     [ delay(number)
  170                     ]).  171:- predicate_options(pengine_event/2, 2,
  172                     [ listen(atom),
  173                       pass_to(system:thread_get_message/3, 3)
  174                     ]).  175:- predicate_options(pengine_pull_response/2, 2,
  176                     [ pass_to(http_open/3, 3)
  177                     ]).  178:- predicate_options(pengine_event_loop/2, 2,
  179                     []).                       % not yet implemented
  180
  181% :- debug(pengine(transition)).
  182:- debug(pengine(debug)).               % handle pengine_debug in pengine_rpc/3.
  183
  184goal_expansion(random_delay, Expanded) :-
  185    (   debugging(pengine(delay))
  186    ->  Expanded = do_random_delay
  187    ;   Expanded = true
  188    ).
  189
  190do_random_delay :-
  191    Delay is random(20)/1000,
  192    sleep(Delay).
  193
  194:- meta_predicate                       % internal meta predicates
  195    solve(+, ?, 0, +),
  196    findnsols_no_empty(+, ?, 0, -),
  197    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. */

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

 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.

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

 pengine_send(+NameOrID, +Term) is det

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

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

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

 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.

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

 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.

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

*/

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

*/

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

*/

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

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

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

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

 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)

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

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

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

 pengine_self(-Id) is det

True if the current thread is a pengine with Id.

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

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

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

*/

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

 current_pengine_application(?Application) is nondet

True when Application is a currently defined application.

See also

- pengine_application/1

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

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

 pengine_output(+Term) is det

Sends Term to the parent pengine or thread. */

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

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

 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.

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

 thread_pool:create_pool(+Application) is det

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

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

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

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

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

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

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

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

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

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

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

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

 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.

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

 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.

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

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

 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?

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

 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.

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

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

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

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

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

*/

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

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

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

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

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

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

*/

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