=encoding utf8

=head1 TITLE

Synopsis 17: Concurrency

=head1 VERSION

    Created: 3 Nov 2013

    Last Modified: 27 December 2014
    Version: 25

This synopsis is based around the concurrency primitives and tools currently being implemented in Rakudo on MoarVM and the JVM.  It covers both things that are already implemented today, in addition to things expected to be implemented in the near future (where "near" means O(months)).

=head1 Design Philosophy

=head2 Focus on composability

Perl 6 generally prefers constructs that compose well, enabling large problems to be solved by putting together solutions for lots of smaller problems. This also helps make it easier to extend and refactor code.

Many common language features related to parallel and asynchronous programming lack composability. For example:

=over

=item *

Locks do not compose, since two independently correct operations using locks may deadlock when performed together.

=item *

Callback-centric approaches tend to compose badly, with chains of asynchronous operations typically leading to deeply nested callbacks. This essentially is just leaving the programmer to do a CPS transform of their own logical view of the program by hand.

=item *

Directly spawning threads on a per-component basis tends to compose badly, as when a dozen such components are used together the result is a high number of threads with no ability to centrally schedule or handle errors.

=back

In Perl 6, concurrency features aimed at typical language users should have good composability properties, both with themselves and also with other language features.

=head2 Boundaries between synchronous and asynchronous should be explicit

Asynchrony happens when we initiate an operation, then continue running our own idea of "next thing" without waiting for the operation to complete. This differs from synchronous programming, where calling a sub or method causes the caller to wait for a result before continuing.

The vast majority of programmers are much more comfortable with synchrony, as in many senses it's the "normal thing". As soon as we have things taking place asynchronously, there is a need to coordinate the work, and doing so tends to be domain specific. Therefore, placing the programmer in an asynchronous situation when they didn't ask for it is likely to lead to confusion and bugs. We should try to make places where asynchrony happens clear.

It's also worthwhile trying to make it easy to keep asynchronous things flowing asynchronously. While synchronous code is pull-y (for example, eating its way through iterable things, blocking for results), asynchronous code is push-y (results get pushed to things that know what to do next).

Places where we go from synchronous to asynchronous, or from asynchronous to synchronous, are higher risk areas for bugs and potential bottlenecks. Thus, Perl 6 should try to provide features that help minimize the need to make such transitions.

=head2 Implicit parallelism is OK

Parallelism is primarily about taking something we could do serially and using multiple CPU cores in order to get to a result more quickly. This leads to a very nice property: a parallel solution to a problem should give the same answer as a serial solution.

While under the hood there is asynchrony and the inherent coordination it requires, on the outside a problem solved using parallel programming is still, when taken as a whole, a single, synchronous operation.

Elsewhere in the specification, Perl 6 provides several features that allow the programmer to indicate that parallelizing an operation will produce the same result as evaluating it serially:

=over

=item *

Hyper operators (L<S03/Hyper operators>) express that parallel operator application is safe.

=item *

Junctions (L<S09/Junctions>) may auto-thread in parallel.

=item *

Feeds (L<S06/Feed operators>) form pipelines and express that the stages may be executed in parallel in a producer-consumer style relationship (though each stage is in itself not parallelized).

=item *

C<hyper> and C<race> list operators (L<S02/The hyper operator>) express that iteration may be done in parallel; this is a generalization of hyper operators.

=back

=head2 Make the hard things possible

The easy things should be easy, and able to be built out of primitives that compose nicely. However, such things have to be built out of what VMs and operating systems provide: threads, atomic instructions (such as CAS), and concurrency control constructs such as mutexes and semaphores. Perl 6 is meant to last for decades, and the coming decades will doubtless bring new ways do do parallel and asynchronous programming that we do not have today. They will still, however, almost certainly need to be built out of what is available.

Thus, the primitive things should be provided for those who need to work on such hard things. Perl 6 should not hide the existence of OS-level threads, or fail to provide access to lower level concurrency control constructs. However, they should be clearly documented as I<not> the way to solve the majority of problems.

=head1 Schedulers

Schedulers lie at the heart of all concurrency in Perl 6. While most users are unlikely to immediately encounter schedulers when starting to use Perl 6's concurrency features, many of them are implemented in terms of it. Thus, they will be described first here to avoid lots of forward references.

A scheduler is something that does the C<Scheduler> role. Its responsibility is taking code objects representing tasks that need to be performed and making sure they get run, as well as handling any time-related operations (such as, "run this code every second").

The current default scheduler is available as C<$*SCHEDULER>. If no such dynamic variable has been declared, then C<$PROCESS::SCHEDULER> is used. This defaults to an instance of C<ThreadPoolScheduler>, which maintains a pool of threads and distributes scheduled work amongst them. Since the scheduler is dynamically scoped, this means that test scheduler modules can be developed that poke a C<$*SCHEDULER> into C<EXPORT>, and then provide the test writer with control over time.

The C<cue> method takes a C<Callable> object and schedules it.

    $*SCHEDULER.cue: { say "Golly, I got scheduled!" }

Various options may be supplied as named arguments.  (All references to time are taken to be in seconds, which may be fractional.)  You may schedule an event to fire off after some number of seconds:

    $*SCHEDULER.cue: in=>10, { say "10s later" }

or at a given absolute time, specified as an C<Instant>:

    $*SCHEDULER.cue: at=>$instant, { say "at $instant" }

If a scheduled item dies, the scheduler will catch this exception and pass it to a C<handle_uncaught> method, a default implementation of which is provided by the C<Scheduler> role. This by default will report the exception and cause the entire application to terminate. However, it is possible to replace this:

    $*SCHEDULER.uncaught_handler = sub ($exception) {
        $logger.log_error($exception);
    }

For more fine-grained handling, it is possible to schedule code along with a code object to be invoked with the thrown exception if it dies:

    $*SCHEDULER.cue:
        { upload_progress($stuff) },
        catch => -> $ex { warn "Could not upload latest progress" }

