Cooperative Concurrency (and Parallelism!)

[jjtolton]

(Tested on v0.9.4-548-g97a7e37b)

I have been dabbling with ways to work towards making Torbjorn's Prolog Actors a reality in Scryer, and I stumbled across an interesting technique for cooperative concurrency in Prolog.

(Note: this work is somewhat superseded by Part 2)

I am certainly not the first person to figure this out (big shout out to @bakaq for many long conversations on this), but I was actually surprised how elegant it was to implement in Prolog when I finally figured it out!

I could use some help with a few parts, particularly the queueing, if anyone has any ideas.

The bulk of it can be expressed in a few lines:

run_task_q(Q) :-
        taskqcall_deque(Q, Nq),
        run_task_q(Nq).
run_task_q([]).

taskqcall_deque(Q, Nq) :-
        q_deque_last(Q,Q0,T),
        call(T,Q0,Nq).

I could not figure out how to use library(queue) (the behavior of queue_last/3 was very strange), so I made my own constructs:

item_q_enque(X, [], [X]).
item_q_enque(X, [Y|Ys], [X,Y|Ys]).

q_deque_last(Xs, Ys, Item) :-
        reverse(Xs,[Item|Sy]), % this is really inefficient, help wanted
        reverse(Sy,Ys).
        
que_task_enque(Q,T,Nq) :-
        item_q_enque(T,Q,Nq).

Here are some example concurrent tasks:

write_task(Message,Q,Q) :-
        write(Message),
        nl.

writem_task([Message|Messages],Q,Nq) :-
        que_task_enque(Q,write_task(Message),Nq0),
        que_task_enque(Nq0,writem_task(Messages),Nq).
writem_task([],Q,Q).

countable_task(N,Max,Q,Nq) :-
        if_(clpz_t(N #>= Max),
            Q=Nq,
            (   que_task_enque(Q,write_task(N),Nq0),
                N1#=N+1,
                que_task_enque(Nq0,countable_task(N1,Max),Nq)
            )
           ).

They can be used like so:

?- run_task_q([writem_task([cruel]),writem_task([hello,world])]).
   %@ hello
   %@ cruel
   %@ world
   %@    true.

   ?- run_task_q([countable_task(0,5), writem_task([a,b,c]), writem_task([x,y,z])]).
   %@ x
   %@ a
   %@ 0
   %@ y
   %@ b
   %@ 1
   %@ z
   %@ c
   %@ 2
   %@ 3
   %@ 4
   %@    true.

Most excitingly, thanks to @Skgland's contribution with library(process), we can combine cooperative concurrency with processes to achieve ad-hoc parallelism in Scryer!

process_task(proc(Name, Args, Opts), Q, Nq) :-
        process_create(Name,Args,Opts),
        memberchk(process(Proc), Opts),
        que_task_enque(Q,process_task(wait(Proc,Opts)),Nq).
        
process_task(wait(Proc,Opts),Q,Nq) :-
        process_wait(Proc, Status, [timeout(0)]),
        if_(Status=exit(0),
            (   memberchk(stdout(pipe(Stream)), Opts),
                get_n_chars(Stream,_,Chars),
                que_task_enque(Q,format_task("~s", [Chars]),Nq)
            ),
            que_task_enque(Q,process_task(wait(Proc,Opts)),Nq)
        ).
        
random_sleep_write_task(Proc) :-
        Program="N=3; rand=$(( (RANDOM % N) + 1 )); sleep $rand; echo $rand",
        Proc=process_task(proc("bash", ["-c", Program], [stdout(pipe(Stream)), process(P)])).

   ?- time(_+\once(( maplist(random_sleep_write_task, Tasks),
           length(Tasks,16),
           run_task_q(Tasks)
         ))).
   %@ 1
   %@ 1
   %@ 3
   %@ 4
   %@ 4
   %@ 4
   %@ 5
   %@ 5
   %@ 6
   %@ 7
   %@ 7
   %@ 9
   %@ 9
   %@ 9
   %@ 10
   %@ 10
   %@    % CPU time: 10.001s, 72_832_865 inferences
   %@    true.

I'm sure there are plenty of mistakes in here, but I was excited with the progress and I think there is room to push forward.

For instance, the fact that I have to use once is an indication there is a serious problem. If I do not use once, the query does not terminate after the first leaf answer.

Edit: Oh, interesting. Putting length/2 after maplist/2 to constrain Tasks does not seem to work.

   ?- length(Xs, 1), maplist(=, Xs, Xs).
   %@    Xs = [_A].
   ?- maplist(=, Xs, Xs), length(Xs, 1).
   %@    Xs = [_A]
   %@ ;  error('$interrupt_thrown',repl/0).

That caught me by surprise!

This then works as expected:

      ?- time(_+\(length(Tasks,16),
               maplist(random_sleep_write_task, Tasks),
               run_task_q(Tasks)
              )
          ).
   %@ 1
   %@ 1
   %@ 1
   %@ 1
   %@ 1
   %@ 2
   %@ 2
   %@ 2
   %@ 2
   %@ 2
   %@ 2
   %@ 2
   %@ 3
   %@ 3
   %@ 3
   %@ 3
   %@    % CPU time: 3.009s, 16_995_447 inferences
   %@    true.

[jjtolton]

The Q,Nq pattern on the tasks looks like some other patterns in Prolog, such as DCGs, which makes me think there is probably a meta-interpreter solution that could make this approach much more elegant.

[jjtolton]

Full file:

:- use_module(library(lists)).
:- use_module(library(debug)).
:- use_module(library(time)).
:- use_module(library(format)).
:- use_module(library(charsio)).
:- use_module(library(process)).
:- use_module(library(clpz)).
:- use_module(library(reif)).
:- use_module(library(lambda)).

run_task_q(Q) :-
        taskqcall_deque(Q, Nq),
        run_task_q(Nq).
run_task_q([]).

taskqcall_deque(Q, Nq) :-
        q_deque_last(Q,Q0,T),
        call(T,Q0,Nq).


item_q_enque(X, [], [X]).
item_q_enque(X, [Y|Ys], [X,Y|Ys]).

q_deque_last(Xs, Ys, Item) :-
        reverse(Xs,[Item|Sy]),
        reverse(Sy,Ys).
        
que_task_enque(Q,T,Nq) :-
        item_q_enque(T,Q,Nq).

write_task(Message,Q,Q) :-
        write(Message),
        nl.

format_task(Format,Message,Q,Q) :-
        format(Format,Message).

writem_task([Message|Messages],Q,Nq) :-
        que_task_enque(Q,write_task(Message),Nq0),
        que_task_enque(Nq0,writem_task(Messages),Nq).
writem_task([],Q,Q).

countable_task(N,Max,Q,Nq) :-
        if_(clpz_t(N #>= Max),
            Q=Nq,
            (   que_task_enque(Q,write_task(N),Nq0),
                N1#=N+1,
                que_task_enque(Nq0,countable_task(N1,Max),Nq)
            )
           ).


wait(S) :-
        phrase(format_("sleep ~q", [S]), Sleep),
        process_create("bash", ["-c", Sleep], [process(P)]),
        process_wait(P, exit(0)).

process_task(proc(Name, Args, Opts), Q, Nq) :-
        process_create(Name,Args,Opts),
        memberchk(process(Proc), Opts),
        que_task_enque(Q,process_task(wait(Proc,Opts)),Nq).
        
process_task(wait(Proc,Opts),Q,Nq) :-
        process_wait(Proc, Status, [timeout(0)]),
        if_(Status=exit(0),
            (   memberchk(stdout(pipe(Stream)), Opts),
                get_n_chars(Stream,_,Chars),
                que_task_enque(Q,format_task("~s", [Chars]),Nq)
            ),
            (   wait(0.01),
                que_task_enque(Q,process_task(wait(Proc,Opts)),Nq)
            )
        ).
        
random_sleep_write_task(Proc) :-
        Program="N=3; rand=$(( (RANDOM % N) + 1 )); sleep $rand; echo $rand",
        Proc=process_task(proc("bash", ["-c", Program], [stdout(pipe(_Stream)), process(_P)])).


/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
   ?- Nq+\( 
   que_task_enque([], write_task(hello), Nq0),
   que_task_enque(Nq0, write_task(cruel), Nq1),
   que_task_enque(Nq1, write_task(world), Nq),
   run_task_q(Nq)
   ).

   
   ?- run_task_q([countable_task(0,5), writem_task([a,b,c]), writem_task([x,y,z])]).
   
   ?- time(_+\(length(Tasks,16),
               maplist(random_sleep_write_task, Tasks),
               run_task_q(Tasks)
              )
          ).

- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - */