CSP (communicating sequential process) aka channel semantics for Scryer Prolog

[jjtolton]

+This picture from Torbjorn's book really stuck in my head from his talk at the last meetup:

We have most of the ingredients we need now to make Erlang-style actors a reality in Scryer.

1. Async code ✔️ (almost!)
2. Processes ✔️
3. non-blocking I/O 🚧 in progress
4. Mailboxes! ✔️ (see below!)

Torbjorn's book demonstrates the following snippet of code:

receive({
         hello(Name) ->
         format("Hello, ~s!~n", [Name]) ;
         _ ->
         format("Unknown message received.~n")
        }).

We could imagine that being rewritten/expanded or meta-interpreted to be something like:

receive(Chan, Callbacks) :-
        take(Message, Chan),
        freeze(Message, handle_callbacks(Callbacks, Message)).

handle_callbacks([(Head -> Body)|Callbacks], Message) :-
        (   Head=Message -> Body
        ;   handle_callbacks(Callbacks, Message)
        ).
handle_callbacks([], _).

All we would need is some mailbox semantics! CSP is one method I know of to achieve this. There is probably a cleaner way to do it, but I implemented this as a dirty hack on top of queues.

Taking advantage of the fact that queues create "holes" to be filled in later:

:- use_module(library(lambda)).
:- use_module(library(queues)).
?- X+\(queue(Q), queue_head(X,_,Nq), queue_last(1,Q,Nq)).
%@    X = 1.

?- X+\(queue(Q), queue_last(1,Q,Nq), queue_head(X,_,Nq)).
%@    X = 1.

I put together the following semantics:

?- X+\(channel(C), put(a(1), C), take(a(X), C)).
%@    X = 1.

?- X+\(channel(C), take(a(X), C), put(a(1), C)).
%@    X = 1.

which are mostly a dirty hack on top of queues using the blackboard. (we shall see if this turns out to be a memory leak and needs to be implemented differently...)

Why not just use queues? The reason is a bit nuanced but an example will highlight:

%% trying to pop from an empty queue results in failure
?- X+\(queue(Q), queue_last(1,Q,Nq), queue_head(X,Nq0,Nq), queue_head(Y,Nq1,Nq0)).
%@    false.

%% where this differs from vanilla queues is that multiple takes on a channel does
% not cause a failure, it simply "enqueus a take" that will wait for a corresponding "put"
?- X+\(channel(C), put(a(1), C), take(a(X), C), take(Y, C)).
%@    X = 1.

The fact that a take does not fail but waits for a put allows for this pattern:

receive(Chan, Callbacks) :-
        take(Message, Chan),
        freeze(Message, handle_callbacks(Callbacks, Message)).

However, in the meantime, we only have one puzzle piece left to achieve Erlang-style actors and open up a huge world of possibility in terms of possible applications for Scryer (web servers, testing frameworks, parallel computing, distributed computing, infrastructure management).

These things ARE possible in synchronous Prolog, but Erlang semantics were built from the ground up for distributed computing, and it would be great to bring those back home to Prolog!

CSP Draft Implementation

:- use_module(library(assoc)).
:- use_module(library(iso_ext)).
:- use_module(library(uuid)).
:- use_module(library(queues)).

/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
   Public API
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - */
channel(Id) :-
        channel_(Id, Chan),
        bb_get('async/channels', Channels),
        put_assoc(Id, Channels, Chan, Channels1),
        bb_put('async/channels', Channels1).


put(Value, ChanId) :-
        bb_get('async/channels', Channels),
        get_assoc(ChanId, Channels, Chan),
        put_(Value,Chan,Chan1),
        put_assoc(ChanId,Channels,Chan1,Channels1),
        bb_b_put('async/channels', Channels1).

take(Value, ChanId) :-
        bb_get('async/channels',Channels),
        get_assoc(ChanId,Channels,CurChan),
        take_(Value,CurChan,Chan1),
        put_assoc(ChanId,Channels,Chan1,Channels1),
        bb_b_put('async/channels', Channels1).

%% must be used as a directive to make channels available!
init_channels.

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

term_expansion((:- init_channels), [channels_initialized]) :-
  (   \+ bb_get('async/channels', _)
  ->  empty_assoc(A), bb_put('async/channels', A)
  ).


channel_(Id, channel(Id, puts(0,Puts), takes(0,Takes))) :-
        uuidv4_string(Id),
        queue(Puts),
        queue(Takes).


put_(Value, channel(Id, puts(Nputs,Puts), takes(Ntakes,Takes)), C2) :-
        put_(Ntakes,Nputs,Value,Puts,Takes, Id, C2).

put_(0,Nputs, Value,Puts,Takes, Id, channel(Id, puts(s(Nputs),Puts1),takes(0,Takes))) :-
        enqueue(Value,Puts,Puts1).

put_(s(Ntakes),Nputs,Value,Puts,Takes, Id, channel(Id, puts(Nputs,Puts),takes(Ntakes,Takes0))) :-
        dequeue(Take,Takes,Takes0),
        Value=Take.


take_(Value, channel(Id, puts(Nputs,Puts), takes(Ntakes,Takes)), C2) :-
        take_(Nputs, Ntakes, Value, Puts, Takes, Id, C2).

take_(0,Ntakes,Value,Puts,Takes,Id,channel(Id,puts(0,Puts),takes(s(Ntakes),Takes1))) :-
        enqueue(Value,Takes,Takes1).

take_(s(Nputs),Ntakes,Value,Puts,Takes,Id,channel(Id,puts(Nputs,Puts0),takes(Ntakes,Takes))) :-
        dequeue(Put,Puts,Puts0),
        Value=Put.

/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
   Internal Queue implementation
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - */

enqueue(X,Q,Nq) :-
        queue_last(X,Q,Nq).

dequeue(X,Q,Nq) :-
        queue_head(X,Nq,Q).

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

:- init_channels.

:- use_module(library(lambda)).
% ?- X+\(queue(Q), queue_head(X,_,Nq), queue_last(1,Q,Nq)).
%@    X = 1.

% ?- X+\(queue(Q), queue_last(1,Q,Nq), queue_head(X,_,Nq)).
%@    X = 1.

% ?- X+\(channel(C), put(a(1), C), take(a(X), C)).
%@    X = 1.

% ?- X+\(channel(C), take(a(X), C), put(a(1), C)).
%@    X = 1.


/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
   Possible Usage
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -

receive({
   hello(Name) ->
   format("Hello, ~s!~n", [Name]) ;
   _ ->
   format("Unknown message received.~n")
   }).


receive(Chan, Callbacks) :-
        take(Message, Chan),
        freeze(Message, handle_callbacks(Methods, Message)).

handle_callbacks([(Head -> Body)|Callbacks], Message) :-
        (   Head=Message -> Body
        ;   handle_callbacks(Callbacks, Message)
        ).
handle_callbacks([], _).

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

[jjtolton]

Actors will need some sort of "Actor System" container, so rather than store these in the blackboard, it might make more sense to manage these with the actor system. That way they could be

channels(Channels) :- empty_assoc(Channels).
processes(Processes) :- empty_assoc(Processes). % erlang-style processes/actors, not library(process) processes

:- meta_predicate(actor_system(2)).
actor_system(Goal) :- 
    channels(Channels),
    processes(Processes),
    async(actor_system_(Channels, Processes, Goal)),
    run_event_loop.

:- meta_predicate(actor_system_(+,+,2)).    
actor_system_(Channels, Processes, Goal) :- 
    yield(call(Goal, Channels, Processes)),
    yield(actor_system_(Channels, Processes, Goal)).

This would be better life-cycle management of the mailboxes than in the blackboard. 🤔