Use C<:every> to schedule a task to run at a fixed interval, possibly with a delay before the first scheduling.

    # Every second, from now
    $*SCHEDULER.cue: :every(1), { say "Oh wow, a kangaroo!" };

    # Every 0.5s, but don't start for 2s.
    $*SCHEDULER.cue: { say "Kenya believe it?" }, :every(0.5), :in(2);

Since this will cause the given task to be executed at the given interval ad infinitum, there are two ways to make sure the scheduling of the task is halted at a future time.  The first is provided by specifying the C<:times> parameter in the .cue:

    # Every second, from now, but only 42 times
    $*SCHEDULER.cue: :every(1), :times(42), { say "Oh wow, a kangaroo!" };

The second is by specifying code that will be checked at the end of each interval.  The task will be stopped as soon as it returns a True value. You can do this with the C<:stop> parameter.

    # Every second, from now, until stopped
    my $stop;
    $*SCHEDULER.cue: :every(1), :stop({$stop}), { say "Oh wow, a kangaroo!" };
    sleep 10;
    $stop = True;  # task stopped after 10 seconds

The C<.cue> method returns a C<Cancellation> object, which can also be used to stop a repeating cue:

    my $c = $*SCHEDULER.cue: :every(1), { say "Oh wow, a kangaroo!" };
    sleep 10;
    $c.cancel;  # task stopped after 10 seconds

Schedulers also provide counts of the number of operations in various states:

    say $*SCHEDULER.loads;

This returns, in order, the number of cues that are not yet runnable due to delays, the number of cues that are runnable but not yet assigned to a thread, and the number of cues that are now assigned to a thread (and presumably running).
[Conjecture: perhaps these should be separate methods.]

Schedulers may optionally provide further introspection in order to support tools such as debuggers.

