sketch of Async/Await semantics (Concurrency/Parallelism: Part 2) #3034
Replies: 5 comments 13 replies
-
|
As an aside, I tried an implementation of queues using |
Beta Was this translation helpful? Give feedback.
-
|
Awesome, thank you a lot for working on this! Regarding these points: So you really want to be invoking it as reset(Goal, Val, cont(Cont)) rather than reset(Goal, Val, Cont)
So, we cannot in general assume that ?- reset(true, B, Cont). Cont = none. The good news is that the representation of possible kinds of results is clean (as opposed to defaulty), and we can therefore for instance write: process_cont(none). process_cont(cont(C)) :- C. This was the motivation behind the API used in Scryer, also encouraged by @samer-- when the API of the original library was discussed. Please also see #136. One other point raised by @samer-- back then is that |
Beta Was this translation helpful? Give feedback.
-
|
Yes. I agree with Samer, who I think at one point also mentioned better name proposals by other specialists in this area. |
Beta Was this translation helpful? Give feedback.
-
|
These are great points, @triska! This obviates the need for the One problem with this approach that concerns me, if I were to turn this into an async library, is would this usage of I cannot think of an issue where it causes interference, but it is hard to think through all the possibilities. |
Beta Was this translation helpful? Give feedback.
-
|
Excellent point! Maybe we need not only better names but even a better interface, similar to exceptions which indicate what they are. |
Beta Was this translation helpful? Give feedback.
-
|
Do you mean something like Oh interesting, this already works -- how extremely clever! a :-
write(a_start), nl,
reset(b, here(Val), Cont),
write(a_end), nl,
Cont=cont(C),
C.
b :-
write(b_start),nl,
shift(not_here),
write(b_end),nl.
?- a.
%@ a_start
%@ b_start
%@ false.vs a :-
write(a_start), nl,
reset(b, here(Val), Cont),
write(a_end), nl,
Cont=cont(C),
C.
b :-
write(b_start),nl,
shift(here(i_am)), % unify with here/1 above
write(b_end),nl.
?- a.
%@ a_start
%@ b_start
%@ a_end
%@ b_end
%@ true.This provides a very expedient way to ensure no collision. In fact, this means we could define: await(G) :- shift(zurück(G)).
handle_await(G,Val,Cont) :- reset(G, zurück(Val), Cont).which would prevent unintentionally unifying with the wrong |
Beta Was this translation helpful? Give feedback.
-
|
Yes exactly, that's what I meant with the exception terms. These wrappers can be library-specific so that each library can tell its own continuations apart from others (more or less, likely more). |
Beta Was this translation helpful? Give feedback.
-
|
I finally figured out how to use is_queue(q(0,_,_)).
is_queue(q(_N,_,_)).
?- queue(Q), is_queue(Q).
%@ Q = q(0,_A,_A)
%@ ; Q = q(0,_A,_A).This is obviously problematic. This made me realize why they don't use integers for the first argument of This allows for: is_queue(q(0,_,_)).
is_queue(q(s(N),_,_)).
?- queue(Q), is_queue(Q).
%@ Q = q(0,_A,_A)
%@ ; false.Which unfortunately still results in a redunant choice point. Fortunately, I remembered @triska had a video on this (that I thought I would never use) and it came in very handy! is_queue(q(N,A,B)) :-
is_queue_(N,A,B).
is_queue_(0,_,_).
is_queue_(s(_),_,_).
?- queue(Q), is_queue(Q).
%@ Q = q(0,_A,_A).🥳 🥳 This makes the code MUCH cleaner! run_task_q(q(N,A,B)) :-
run_task_q_(N,A,B).
run_task_q_(s(N),A,B) :-
taskqcall_deque(q(s(N),A,B), Nq),
run_task_q(Nq).
run_task_q_(0,_,_).and for Before: taskqcall_deque(Q, Nq) :-
que_item_popped(Q,T,Q0),
reset_t(T,T1,cont(Cont),More),
if_(More=true,
( que_task_enque(Q0,T1,Nq0),
resume(Cont,Nq0,Nq)
),
Q0=Nq
).
/* - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
Event Loop Helpers
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - */
reset_t(Goal, Val, Cont, T) :-
( reset(Goal,Val,Cont)
-> T=true
; T=false
).
resume(Cont,Q,Nq) :-
if_(Cont=true,
Q=Nq,
que_task_enque(Q,resume(Cont),Nq)
).
resume(C) :- C.After: taskqcall_deque(Q, Nq) :-
que_item_popped(Q,T,Q0),
handle_await(T,T1,Cont),
resume_internal(Cont,T1,Q0,Nq).
resume_internal(none,_Meaningless,Q,Q).
resume_internal(cont(Cont),Task,Q,Nq) :-
resume_internal_(Cont,Task,Q,Nq).
resume_internal_(true,Task,Q,Nq) :-
que_task_enque(Q,Task,Nq).
resume_internal_(cont:call_continuation(Args),Task,Q,Nq) :-
que_task_enque(Q,Task,Nq0),
que_task_enque(Nq0,cont:call_continuation(Args),Nq).🥳 🥳 🥳 I am extremely pleased with this refactor, thank you for the great tutorial on this topic, @triska ! |
Beta Was this translation helpful? Give feedback.
-
|
Awesome! It's great to see you are making so much excellent progress! One thing I noticed is that |
Beta Was this translation helpful? Give feedback.
-
I understand what you are saying. It is not so important that it be the actual predicate that is exposed, but it is very handy that the Should we make the return of some non-defaulty functor part of the public API? |
Beta Was this translation helpful? Give feedback.
-
|
For the library version I will use: resume_internal(none,_Meaningless,Q,Q).
resume_internal(cont(Cont),Task,Q,Nq) :-
que_task_enque(Q,Task,Nq0),
if_(Cont=true,
Nq=Nq0,
que_task_enque(Nq0,Cont,Nq)
).Assuming that |
Beta Was this translation helpful? Give feedback.
-
|
That, or can you not enqueue (On the understanding that when that |
Beta Was this translation helpful? Give feedback.
-
Brilliant! Works perfectly!! |
Beta Was this translation helpful? Give feedback.
-
|
An interesting application could be asynchronous read and write of files, which Windows has had since about 1992. Linux and perhaps other Unix like OS more recently. |
Beta Was this translation helpful? Give feedback.
-
|
I think there is no established single best logic formalism for event driven programming, that's why every Prolog tried to do something different. Correct me if I'm wrong. offtopicio_uring isn't only about asynchronicity, it is about performance, reducing context switches and utilizing DMA to the fullest possible extent – use that API only if you know exactly why you need it.
Asynchronous IO first appeared in BSD in early '80 and POSIX adopted it '93 🤓
|
Beta Was this translation helpful? Give feedback.
-
|
I've recently also wrote a small program that speaks binary protocol via 2 pipes and I solved it by using this predicate: get_bytes(Stream, B, A) :-
get_byte(Stream, Byte), Byte >= 0, A = [+Byte|X], (X = B; get_bytes(Stream, B, X)).It is hard to do it as DCG, but nevertheless you can plug it into any DCG rule and bytes will "magically" appear one-by-one for parsing. The rule can look like: message_exchange --> get_bytes(read_pipe), request(RequestedData), response(RequestedData), put_bytes(write_pipe).Please note that input bytes have It isn't as fast as your async library, because it reads only 1 byte at a time and does a lot of backtracking, so waiting for new developments. |
Beta Was this translation helpful? Give feedback.
-
|
So simple echo server can look like this: :- initialization(open('in.fifo', read, _, [type(binary),alias(read_pipe),eof_action(eof_code)])).
:- initialization(open('out.fifo', write, _, [type(binary),alias(write_pipe),eof_action(error)])).
main :- phrase((get_bytes(read_pipe), echo, put_bytes(write_pipe)), []) -> main; true.
echo, [-Byte] --> [+Byte].
put_bytes(Stream) --> [] | [-Byte], { put_byte(Stream, Byte) }, put_bytes(Stream).
get_bytes(Stream, B, A) :-
get_byte(Stream, Byte), Byte >= 0, A = [+Byte|X], (X = B; get_bytes(Stream, B, X)). |
Beta Was this translation helpful? Give feedback.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
-
Update:
library(async)PR now available!Edit: significantly cleaned up file, fixed queue implementation, fixed late binding footgun
Current Version
Yesterday, I showed some interesting concurrency semantics I stumbled across. However, there were a few issues that didn't sit well with me.
I realized that was ok for certain kinds of workflows but I was interested in a workflow that would print
which is much more in keeping with standard coroutine control flow.
After some stumbling around, I realized what I really needed was continuations. I had never really had a strong usecase for them before, and I could never quite figure out
library(cont)-- but now I finally did. Now that I understand it, I actually like it better than the semantics in the paper!In a nutshell, it works like this:
Some important caveats -- the third argument of
resetis a functorcont/1. So you really want to be invoking it asreset(Goal, Val, cont(Cont))rather thanreset(Goal, Val, Cont)-- this was really annoying to figure out. If you invoke it asreset(Goal, Val, cont(Cont)), thenContwill be a goal that when invoked will backtrack to where the correspondingshift/1was called from! As you can see, you can REPEAT the computation multiple times if desired -- VERY handy!So, using this, I changed the event loop from
to a slightly more complex (sorry)
Auxiliary Functions for the above snippet
The purpose of this is to allow ANY goal to be run with
shift/1without needing to thread the queue arguments!So rather than needing to write
NOW you can write
Which is significantly more elegant (in my opinion). As a bonus, you get the desired output of
This still allows for parallelism, but with a much cleaner API:
Unfortunately we are missing an important tool for streaming process communication. It is very handy that
process_wait/3has atimeout/1option, but most blocking IO operations do not have something like that. This would make it very difficult to achieve the effect of "busy polling streams" for content without a LOT of gymnastics.Overall it mostly works very well -- there are a few footguns. For example:
So ironically, this works fine:
but if I change the definition of
random_countable_task/0tothe query does
not(Edit: fixed) terminate. I am not sure why this is happening yet, but it's unexpected.Anyway, here is the
full file(Edit: old version) if you want to play with it:Original Version
Beta Was this translation helpful? Give feedback.
All reactions