The "lists of (meta-)goals" problem

[jjtolton]

I have run across this problem a few different times.

Awhile ago I had a failed attempt to implement to implement cond_t/2.

Recently I've been struggling with the test_framework module having (seemingly) different behavior than user code.

Most recently, I've been struggling with the library(async) PR.

I've had to resort to this:

:- meta_predicate(async_event_loop(0)).
:- meta_predicate(async_event_loop(0,0)).
:- meta_predicate(async_event_loop(0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0,0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0,0,0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0,0,0,0,0,0,0,0,0)).
:- meta_predicate(async_event_loop(0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0)).

async_event_loop(A) :-
    list_queue([A], Q),
    run_task_q(Q).

async_event_loop(A,B) :-
    list_queue([A,B], Q),
    run_task_q(Q).

async_event_loop(A,B,C) :-
    list_queue([A,B,C], Q),
    run_task_q(Q).

...

This seems to work for user code (still investigating) but does not work for the test framework code!

Meaning:

aggressive_target(X) :-
        %% inside an internal reset, you are now in a syncronous context until you shift/1 back
        shift(X),
        shift(X),
        await(true).

aggressive_jump(X) :-
        reset(aggressive_target(X), X, cont(Cont)),
        reset(Cont,X,cont(C1)),
        C1.

aggressive_loop(N,M,A,B) :-
        async_event_loop(async_event_loop(aggressive_jump(42)),
                         aggressive_jump(42),
                         async_event_loop(async_event_loop(aggressive_jump(N),
                                                           aggressive_jump(M)),
                                                           async_event_loop(aggressive_jump(A),
                                                                            aggressive_jump(B)))
                         ).

?- (
    aggressive_loop(N,M,A,B),
    N=1,
    M=2,
    A=3,
    B=4,
    42 #= N+M+A+B+32
   ).
%@    N = 1, M = 2, A = 3, B = 4.

works but

aggressive_target(X) :-
        %% inside an internal reset, you are now in a syncronous context until you shift/1 back
        shift(X),
        shift(X),
        await(true).

aggressive_jump(X) :-
        reset(async_tests:aggressive_target(X), X, cont(Cont)),
        reset(Cont,X,cont(C1)),
        C1.


test("nested async loops -- this is madness, but just for stress testing",
     (   async_event_loop(async_event_loop(async_tests:aggressive_jump(42)),
                          aggressive_jump(42),
                          async_event_loop(async_event_loop(aggressive_jump(N),
                                                            async_tests:aggressive_jump(M)),
                                           async_event_loop(async_tests:aggressive_jump(A),
                                                            async_tests:aggressive_jump(B)))
                         ),
         N=1,M=2,A=3,B=4,
         42 #= N+M+A+B+32
     )
    ).



% ?- test_framework:main(async_tests).
%@ Running test "nested async loops -- this is madness, but just for stress testing"
%@    error(existence_error(procedure,aggressive_jump/1),aggressive_jump/1).

but it does work if I fully supply the modules:

test("nested async loops -- this is madness, but just for stress testing",
     (   async_event_loop(async_event_loop(async_tests:aggressive_jump(42)),
                          async_tests:aggressive_jump(42),
                          async_event_loop(async_event_loop(async_tests:aggressive_jump(N),
                                                            async_tests:aggressive_jump(M)),
                                           async_event_loop(async_tests:aggressive_jump(A),
                                                            async_tests:aggressive_jump(B)))
                         ),
         N=1,M=2,A=3,B=4,
         42 #= N+M+A+B+32
     )
    ).

% ?- test_framework:main(async_tests).
%@ Running test "nested async loops -- this is madness, but just for stress testing"

And I believe this has to do with the "list of goals" problem:

% module(test_framework)

main(TestModule) :-
    findall(test(Name, TestModule:Goal), TestModule:test(Name, Goal), Tests),
    run_tests(Tests, Failed),
    show_failed(Failed),
    halt.

Note that Tests is a "list of goals" here.

This is not to say that there is anything wrong with the testing framework, per se, it's more pointing out that the "list of goals" problem is a really tricky problem.

I have tried a LOT of different things to attempt to deal with this problem but I think what I am realizing is maybe it's just not a very good API in Prolog?

Is that the consensus? Does anyone have any advice on how to deal with this?

To be clear, the original definition of async_event_loop/1 was

async_event_loop([X|Xs]) :- 
    list_queue([X|Xs], Q),
    run_task_q(Q).

Very clean, but we lose the module data necessary to call the metagoals later.

[jjtolton]

This response from @UWN is probably the answer.

[triska]

Can you maybe queue the goals individually, i.e., an API for adding a single goal to an existing queue?

The argument for the single goal would have specifier 0.

[triska]

Better call it list difference. It's not a special kind of list (such as: "blabla...list"), rather it's a difference of lists that describes the tasks here: a partial list "minus" the variable tail.

[jjtolton]

Better call it list difference. It's not a special kind of list (such as: "blabla...list"), rather it's a difference of lists that describes the tasks here: a partial list "minus" the variable tail.

Ah, ok -- sorry, did not have time to correct the terminology. But check this out -- it works!!

I'm still trying to wrap my head around it, I'm don't even completely understand how it works (list differences are quite new to me), but amazingly:

:- use_module(library(reif)).
:- use_module(library(cont)).
:- use_module(library(debug)).
:- use_module(library(lambda)).
:- use_module(library(lists)).
:- use_module(library(iso_ext)).

user:term_expansion((:- init_queue), [queue_init]) :-
    bb_put('task/queue', _).

:- init_queue.


run_event_loop :-
        bb_get('task/queue', Q),
        dlist_length(Q,N),
        if_(N=0,
            true,
            run_task_q(Q)
           ).

run_task_q(Dlist) :-
        taskqcall_deque(Dlist,Ndlist),
        bb_put('task/queue', Ndlist),
        run_event_loop.

run_task_q(A-A) :- var(A).



:- meta_predicate(async_(0,+,-)).
async_(Task, Q0, Q) :-
        bb_get('task/queue', Q0),
        (   var(Q0) -> dlist(Task, Q)
        ;   item_dlist_new(Task, Q0, Q)
        ),
        bb_put('task/queue', Q)
        .

:- meta_predicate(async(0)).
async(Task) :-
        async_(Task,_Q0,_Q).


:- meta_predicate(yield(0)).
yield(Q) :- shift(Q).


dlist(X, [X|Xs]-Xs).

item_dlist_new(X, Xs-[X|Ys], Xs-Ys).

item_dlist_dequeue(Elem, Front-Back, NewFront-Back) :-
        nonvar(Front),
        Front = [Elem|Rest],
        NewFront = Rest.

dlist_length(Front-Back,Length) :-
        copy_term(Front-Back,A-B),
        B=[],
        length(A,Length).

taskqcall_deque(Q, Nq) :-
        item_dlist_dequeue(T,Q,Nq0),
        bb_put('task/queue', Nq0),
        reset(T, T1, Cont),
        bb_get('task/queue', Q0),
        resume_internal(Cont,T1,Q0,Nq).

:- meta_predicate(resume_internal(+,0,+,-)).
resume_internal(none,_Meaningless,Q,Q).
resume_internal(cont(Cont),Task,Q,Nq) :-
        item_dlist_new(Task,Q,Nq0),
        item_dlist_new(Cont,Nq0,Nq).

t(X) :- atom_concat(t, X, T), write(T), nl.

foo2 :-
        async(t(a)),
        yield(t(b)),
        async(t(c)).

bar2 :-
        async(t(d)),
        yield(t(e)),
        async(t(f)).

?- async(foo2), async(bar2), run_event_loop.
%@ ta
%@ tb
%@ td
%@ te
%@ tc
%@ tf
%@    true
%@ ;  false.

Except for one redundant choice point, this is perfect.

I don't want to get too far ahead of myself, I still need to write up the tests to make sure that I'm not going to have more meta-predicate problems, but this is way more powerful than the previous version!

[jjtolton]

The following changes eliminate the redundant choicepoint:

run_event_loop :-
        bb_get('task/queue', Q),
        run_task_q(Q).

run_task_q(Dlist) :-
        dlist_length(Dlist, N),
        if_(N=0,
            true,
            run_task_q_(Dlist)
           ).

run_task_q_(Dlist) :-
        taskqcall_deque(Dlist,Ndlist),
        bb_put('task/queue', Ndlist),
        run_event_loop.

%% ...snip...

?- async(foo2), async(bar2), run_event_loop.
%@ ta
%@ tb
%@ td
%@ te
%@ tc
%@ tf
%@    true.

Can you believe that some Prolog systems do not have reif? 😰

[triska]

copy_term(Front-Back,A-[])

[jjtolton]

It simplifies so nicely. And the library version will allow for running sync or async depending on the directive you use.

Latest:

:- use_module(library(reif)).
:- use_module(library(cont)).
:- use_module(library(debug)).
:- use_module(library(lambda)).
:- use_module(library(lists)).
:- use_module(library(iso_ext)).

user:term_expansion((:- init_queue), [queue_init]) :-
    % todo: make async/yield syncronous unless
    % init_queue (or whatever it will be called)
    % is added as a directive, then change
    % install term expansions which change the public definitions
    % to the async counterparts
    bb_put('task/queue', _).

:- init_queue.

/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
   Public API
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - */

:- meta_predicate(async(0)).
async(Task) :-
        async_(Task,_Q0,_Q).


:- meta_predicate(yield(0)).
yield(Q) :- shift(Q).


run_event_loop :-
        bb_get('task/queue', Q),
        run_task_q(Q).


/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
   Internal 
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - */

run_task_q(Dlist) :-
        dlist_length(Dlist, N),
        if_(N=0,
            true,
            run_task_q_(Dlist)
           ).

run_task_q_(Q) :-
        item_dlist_dequeue(T,Q,Nq),
        % the following reset may add more tasks to the task queue,
        % so we need to recache here
        bb_put('task/queue', Nq),
        reset(T, T1, Cont),
        resume_internal(Cont,T1), % add new tasks if needed
        run_event_loop.

:- meta_predicate(resume_internal(+,0,+,-)).
resume_internal(none,_Meaningless).
resume_internal(cont(Cont),Task) :-
        async(Task),
        async(Cont).

:- meta_predicate(async_(0,+,-)).
async_(Task, Q0, Q) :-
        bb_get('task/queue', Q0),
        (   var(Q0) -> dlist(Task, Q)
        ;   item_dlist_new(Task, Q0, Q)
        ),
        bb_put('task/queue', Q).

/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
   Internal List Different Utilities
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - */

dlist(X, [X|Xs]-Xs).

item_dlist_new(X, Xs-[X|Ys], Xs-Ys).

item_dlist_dequeue(Elem, Front-Back, NewFront-Back) :-
        nonvar(Front),
        Front = [Elem|Rest],
        NewFront = Rest.

dlist_length(Front-Back,Length) :-
        copy_term(Front-Back,A-[]),
        length(A,Length).

/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
   Examples
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - */

t(X) :- atom_concat(t, X, T), write(T), nl.

foo :-
        async(t(a)),
        yield(t(b)),
        async(t(c)).

bar :-
        async(t(d)),
        yield(t(e)),
        async(t(f)).

baz :- t(g).

?- async(foo), async(bar), baz, async(yield(baz)), run_event_loop.
%@ tg
%@ ta
%@ tb
%@ td
%@ te
%@ tg
%@ tc
%@ tf
%@    true.

[UWN]

Just a guess: Sometimes you use explicit qualification for agressive_jump(42) and sometimes you don't. Seems one such unqualified version got called. The module system has this odd default behaviour of taking the current module in case there is no qualification. In many situations that is undesirable. It could make sense (in some situations, but not all of them) to refuse this default and produce an error in its stead. Ideally all of this could be hidden in a (very rudimentary) type system that just knows terms, and goals with 0, 1 ... arguments lacking.

[jjtolton]

My understanding of what is happening is that the test_framework module gathers all of the tests and puts them in a list, prefixing them with the module, so while the test is written as:

test("nested async loops -- this is madness, but just for stress testing",
     (   async_event_loop(async_event_loop(async_tests:aggressive_jump(42)),
                          async_tests:aggressive_jump(42),
                          async_event_loop(async_event_loop(async_tests:aggressive_jump(N),
                                                            async_tests:aggressive_jump(M)),
                                           async_event_loop(async_tests:aggressive_jump(A),
                                                            async_tests:aggressive_jump(B)))
                         ),
         N=1,M=2,A=3,B=4,
         42 #= N+M+A+B+32
     )
    ).

the goal which is executed is

     async_tests:(   async_event_loop(async_event_loop(async_tests:aggressive_jump(42)),
                          async_tests:aggressive_jump(42),
                          async_event_loop(async_event_loop(async_tests:aggressive_jump(N),
                                                            async_tests:aggressive_jump(M)),
                                           async_event_loop(async_tests:aggressive_jump(A),
                                                            async_tests:aggressive_jump(B)))
                         ),
         N=1,M=2,A=3,B=4,
         42 #= N+M+A+B+32
     )

because of this line.

And so what is happening is that these of test/2 predicates get despecialized to test/2 functors by the findall/3. The module name is attached to the outside of the goal but it does not propagate inwards, and as you observed I hit the unlabeled aggressive_jump/1.

The test_framework makes efforts to reattach the module information to the outer goal, but the cross-linking of inner goals gets lost.

Somehow, using metapredicate declarations instead of lists, you are able to preserve the correct information.

I have tried things like,

list_of_goals(Xs, Ys) :- annotate(Xs), <...> .

annotate([X|Xs]) :- annotate_(X), annotate(Xs).
annotate(Xs).

:- meta_predicate(annotate_(0)).
annotate_(X).

but these hacks did not work 😅

[jjtolton]

The more I learn the more I am astonished by reif and lambda, btw.

[hurufu]

Just throwing an idea, instead of "adding" tasks which is very procedural, is it possible to do it more declaratively?

For example mark predicates with :- task(PredicateIndicator). and collect all of them as tasks, and automatically solving dependencies between them...

[jjtolton]

It's certainly worth looking into. The only caveat is that the only sort of code that this async work would be useful for is procedural work, presumably I/O. If there are no side effects such as write/1, the semantics of an application using async or not would be logically identical.