There is also a C<CurrentThreadScheduler>, which always schedules things on the current thread. It provides the same methods, just no concurrency, and any exceptions are thrown immediately. This is mostly useful for forcing synchrony in places that default to asynchrony.  (Note that C<.loads> can never return anything but 0 for the currently running cues, since they're waiting on the current thread to stop scheduling first!)

=head1 Promises

A C<Promise> is a synchronization primitive for an asynchronous piece of work that will produce a single result (thus keeping the promise) or fail in some way (thus breaking the promise).

The simplest way to use a C<Promise> is to create one:

    my $promise = Promise.new;

And then later C<keep> it:

    $promise.keep;      # True
    $promise.keep(42);  # a specific return value for kept Promise

Or C<break> it:

    $promise.break;                             # False
    $promise.break(X::Some::Problem.new);       # With exception
    $promise.break("I just couldn't do it");    # With message

The current status of a C<Promise> is available through the C<status> method, which returns an element from the C<PromiseStatus> enumeration.

    enum PromiseStatus (:Planned(0), :Kept(1), :Broken(2));

The result itself can be obtained by calling C<result>. If the C<Promise> was already kept, the result is immediately returned. If the C<Promise> was broken then the exception that it was broken with is thrown. If the C<Promise> is not yet kept or broken, then the caller will block until this happens.

A C<Promise> will boolify to whether the C<Promise> is already kept or broken. There is also an C<cause> method for extracting the exception from a C<Broken> C<Promise> rather than having it thrown.

    if $promise {
        if $promise.status == Kept {
            say "Kept, result = " ~ $promise.result;
        }
        else {
            say "Broken because " ~ $promise.cause;
        }
    }
    else {
        say "Still working!";
    }

You can also simply use a switch:

    given $promise.status {
        when Planned { say "Still working!" }
        when Kept    { say "Kept, result = ", $promise.result }
        when Broken  { say "Broken because ", $promise.cause }
    }

There are various convenient "factory" methods on C<Promise>. The most common is C<start>.

    my $p = Promise.start(&do_hard_calculation);

This creates a C<Promise> that runs the supplied code, and calls C<keep> with its result. If the code throws an exception, then C<break> is called with the C<Exception>. Most of the time, however, the above is simply written as:

    my $p = start {
        # code here
    }

Which is implemented by calling C<Promise.start>.

There is also a method to create a C<Promise> that is kept after a number of seconds, or at a specific time:

    my $kept_in_10s      = Promise.in(10);
    my $kept_in_duration = Promise.in($duration);
    my $kept_at_instant  = Promise.at($instant);

The C<result> is always C<True> and such a C<Promise> can never be broken. It is mostly useful for combining with other promises.

There are also a couple of C<Promise> combinators. The C<anyof> combinator creates a C<Promise> that is kept whenever any of the specified C<Promise>s are kept. If the first promise to produce a result is instead broken, then the resulting C<Promise> is also broken. The cause is passed along. When the C<Promise> is kept, it has a C<True> result.

    my $calc     = start { ... }
    my $timeout  = Promise.in(10);
    my $timecalc = Promise.anyof($calc, $timeout);

There is also an C<allof> combinator, which creates a C<Promise> that will be kept when all of the specified C<Promise>s are kept, or broken if any of them are broken.

[Conjecture: there should be infix operators for these resembling the junctionals.]

The C<then> method on a C<Promise> is used to request that a certain piece of code should be run, receiving the C<Promise> as an argument, when the C<Promise> is kept or broken. If the C<Promise> is already kept or broken, the code is scheduled immediately. It is possible to call C<then> more than once, and each time it returns a C<Promise> representing the completion of both the original C<Promise> as well as the code specified in C<then>.

    my $feedback_promise = $download_promise.then(-> $res {
        given $res.status {
            when Kept   { say "File $res.result().name() download" }
            when Broken { say "FAIL: $res.cause()"                 }
        }
    });

[Conjecture: this needs better syntax to separate the "then" policies from the "else" policies (and from "catch" policies?), and to avoid a bunch of switch boilerplate. We already know the givens here...]

One risk when working with C<Promise>s is that another piece of code will sneak in and keep or break a C<Promise> it should not.  The notion of a promise is user-facing.  To instead represent the promise from the viewpoint of the promiser, the various built-in C<Promise> factory methods and combinators use C<Promise::Vow> objects to represent that internal resolve to fulfill the promise.  ("I have vowed to keep my promise to you.")  The C<vow> method on a C<Promise> returns an object with C<keep> and C<break> methods. It can only be called once during a C<Promise> object's lifetime. Since C<keep> and C<break> on the C<Promise> itself just delegate to C<self.vow.keep(...)> or C<self.vow.break(...)>, obtaining the vow before letting the C<Promise> escape to the outside world is a way to take ownership of the right to keep or break it. For example, here is how the C<Promise.in> factory is implemented:

    method in(Promise:U: $seconds, :$scheduler = $*SCHEDULER) {
        my $p = Promise.new(:$scheduler);
        my $v = $p.vow;
        $scheduler.cue: { $v.keep(True) }, :in($seconds);
        $p;
    }

The C<await> function is used to wait for one or more C<Promise>s to produce a result.

    my ($a, $b) = await $p1, $p2;

This simply calls C<result> on each of the C<Promise>s, so any exception will be thrown.

=head1 Channels

A C<Channel> is essentially a concurrent queue. One or more threads can put values into the C<Channel> using C<send>:

    my $c = Channel.new;
    $c.send($msg);

The call to C<.send> does not block.

Meanwhile, other threads can C<.receive> the values:

    my $msg = $c.receive;

The C<.receive> call does block.  Alternatively, the C<.poll> method takes a message from the channel if one is there or immediately returns C<Nil> if nothing is there.  Only one call to C<.receive> or C<.poll> returns per call of C<.send>, and which listening thread receives each value is left up to the implementation's scheduler.

Channels are ideal for producer/consumer scenarios, and since there can be many senders and many receivers, they adapt well to scaling certain pipeline stages out over multiple workers also. [Conjectural: The two feed operators C<< ==> >> and C<< <== >> are implemented using Channel to connect each of the stages.]

A C<Channel> may be "forever", but it is possible to close it to further sends by telling it to C<close>:

    $c.close();

Trying to C<send> any further messages on a closed channel will throw the C<X::Channel::SendOnDone> exception. Closing a channel has no effect on the receiving end until all sent values have been received. At that point, any further calls to receive will throw C<X::Channel::ReceiveOnDone>. The C<closed> method returns a C<Promise> that is kept when a sender has called C<close> and all sent messages have been received.  Note that multiple calls to the same channel's C<closed> method return the same promise, not a new one.

Like C<.close>, calling C<.fail> will close the channel.  However this represents an abnormal termination, and so, the method must be provided an exception which should be thrown instead of C<X::Channel::ReceiveOnDone>.

A C<Promise> may be obtained from a C<Channel> via the C<.closed> method.  This C<Promise> will be kept when the C<Channel> is closed or broken if it is failed.

A C<whenever> clause (described below) on a C<Channel> will fire for each received value, and may also mark the whenever as "done" using the same criteria as C<.closed>, so it can also be used in order to write a reactor to receive from a channel until it is closed:

    react {
        whenever $channel {
            "Got a $_".say;
        }
    }
    "Done".say;

A C<whenever> clause on a C<Channel> competes for values inside the scheduler alongside any C<.receive> and C<.poll>
calls on that C<Channel> from other threads.  This also is true when the clause is part of a C<supply> block (described
below).

A C<Channel> in list context will iterate all received values lazily, and stop iterating when the channel is closed:

    for @$channel -> $val { ... }
    for $channel.list -> $val { ... }

Note that this is not a combinator, but a means for transfering data from the reactive realm to the lazy realm.  Some reasonable amount of buffering is assumed between the two.  Just like all the above constructs, it competes for values in the scheduler against any other readers.

=head1 Supplies

Channels are good for producer/consumer scenarios, but because each worker blocks on receive, it is not such an ideal construct for doing fine-grained processing of asynchronously produced streams of values. Additionally, there can only be one receiver for each value. Supplies exist to address both of these issues.

A C<Supply> pushes or pumps values to one or more receivers who have registered their interest, by (often indirecty) calling the C<.tap> method on the C<Supply>. This returns a C<Tap> object unique to the receiver, which one may then C<.close> to tell the C<Supply> that the receiver is no longer interested.  Note that this does B<not> happen automatically if the C<Tap> is thrown away, as the C<Supply> also keeps a reference to the C<Tap>. The C<.tap> method takes up to three callables as arguments, one block and two optional named arguments:

    $supply.tap: -> $value { say "Got a $value" },
        done => { say "Reached the end" },
        quit => {
            when X::FooBar { die "Major oopsie" };
            default        { warn "Supply shut down early: $_" }
        }

The first unnamed closure is known as the C<emit> closure. It is invoked whenever a value is emitted into the thing that has been tapped.  The optional named parameter C<done> specifies the code to be invoked when all expected values have been produced and no more values will be emitted. The optional named parameter C<quit> specifies the code to be invoked if there is an error. This also means there will be no further values.

For the same reasons that we have C<Vow>s for C<Promise>s, supplies are split into a C<Supply> role and a C<Supplier> role.  The C<Supplier> role has a C<.Supply> method used to create a live C<Supply>, and C<emit>, C<done>, and C<quit> methods which send the corresponding events to any C<Supply> objects so created.  A C<Supplier> is not tapped directly.  The C<Supply> role does not have C<emit>, C<done>, or C<quit> methods, and may be tapped.  This allows the provider of a C<Supply>, if they so desire, to be sure that one user cannot inject data to other users of a C<Supply>.  Multiple supplies may be created from a C<Supplier>, and all taps of all such supplies receive events sent using said C<Supplier>.

    my $e = Supplier.new;
    my $y = $e.Supply;

    my $t1 = $y.tap({ say $_ });
    $e.emit(1);                              # 1\n
    $e.emit(2);                              # 2\n

    my $t2 = $y.tap({ say 2 * $_ },
              :done({ say "End"  }));
    $e.emit(3);                              # 3\n6\n

    $t1.close;
    $e.emit(4);                              # 8\n
    $e.done;                                 # End\n

The above example demonstrates a simple C<live> supply -- note that C<$t2> missed events 1 and 2 because it was not listening at the time they happened.  From an outside perspective, a live supply will behave as if it were alive and happily producing values even when it never had taps or all its taps have been closed:

    my $r = Supplier.new;
    my $z = $r.Supply;
    $r.emit("A tree fell in the woods");     #

A more common type of supply is an I<on-demand> supply, which creates multiple independent streams of events, starting a new stream each time it is tapped.  Each C<Tap> receives only the events emitted on its private stream. For example, the factory method C<Supply.interval> produces a fresh timer with the appropriate interval each time it is tapped, each of which will stop emitting values when its corresponding C<Tap> is closed:

    my $i = Supply.interval(2);
    my $tick = $i.tap({ say " tick $_" });
    sleep 3;                                 # tick 0\n tick 1\n
    my $tock = $i.tap({ say " tock $_" });
    sleep 3.5;                               # tock 0\n tick 2\n tock 1\n tick 3\n
    $tock.close;
    sleep 3;                                 # tick 4\n
    $tick.close;

By default most C<Supply> objects are serial, meaning they will not emit simultaneous events (be they C<emit>, C<done>, or C<quit> events.)  Other supplies are entirely asyncronous: it is possible for values to be pumped from an any number of asynchronous workers running on different threads.  To tell the difference between the two, the C<.serial> method may be used.  Note that calls to a C<Supplier>'s C<.emit>, C<.done>, or C<.quit> methods always block (and may deadlock in the presence of feedback) until all taps have been run -- the difference between a serial and a non-serial C<Supply> is when multiple threads call C<.emit> simultaneously: on a non-serial C<Supply>, multiple taps may run in parallel, one on each thread, each processing values emitted by that thread.  On a serial C<Supply>, only one tap may run at the same time, and the threads calling C<.emit> may block on the taps processing each other's values, not just their own.  Because of the semantic of "not emit simultaneous events" it is also guaranteed, on a serial C<Supply>, that all taps corresponding to one thread emission will complete before any of the taps from another thread's emission are run.

The C<Supply> class has various methods that produce more interesting kinds of C<Supply>.  These mostly produce serial supplies.  Even the C<Supplier.Supply> method produces a serial supply -- to get a fully asyncronous live supply one must use the C<Supplier.unsanitized-supply> method.  Most factory methods handle creation/retention/destruction of a C<Supplier> internally, or combine prexisting C<Supply> objects, so their users need only to concern themselves with C<Supply> objects.

The following methods are class methods which create a new C<Supply> without requiring an existing C<Supply>.  The methods C<merge>, C<zip> and C<zip-latest> also have class method forms, documented further below with their instance method counterparts.

=over 4

=item from-list

  my $fl = Supply.from-list(^10);

Takes a (potentially lazy) list of values, and returns an on-demand C<Supply> that, when tapped, will iterate over the values and invoke the C<Tap>'s C<emit> callable for each of them, and any C<done> callable at the end. If the iteration at some point produces an exception, then the C<Tap>'s C<quit> callable will be invoked to pass along the exception.

=item interval

  my $e1    = Supply.interval(1);     # Once a second, starting now
  my $e5i10 = Supply.interval(5, 10); # Each 5 seconds, starting in 10 seconds

Produces an on-demand C<Supply> that, when tapped, will produce successive integer values at a regular time interval, with the integer values counting up from 0.

Take the returned tap object and close it to stop the ticks:

    my $e1 = Supply.interval(1).tap(&say);
    # ...later...
    $e1.close();

=back

Supplies are mathematically dual to iterators, and so it is possible to define the same set of operations on them as are available on lazy lists. The key difference is that, while C<grep> on a lazy list I<pulls> a value to process, working synchronously, C<grep> on a Supply has values I<pushed> through it, and pushes those that match the filter onwards to anything that taps it.

The following methods are available on an instantiated C<Supply> (C<$s> in these examples).  Note that calling most of these methods on an on-demand C<Supply> does not constitute demanding a fresh set of values from the C<Supply> -- a tap is not immediately performed.   Instead, tapping the resulting C<Supply> will do so, per C<Tap>.  In other words, connected networks of Supplies do not sit around in the background pointlessly feeding each other values internally, they wait until something is listening.  (This behavior is generally accomplished via C<whenever> clauses, described below.)

=over

=item list

  my @l := $s.list;

Produces a lazy C<List> with the values of the C<Supply>.

=item wait

  $s.wait;

Waits until the specified C<Supply> is C<done> or C<quit>.  In the latter case, throws the exception with which the C<Supply> was quit. 

=item Channel

  my $c = $s.Channel;

Produces a C<Channel> of the values of the given C<Supply>.

=item Promise

  my $p = $s.Promise;

Produces a C<Promise> that will be kept when the C<Supply> is done or broken if it is quit.

=item last

  my $l = $s.last(42);  # default: 1

Produces a C<Supply> that will only emit the last N values of the given C<Supply> (after that C<Supply> is C<done>).  Default is the final value.

=item grab

  my $g = $s.grab( { .sort } ); # sort the values of a Supply

Produces a C<Supply> which will cache all values emitted by the given C<Supply> until it is done.  It will then call the given closure and then C<.emit> each of the return values of the closure, and then call C<.done> on itself.

=item flat

  my $f = $s.flat;

Produces a C<Supply> in which all values of the original supply are flattened then individually emitted.

=item do

  my $seen;
  my $d = $s.do( {$seen++} );

Produces a C<Supply> that is identical to the original supply, but will execute the given code for its side-effects, once for each emitted value, before running any taps.  Only one thread will ever be executing the side-effect code at a time; others will block behind it.  In addition to serializing the side-effect, the resulting supply is also serial, even if the one it was created from is not.  The side-effect code is only run for emitted values, not when the original is quit or done.

=item act

  my $seen;
  $s.act( {$seen++} );

This is a special case of C<Supply>.do, that will also create a tap on the given C<Supply>, so that you only need to worry about writing the side-effect code.  Returns the C<Tap>.

=item grep

  my $g = $s.grep( * > 5 );
  my $g = $s.grep(Int);

Produces a C<Supply> that only provides values that you want.  Takes either a C<Callable> (which is supposed to return a C<True> value to pass on emitted values) or a value to be smartmatched against.

=item map

Analogous to C<List>'s C<.map> method, but produces a C<Supply>.

  my $m = $s.map( * * 5 );

Produces a C<Supply> that provides its original's Supply values multiplied by 5.

  my $m2 = $s.map( { $_ xx 2 } );

Produces a C<Supply> that provides its original's Supply values twice.

=item unique

  my $u = $s.unique( :as( {$_} ), :with( &[===] ), :expires(1) );

Produces a C<Supply> that only provides unique values, as defined by the optional C<as> and C<with> named parameters (same as L<List.unique>). The optional C<expires> parameter specifies how long to wait (in seconds) before "resetting" and not considering a value to have been seen, even if it's the same as an old value.

=item squish

  my $q = $s.squish( :as( {$_} ), :with( &[===] ), :expires(1) );

Produces a C<Supply> that only provides sequentially different values, as defined by the optional C<as> and C<with> named parameters (same as L<List.squish>). The optional C<expires> parameter specifies how long to wait (in seconds) before "resetting" and not squishing a new value with an old one, even if they are the same.

=item max

  my $a = $s.max(&by); # default &infix:<cmp>

Produces a C<Supply> that produces the B<maximum> values of the specified C<Supply>.  In other words, from a continuously ascending C<Supply> it will produce all the values.  From a continuously descending C<Supply> it will only produce the first value.  The optional parameter specifies the comparator, just as with C<Any.max>.

=item min

  my $i = $s.min(&by); # default &infix:<cmp>

Produces a C<Supply> that produces the B<minimum> values of the specified C<Supply>.  In other words, from a continuously descending C<Supply> it will produce all the values.  From a continuously ascending C<Supply> it will only produce the first value.  The optional parameter specifies the comparator, just as with C<Any.min>.

=item minmax

  my $m = $s.minmax(&by); # default &infix:<cmp>

Produces a C<Supply> that, for each value emitted, produces C<Range>s with the B<minimum> and B<maximum> values seen thus far on the specified C<Supply>.  The optional parameter specifies the comparator, just as with C<Any.minmax>.

=item batch

  my $b = $s.batch( :elems(100), :seconds(1) );

Produces a C<Supply> that batches the values of the given Supply by either the number of elements (using the C<elems> named parameter) or the maximum number of seconds (using the C<seconds> named parameter) or both.  Values are grouped in a single array element when flushed.

=item throttle

  my $t = $s.throttle( $elems, $seconds );

Produces a C<Supply> that throttles emitting the values of the given Supply on the created C<Supply> by the number of elements (specified by the first parameter) per number of seconds (specified by the second parameter).

=item elems

  my $e = $s.elems($seconds?); # default: see all

Produces a C<Supply> that, for each value emitted, produces the number of elements seen thus far in the given C<Supply>.  You can also specify an interval to only see the number of elements seen once per that interval.

=item rotor

  my $b = $s.rotor(@cycle);

Produces a "rotoring" C<Supply> with the same semantics as List.rotor.

=item delayed

  my $d = $s.delayed( 3.5 );  # delay supply 3.5 seconds

Produces a C<Supply> that passes on the values of the given Supply with the given delay (in seconds).

=item stable

  my $u = $s.stable( $seconds, :$scheduler );

Produces a C<Supply> that only passes on a value if it wasn't superseded by another value in the given time (in seconds).  Optionally uses another scheduler than the default scheduler, using the C<scheduler> named parameter.

=item start

  my $t = $s.start( {...} );

Takes a closure and, for each supplied value, schedules the closure to run on another thread. It then emits a Supply (resulting in us having a supply of supplies) that will either have a single value emitted and then be done if the async work completes successfully, or quit if the work fails. Useful for kicking off work on the thread pool if you do not want to block up the thread pushing values at you (maybe 'cus you are reacting to UI events, but have some long-running work to kick off). Usually used in combination with C<migrate>.

=item migrate

  my $m = $t.migrate;

Produces a continuous C<Supply> from a C<Supply>, in which each value is a C<Supply> which emits a value from the original C<Supply> unchanged.  As soon as a new C<Supply> appears, it will close the previously emitted C<Supply> and send the next value from the original C<Supply> to the most recently emitted C<Supply> instead.  Can be used in combination with C<schedule-on>.

=item schedule-on

  my $o = $m.schedule-on( $scheduler );

This allows a C<Supply>'s emit/done/quit to be scheduled on another scheduler. Useful in GUI situations, for example, where the final stage of some work needs to be done on some UI scheduler in order to have UI updates run on the UI thread.

=item reduce

  my $r = $s.reduce( {...} );

Produces a C<Supply> that will emit each reduction from the given C<Supply>, much like the triangular reduction metaoperator (C<[\...]>) on C<List>s.

=item lines

  my $l = $s.lines;              # chomp lines
  my $l = $s.lines( :!chomp );   # do *not* chomp lines

Produces a C<Supply> that will emit the characters coming in line by line from a C<Supply> that's usually created by some asynchronous I/O operation. The optional C<:chomp> named parameter indicates whether to remove line separators: the default is C<True>.

=item words

  my $w = $s.words;

Produces a C<Supply> that will emit the characters coming in word by word from a C<Supply> that's usually created by some asynchronous I/O operation.

=item classify

  my $c = $s.classify( {.WHAT} );  # one Supply per type of value
  my $h = $s.classify( %mapper );
  my $a = $s.classify( @mapper );

Produces a C<Supply> which emits C<Pair>s consisting of a classification value as C<.key>, and a C<Supply> as the C<.value>.  That supply will then emit any values from the original C<Supply> which match the classifier.  Parameters behave similar to C<List.classify>, but does not support multi-level classification.

=item categorize

  my $c = $s.categorize( {@categories} );
  my $h = $s.categorize( %mapper );
  my $a = $s.categorize( @mapper );

Produces a C<Supply> in which emits C<Pair>s  consisting of a classification value as C<.key>, and a C<Supply> as the C<.value>.  That supply will then emit any values from the original C<Supply> which match the classification value.  Unlike C<.classify>, more than one C<Pair> may be emitted per value emitted from the original C<Supply>.  See C<List.categorize> for a description of the behavior of the parameters.

=item reverse

  my $r = $s.reverse;

Produces a C<Supply> that emits the values of the given Supply in reverse order. Please note that this C<Supply> will only start delivering values when the given C<Supply> is C<.done>, much like C<.grab>.

=item sort

  my $o = $s.sort(&by);  # default &infix:<cmp>

Produces a C<Supply> that emits the values of the given Supply in sorted order. Please note that this C<Supply> will only start delivering values when the given C<Supply> is C<.done>.  Optionally accepts a comparator C<Block>.

=back

There are some combinators that deal with bringing multiple supplies together:

=over

=item C<merge>

  my $m = $s1.merge($s2);

  my $m = Supply.merge(@s);  # also as class method

Produces a C<Supply> containing the values produced by given and the specified supply or supplies, and triggering C<done> once all of the supplies have done so.  The resulting supply is serial, even if any or all of the merged supplies are not.

=item C<zip>

  my $z = $s1.zip($s2);                   # defaults to :with( &[,] )

  my $z = Supply.zip(@s, :with( &[,] ));  # also as class method

Produces a C<Supply> that pairs together items from the given and the specified supply or supplies, using C<< infix:<,> >> by default or any other user-supplied function with the C<with> named parameter. The resulting supply is serial, even if any or all of the zipped supplies are not.

=item C<zip-latest>

  my $z = $s1.zip-latest($s2);                  # like zip, defaults to :with( &[,] )

  my $z = Supply.zip-latest(@s, :with( &[,] )); # also a method on Supply.

  my $z = Supply.zip-latest( @s, :initial(42,63) ); # initial state

Produces a C<Supply> that will emit tuples of values as soon as any combined Supply produces a value. Before any tuples are emitted, all supplies have to have produced at least one value. By default, it uses C<< infix:<,> >> to produce the tuples, but the named parameter C<with> can override that. The resulting supply is serial, even if any or all of the merged supplies are not.

The named parameter C<initial> can optionally be used to indicate the initial state of the values to be emitted.

=back

[TODO: plenty more of these: while, until...]

The above combinators which involve multiple source supplies need care in their implementation, since values may arrive at any point from each source, and possibly at the same time.  To help write such combinators, the C<supply> block and C<whenever> clause are used.  These, along with the C<react> block, are available for general use as well.

A C<supply> block is a convenient way to create an on-demand C<Supply>.  It is just a declaration and does not run until the C<Supply> is tapped (which may happen multiple times resulting in multiple runs/clones of the block.)  Within a C<supply> block, emit can be used to emit values to the tapper, and done can be used to convey that there will be no more values.

  my $s = supply {
      emit 'a value!';
      emit 'another!';
      done;
  }

The emit and done can be in nested scopes, and follow the same rules as C<gather> and C<take>, except they look for an enclosing C<supply> or C<react> rather than a C<gather>.   There is no corresponding C<quit> statement: instead, any unhandled exception thrown inside of a supply block will be passed onwards using C<.quit>.

Likewise, a C<whenever> clause also looks around itself lexotically for an enclosing C<supply> or C<react>.  A C<whenever> clause takes a predicate which may be a C<Supply>, C<Channel>, C<Promise> or C<Iterable>, followed by a consequent block.  These clauses guarantee that only one thread will enter their consequent or that of any other C<whenever> clause inside the same enclosing C<supply> or C<react> block at the same time -- blocking any C<.emit> call on a C<Supply> or leaving any additional values in a Channel's C<.send> queue in the scheduler.

  my $s = Supply.interval(1);
  my $c = Channel.new;
  my $l := (while $++ < 4 { NEXT { sleep 1 }; Int(now - BEGIN now) });
  my $p = Promise.in(3);
  my %contended_hash;
  start {
      for 0..4 { sleep 1; $c.send($_); LAST { $c.close } }
  }
  react {
      # This is safe, because the whenevers protect %contended_hash
      whenever $s { %contended_hash{$_}++; done if $_ > 4 };
      whenever $c { %contended_hash{$_}++ };
      whenever $l { %contended_hash{$_}++ };
      whenever $p { %contended_hash{$_}++ };
  }
  %contended_hash.say; # 0 => 3, 1 => 3, 2 => 3, 3 => 3, 4 => 2, 5 => 1, True => 1

A whenever with a C<Supply> as a predicate will tap that predicate when the enclosing block is run (assuming control reaches the whenever clause.)  It will then offer its consequent block as the emit closure for the resulting C<Tap>.  However, it is sometimes best just to think of a C<whenever> as a loop which runs once for each value produced by the predicate C<Supply>, using the produced values as the topic.  Following this model, a C<QUIT> or C<LAST> phaser inside the consequent block may be used to provide a quit or done closure (respectively) to the C<Tap>.  The resulting C<Tap> itself is ensconced internally, but may also be accessed as the return value of the entire whenever clause.  This is not necessary just in order for a whenever to close its own tap: C<last> may be called from within the consequent to do so.  Labels may be used as normal to close the taps of outer clauses.  Closing the last active tap closes the generated C<Supply>.

A whenever block with a C<Channel> as a predicate behaves analogously, running the closure once for every value sent to it by the scheduler, as if it were constantly calling C<.receive> on the C<Channel>.  A C<Promise> predicate runs the block when the C<Promise> is kept or broken as if it were calling C<await>.  An C<Iterable> predicate will start a new iterator when the whenever clause is run, entering the consequent closure for each value pulled from the C<Iterator>.  In all such cases, a C<QUIT> block inside the C<whenever> will catch exceptions generated from within the predicate.  Handling these exceptions will prevent C<.quit> from being called on the C<Supply> created by the block.

Note that it is possible for control never to reach a C<whenever> clause when a C<supply> or C<react> block is run.  In this case, the whenever clause has no effect.  If control reaches a C<whenever> clause more than once (for example, if it is inside a loop) multiple taps on the predicate (which may be different each time) are created.  A C<whenever> clause may also be executed later, after a C<supply> block has produced a C<Supply> which has then been tapped, or while waiting for a C<react> block to fall off the bottom.  This could happen from inside another C<whenever> clause in response to an event, or from an externally added C<Tap>.  This is legal, and dynamically adds a new tap (of the predicate) to the C<Supply> which caused this to happen.  That is to say, once active, a C<Supply> may indeed use C<whenever> clauses to self-modify, as long as it was originally syntactically anchored to a C<supply> or C<react> block:

   my $s1 = Supply.interval(1).share;
   my $s2 = supply {
       whenever $s1 {
           emit($_);
           done if $_ > 4;
       }
   };
   $s2.tap({
       if 2 < $_ < 5 {
           "MORE $_".say;
            whenever $s1 {
                "OHAI $_".say;
            }
       }
   });
   sleep 10;
   # MORE 3
   # MORE 4
   # OHAI 4
   # OHAI 5
   # OHAI 5

Destructuring subsignatures may be used with C<whenever> clause topics via pointy block syntax.

A C<supply> block containing C<whenever> clauses will call C<done> on itself when all C<whenever> clause predicates (which have executed) have had C<.done>/C<quit>, C<.close>/C<.fail> or C<.keep>/C<.break> called on them, appropriately, or in the case of an C<Iterable> predicate, when the end has been reached.  These conditions are equivalent to manually closing the clause with C<last>.

A C<react> block will not wait to be tapped -- it will immediately run and then wait for a done or quit condition, after which point, control will resume below the C<react> block.

=head1 System events exposed as Supplies

System events, such as signals, or mouse events, can be exposed as Supplies. Because of lack of portability, these will most likely be implemented as third-party modules.

Basic signal support is offered by the C<signal> function, which takes one or more C<Signal> enums, and an optional C<scheduler> named parameter.  It produces a C<Supply> which, when tapped, will C<emit> any signal coming in. For example:

  signal(SIGINT).tap( { say "Thank you for your attention"; exit 0 } );

would catch Control-C, thank you, and then exit.  Of course, you don't need to exit immediately.  Here's an example of how you would make sure that an iteration in a loop is completed before exiting:

  for @todo {
      state $quitting;
      state $tap = signal(SIGINT).tap( { $quitting = True } );
      LAST  $tap.close;
      LEAVE exit(0) if $quitting;
      ... # code to protect
  }

This probably could use some syntactic sugar.

The list of supported C<Signals> can be found by checking C<Signal::.keys>, as you would any enum.

=head1 I/O features exposed as Supplies

Various I/O-related things are also exposed as supplies. For example, it is possible to get notifications on changes to files or files (directly) in a directory, using:

    IO::Notification.watch-path(".").tap(-> $file {
        say "$file changed";
    });

This is quite a mouthful, so there is a shortcut available with the C<IO> coercer and the C<watch> method:

    ".".IO.watch.tap: -> $file { say "$file changed" };

Note that since I/O callbacks are, by default, scheduled on the thread pool, then it's possible that your callback will be executing twice on the same thread. One way to cope is with C<act>:

    ".".IO.watch.act(-> $file {
        state %changes;
        say "$file changed (change {++%changes{$file}})";
    });

It can also take C<done> and C<quit> named parameters; these go to the tap, while the C<emit> closure is put in a C<do>. A C<Tap> is returned, which may be closed in the usual way. (Note that the name C<act> is also a subtle reference to actor semantics.)

=head1 Inter-Process Communication exposed as Promises and Supplies

Starting external processes is rather easy: C<shell()>, C<run()> and C<qx//>. Having external processes run asynchronously, is slightly more involved. But not much.  The workhorse of asynchronous IPC in Perl 6 is C<Proc::Async>:

    my $proc = Proc::Async.new( $path, @args );

If you like to B<send> data to the process, you need to open it with the C<:w> named parameter.

    my $proc = Proc::Async.new( $path, @args, :w );

By default, the current environment (as available in C<%*ENV>) will be set for the external process.  You can override this with the :ENV named parameter:

    my $proc = Proc::Async.new( $path, @args, :ENV(%hash) );

The returned object can then be called whenever needed to start the external process.  However, before you do that, one needs to be clear what to do about the output of the external process.  Getting information back from the external process's C<STDOUT> or C<STDERR>, is done by a C<Supply> that either gets characters or bytes.

    $proc.stdout.act(&say);   # simply pass it on to our $*OUT as chars
    $proc.stderr.act(&note);  # and $*ERR as chars, but could be any code

or:

    $proc.stdout(:bin).act: { # process STDOUT bytes };
    $proc.stderr(:bin).act: { # process STDERR bytes };

So, to make sure no information will be lost, you need to create and tap the supplies B<before> the process is started.

To start the external process, you need to call the C<.start> method.  It returns a C<Promise> that becomes C<Kept> (and True) if the process concludes successfully, or C<Broken> (and False) if the process failed for some reason.

    my $done = $proc.start( :$scheduler = $*SCHEDULER );

To send data to the running process, you can use the C<.print>, C<.say> and C<.write> methods on the C<Proc::Async> object:

    my $printed = $proc.print( "Hello world\n", :$scheduler = $*SCHEDULER );
    my $said    = $proc.say(   "Hello world",   :$scheduler = $*SCHEDULER );
    my $written = $proc.write( $buffer,         :$scheduler = $*SCHEDULER );

They all also return a C<Promise> that is C<Kept> when communication with the process was successful.

Some programs expect their C<STDIN> to be closed to signify the end of their processing.  This can be achieved with the C<.close-stdin> method:

    $proc.close-stdin;

Finally, if your process as going awry, you can stop it with the C<.kill> method:

    $proc.kill;            # sends HUP signal to process
    $proc.kill("SIGINT");  # send INT signal
    $proc.kill(1);         # if you just know the signal number on your system

The parameter should be something that is acceptable to the Kernel.signal method.

=head1 The Event Loop

There is no event loop.  Previous versions of this synopsis mentioned an event loop that would be underlying all concurrency.  In this version, this is not the case.

=head2 Threads

VM-level threads, which typically correspond to OS-level threads, are exposed through the C<Thread> class. Whatever underlies it, a C<Thread> should always be backed by something that is capable of being scheduled on a CPU core (that is, it may I<not> be a "green thread" or similar). Most users will not need to work with C<Thread>s directly. However, those building their own schedulers may well need to do so, and there may be other exceptional circumstances that demand such low-level control.

The easiest way to start a thread is with the C<start> method, which takes a C<Callable> and runs it on a new thread:

    my $thread = Thread.start({
        say "Gosh, I'm in a thread!";
    });

It is also possible to create a thread object, and set it running later:

    my $thread = Thread.new(code => {
        say "A thread, you say?";
    });
    # later...
    $thread.run();

Both approaches result in C<$thread> containing a C<Thread> object. At some point, C<finish> should be called on the thread, from the thread that started it. This blocks until the thread has completed.

    say "Certainly before the thread is started";
    my $thread = Thread.start({ say "In the thread" });
    say "This could come before or after the thread's output";
    $thread.finish();
    say "Certainly after all the above output";

As an alternative to C<finish>, it is possible to create a thread whose lifetime is bounded by that of the overall application. Such threads are automatically terminated when the application exits. In a scenario where the initial thread creates an application lifetime thread and no others, then the exit of the initial thread will cause termination of the overall program. Such a thread is created by either:

    my $thread = Thread.new(:code({ ... }), :app_lifetime);

Or just, by using the C<start> method:

    my $thread = Thread.start({ ... }, :app_lifetime);

The property can be introspected:

    say $thread.app_lifetime; # True/False

Each thread also has a unique ID, which can be obtained by the C<id> property.

    say $thread.id;

This should be treated as an opaque number. It can not be assumed to map to any particular operating system's idea of thread ID, for example. For that, use something that lets you get at OS-level identifiers (such as calling an OS API using NativeCall).

A thread may also be given a name.

    my $thread = Thread.start({ ... }, :name<Background CPU Eater>);

This can be useful for understanding its usage. Uniqueness is not enforced; indeed, the default is "<anon>".

A thread stringifies to something of the form:

    Thread<id>(name)

For example:

    Thread<1234>(<anon>)

The currently executing thread is available through C<$*THREAD>. This is even available in the initial thread of the program, in this case by falling back to C<$PROCESS::THREAD>, which is the initial thread of the process.

Finally, the C<yield> method can be called on C<Thread> (not on any particular thread) to hint to the OS that the thread has nothing useful to do for the moment, and so another thread should run instead.

=head2 Atomic Compare and Swap

The Atomic Compare and Swap (CAS) primitive is directly supported by most modern hardware. It has been shown that it can be used to build a whole range of concurrency control mechanisms (such as mutexes and semaphores). It can also be used to implement lock-free data structures. It is decidedly a primitive, and not truly composable due to risk of livelock. However, since so much can be built out of it, Perl 6 provides it directly.

A Perl 6 implementation of CAS would look something like this:

    sub cas($ref is rw, $expected, $new) {
        my $seen = $ref;
        if $ref === $expected {
            $ref = $new;
        }
        return $seen;
    }

Except that it happens atomically. For example, a crappy non-reentrant mutex could be implemented as:

    class CrappyMutex {
        has $!locked = 0;

        method lock() {
            loop {
                return if cas($!locked, 0, 1) == 0;
            }
        }

        method unlock() {
            $!locked = 0;
        }
    }

Another common use of CAS is in providing lock-free data structures. Any data structure can be made lock-free as long as you're willing to never mutate it, but build a fresh one each time. To support this, there is another C<&cas> candidate that takes a scalar and a block. It calls the block with the seen initial value. The block returns the new, updated value. If nothing else updated the value in the meantime, the reference will be updated. If the CAS fails because another update got in first, the block will be run again, passing in the latest value.

So, atomically incrementing a variable is done thusly:

    cas $a, { $_.succ };    # $a++

or more generally for all assignment meta-operators:

    cas $a, { $_ * 5 };     # $a *= 5

Another  example, implementing a top-5 news headlines list to be accessed and updated without ever locking, as:

    class TopHeadlines {
        has $!headlines = [];   # Scalar holding array, as CAS needs

        method headlines() {
            $!headlines
        }

        method add_headline($headline) {
            cas($!headlines, -> @current {
                my @new = $headline, @current;
                @new.pop while @new.elems > 5;
                @new
            });
        }
    }

It's the programmer's duty to ensure that the original data structure is never mutated and that the block has no side-effects (since it may be run any number of times).

=head1 Low-level primitives

Perl 6 offers high-level concurrency methods, but in extreme cases, like if you need to implement a fundamentally different mechanism, these primitives are available.

=head2 Locks

Locks are unpleasant to work with, and users are pushed towards higher level synchronization primitives. However, those need to be implemented via lower level constructs for efficiency. As such, a simple lock mechanism - as close to what the execution environment offers as possible - is provided by the C<Lock> class. Note that it is erroneous to rely on the exact representation of an instance of this type (for example, don't assume it can be mixed into). Put another way, treat C<Lock> like a native type.

A C<Lock> is instantiated with C<new>:

    $!lock = Lock.new;

The best way to use it is:

    $!lock.protect: {
        # code to run with the lock held
    }

This acquires the lock, runs the code passed, and then releases the lock. It ensures the lock will be released even if an exception is thrown. It is also possible to do:

    {
        $!lock.lock();
        # do stuff
        LEAVE $!lock.unlock()
    }

When using the C<lock> and C<unlock> methods, the programmer must ensure that the lock is unlocked. C<Lock> is reentrant. Naturally, it's easy to introduce deadlocks. Again, this is a last resort, intended for those who are building first resorts.

=head2 Semaphore

The C<Semaphore> class implements traditional semaphores that can be initiated with a fixed number of permits and offers the operations C<acquire> to block on a positive number of permits to become available and then reduce that number by one, C<tryacquire> to try to acquire a permit, but return C<False> instead of blocking if there are no permits available yet. The last operation is C<release>, which will increase the number of permits by one.

The initial number of permits may be negative, positive or 0.

Some implementations allow for race-free acquisition and release of multiple permits at once, but this primitive does not offer that capability.

=head1 AUTHORS

    Jonathan Worthington <jnthn@jnthn.net>
    Elizabeth Mattijsen <liz@wenzperl.nl>

=for vim:set expandtab sw=4: