### ----------------------------------------------------
### -- ParaSeq
### -- Licenses: Artistic-2.0
### -- Authors: Elizabeth Mattijsen
### -- File: lib/ParaSeq.rakumod
### ----------------------------------------------------
### -- Chunk 1 of 3
# Intended to be as fast as possible and to someday become part of the
# Rakudo core
use v6.*;
use nqp;

#- ParaQueue -------------------------------------------------------------------
# A blocking concurrent queue to which one can nqp::push and from which one
# can nqp::shift
my class ParaQueue is repr('ConcBlockingQueue') { }

#- globals ---------------------------------------------------------------------

my uint $default-batch  = 16;
my uint $default-degree = Kernel.cpu-cores-but-one;
my uint $target-nsecs   = 1_000_000;  # .001 second
my uint $target-nsecs2  = nqp::mul_i($target-nsecs,2);

# Helper sub to determine granularity of batches depending on the signature
# of a Callable
my sub granularity(&callable) {
    my $count := &callable.count;
    $count == Inf ?? 1 !! $count
}

# Helper sub to return the positional argument as fast as possible
my sub identity(\value) is raw { value }

# Shortcut for long names, just for esthetics
my constant IB = IterationBuffer;
my constant IE = IterationEnd;

# An empty IterationBuffer, for various places where it is needed as a source
my constant emptyIB   = nqp::create(IB);
my constant emptyList = emptyIB.List;

# A separator line
my constant separator = "-" x 80;

#- exception handling ----------------------------------------------------------

my $exceptions := nqp::create(ParaQueue);

# Create a Bag of the exceptions so far
my sub bag-the-exceptions() {
    my $buffer := nqp::create(IB);
    nqp::while(
      nqp::elems($exceptions),
      nqp::push($buffer, nqp::shift($exceptions))
    );
    $buffer.List.map(*.gist.chomp).Bag
}

# Show the exceptions that weren't bagged before
END {
    if nqp::elems($exceptions) -> uint $total {
        my      $bagged := bag-the-exceptions;
        my uint $elems   = $bagged.elems;
        my str  $s       = $elems == 1 ?? "" !! "s";

        my $ERR := $*ERR;
        $ERR.say:
          "Caught $elems unique exception$s (out of $total) in hypered code:";
        for $bagged.sort(-*.value) {
            $ERR.say: separator;
            $ERR.print(.value ~ "x: ") if .value > 1;
            $ERR.say: .key;
        }
        $ERR.say: separator;
    }
}

#- BlockMapper -----------------------------------------------------------------

# A role to give a Block extra capabilities, needed to be able to run
# map / grep like parallel sequences
my role BlockMapper {
    has uint $!granularity;  # the granularity of the Block
    has      $!FIRSTs;       # any FIRST phasers
    has      $!LASTs;        # any LAST phasers

    # Set up the extra capabilities
    method setup(uint $granularity, :$no-first, :$no-last) {
        $!granularity = $granularity;
        my $phasers  := nqp::getattr(self,Block,'$!phasers');

        # If it is a single LEAVE phaser, we need to convert it to a
        # multi phaser hash first
        if nqp::isinvokable($phasers) {
            my $LEAVE := nqp::create(IB);
            nqp::push($LEAVE,$phasers);
            $phasers := nqp::hash(
              '!LEAVE-ORDER', $LEAVE, 'LEAVE', $LEAVE
            );
        }

        # Make sure we're going to do all stuff on a clone
        elsif $phasers {
            $phasers := nqp::clone($phasers);
        }

        # Otherwise set up a clean slate
        else {
            $phasers := nqp::hash;
        }

        # Save any FIRST phasers and remove them.  NOTE: this currently
        # only serves as a flag, as the FIRST phasers are actually
        # embedded into the code block itself.  But this may change
        # with RakuAST.
        nqp::deletekey($phasers,"FIRST")
          if $!FIRSTs := self.callable_for_phaser("FIRST");
        $!FIRSTs := Mu if $no-first;

        # Save any LAST phasers and remove them
        nqp::deletekey($phasers,"LAST")
          if $!LASTs := self.callable_for_phaser("LAST");
        $!LASTs := Mu if $no-last;

        # These are the phasers for now
        nqp::bindattr(self,Block,'$!phasers',$phasers);

        # Return invocant for convenience
        self
    }

    # The actual running logic, which makes sure that any FIRST / LAST phasers
    # are only run once, and any "last" in the block is also handled as
    # expected
    method run(
      uint $first,             # is this the first batch
      uint $last,              # is this the last batch
      str  $method,            # method to be executed
           $input     is raw,  # the IterationBuffer to work on
           $result    is raw,  # the result iterator
           $semaphore is raw,  # the semaphore to which result will be pushed
           $postprocess?       # any postprocessing to be done
    ) {
        # Make sure we have our own mapper / phasers
        my $this-mapper  := nqp::clone(self);
        my $this-phasers := nqp::clone(nqp::getattr(self,Block,'$!phasers'));
        nqp::bindattr($this-mapper,Block,'$!phasers',$this-phasers);

        # logic for setting a phaser for this run
        sub set-phaser(str $name, Callable:D $phaser) {
            nqp::push((my $buffer := nqp::create(IB)),$phaser);
            nqp::bindkey($this-phasers,$name,$buffer);
        }

        # Set the FIRST flag on the first batch only
        if $first && $!FIRSTs {

            # Actually make FIRST phaser getting ran (legacy).  Need to
            # create a clone of the code, as otherwise a second thread
            # *may* be running the block with the first flag still set
            # and thus cause the FIRST phaser to be executed (in the
            # legacy runtime)
            nqp::p6setfirstflag(
              nqp::bindattr(
                $this-mapper,Code,'$!do',
                nqp::clone(nqp::getattr($this-mapper,Code,'$!do'))
              )
            );
            set-phaser("FIRST", $!FIRSTs);
        }

        # This will become 1 if a "last" was executed
        my $*LAST;

        # Set LAST phaser, to actually execute if it's the last
        # batch, or a "last" got executed, as long as there was
        # no other thread having executed it
        set-phaser("LAST", {
            if ($last || $*LAST) && $!LASTs {
                $!LASTs();
                $!LASTs := Mu;  # make sure we only call LAST once
            }
        }) if $!LASTs;

        # Fetch the iterator and setup output buffer
        my $iterator := $input.List."$method"($this-mapper, |%_).iterator;
        my $output   := nqp::create(IB);

        # Run element by element if post-processing needed
        if $postprocess {
            nqp::until(
              nqp::eqaddr((my $pulled := $iterator.pull-one),IE),
              $postprocess($iterator, $output, $pulled)
            );
        }

        # Just get all the value
        else {
            $iterator.push-all($output);
        }

        # Mark this semaphore as the last to be handled if a
        # "last" was executed
        $result.set-lastsema($semaphore, $output) if $*LAST;

        # Return the result
        $output
    }

    # Fool the core's mapper into thinking that this block does not actually
    # have FIRST phaser, which stops it from setting the first flag (on
    # Rakudo 2024.06+)
    multi method has-phaser("FIRST" --> False) { }
    multi method has-phaser(Str:D) { nextsame }
}

#- BufferIterator --------------------------------------------------------------
# An iterator that takes a buffer and an iterator, and first produces all
# of the values from the buffer, and then from the iterator

my class BufferIterator does Iterator {
    has $!buffer;  # first buffer to produce from
    has $.source;  # iterator to produce from onwards

    method new(\buffer, \source) {
        my $self := nqp::create(self);
        nqp::bindattr($self,BufferIterator,'$!buffer',buffer);
        nqp::bindattr($self,BufferIterator,'$!source',source);
        $self
    }

    method pull-one(BufferIterator:D:) {
        nqp::elems($!buffer)
          ?? nqp::shift($!buffer)
          !! $!source.pull-one
    }

    method skip-at-least(BufferIterator:D: uint $skipping) {

        # Need to skip more than in initial buffer
        my uint $elems  = nqp::elems($!buffer);
        if $skipping > $elems {
            nqp::setelems($!buffer,0);
            $!source.skip-at-least($skipping - $elems)
        }

        # Less than in inital buffer, remove so many entries from it
        else {
            nqp::splice($!buffer,emptyIB,0,$skipping);
            1
        }
    }

    method push-all(BufferIterator:D: \target) {
        nqp::istype(target,IB)
          ?? nqp::splice(target,$!buffer,nqp::elems(target),0)
          !! $!buffer.iterator.push-all(target);
        nqp::setelems($!buffer,0);

        $!source.push-all(target)
    }

    method is-lazy(BufferIterator:D:) {
        $!source.is-lazy
    }
    method is-deterministic(BufferIterator:D:) {
        $!source.is-deterministic
    }
    method is-monotonically-increasing(BufferIterator:D:) {
        $!source.is-monotonically-increasing
    }

    # If there is a buffer available, return it and reset in object, else Nil
    method zap-buffer(BufferIterator:D:) {
        my $buffer := $!buffer;
        if nqp::elems($buffer) {
            $!buffer := nqp::create(IB);
            $buffer
        }
        else {
            Nil
        }
    }

    method stats(--> emptyList) { }
}

#- SquishIterator --------------------------------------------------------------
# An iterator that takes a source iterator that produces batches of values
# and which compares the last of a batch with the first of the next batch
# using .squish semantics

my class SquishIterator does Iterator {
    has $!source    # source iterator
      handles <is-lazy is-deterministic is-monotonically-increasing>;
    has $!as;       # any transformation prior to comparison
    has $!with;     # comparator to be used
    has $!current;  # the current buffer producing values
    has $!last;     # last value seen of the current buffer

    method new($source, &as, &with) {
        my $self := nqp::create(self);
        nqp::bindattr($self,SquishIterator,'$!source', $source         );
        nqp::bindattr($self,SquishIterator,'$!as',     &as             );
        nqp::bindattr($self,SquishIterator,'$!with',   &with           );
        nqp::bindattr($self,SquishIterator,'$!current',$source.pull-one);
        $self
    }

    method !next-batch() {
        my $current := $!source.pull-one;
        if nqp::eqaddr($current,IE) {
            IterationEnd
        }
        else {
            my $nextlast := nqp::atpos($current,-1);
            nqp::shift($current)
              if $!with($!as(nqp::atpos($current,0)),$!as($!last));

            nqp::elems($!current := $current)
              ?? nqp::shift($current)
              !! self!next-batch
        }
    }

    method pull-one(SquishIterator:D:) {
        nqp::elems(my $current := $!current) > 1
          ?? nqp::shift($current)
          !! nqp::elems($current)
            ?? ($!last = nqp::shift($current))
            !! self!next-batch
    }

    method push-all(SquishIterator:D: \target --> IterationEnd) {
        my $source  := $!source;
        my &as      := $!as;
        my &with    := $!with;
        my $current := $!current;

        my $last := nqp::elems($current) ?? nqp::atpos($current,-1) !! $!last;
        target.append($current);

        nqp::until(
          nqp::eqaddr(($current := $source.pull-one),IE),
          nqp::stmts(
            (my $nextlast := nqp::atpos($current,-1)),
            nqp::if(
              with(as(nqp::atpos($current,0)),as($last)),
              nqp::shift($current)
            ),
            ($last := $nextlast),
            target.append($current)
          )
        );
    }
}

#- WhichIterator ---------------------------------------------------------------
# An iterator that takes a buffer filled with values on even locations, and
# .WHICH strings on odd values, and produces just the values on the even
# positions

my class WhichIterator does Iterator {
    has $!buffer;    # buffer to produce from

    method new(\buffer) {
        my $self := nqp::create(self);
        nqp::bindattr($self,WhichIterator,'$!buffer',buffer);
        $self
    }

    method pull-one() {
        my $buffer := $!buffer;

        nqp::if(
          nqp::elems($buffer),
          nqp::stmts(
            (my $pulled := nqp::shift($buffer)),
            nqp::shift($buffer),
            $pulled
          ),
          IterationEnd
        )
    }

    method skip-at-least(uint $skipping is copy) {
        $skipping = $skipping + $skipping;

        nqp::if(
          $skipping > nqp::elems($!buffer),
          0,
          nqp::stmts(
            nqp::splice($!buffer,emptyIB,0,$skipping),
            1
          )
        )
    }

    method push-all(\target --> IterationEnd) {
        my $buffer := $!buffer;

        nqp::while(
          nqp::elems($buffer),
          nqp::stmts(
            nqp::push(target,nqp::shift($buffer)),
            nqp::shift($buffer)
          )
        );
    }

    method stats(--> emptyList) { }
}

#- UniqueIterator --------------------------------------------------------------
# A variation on the WhichIterator that also takes a hash with lookup status
# If a key exists in the seen hash, and the value is > 0, then it is the first
# time this value is produced.  In that case, produce the value and set the
# value in the hash to 0 (indicating any further occurrences should be ignored

my class UniqueIterator does Iterator {
    has $!seen;    # lookup hash: value > 1 means first, so pass, 0 means skip
    has $!buffer;  # buffer with value / key alternates

    method new(\seen, \buffer) {
        my $self := nqp::create(self);
        nqp::bindattr(
          $self,UniqueIterator,'$!seen',nqp::getattr(seen,Map,'$!storage')
        );
        nqp::bindattr($self,UniqueIterator,'$!buffer',buffer);
        $self
    }

    method pull-one() {
        my $seen   := $!seen;
        my $buffer := $!buffer;

        nqp::while(
          nqp::elems($buffer),
          nqp::stmts(
            (my $pulled := nqp::shift($buffer)),
            nqp::if(
              nqp::existskey($seen,(my str $key = nqp::shift($buffer))),
              nqp::if(
                nqp::atkey($seen,$key),
                nqp::stmts(
                  nqp::bindkey($seen,$key,0),
                  (return $pulled)
                )
              ),
              (return $pulled)
            )
          )
        );

        IterationEnd
    }
}

#- ParaIterator ----------------------------------------------------------------
# An iterator that takes a number of ParaQueues and produces all values
# from them until the last ParaQueue has been exhausted.  Note that the
# queue of queues is also a ParaQueue so that queues can also be added
# and removed in a thread-safe manner

my class ParaIterator does Iterator {
    has $!parent                # ParaSeq parent object
      handles <is-lazy is-deterministic is-monotonically-increasing>;

    has           $!current;    # current producer
    has ParaQueue $!queues;     # other queues to produce from
    has ParaQueue $!pressure;   # backpressure provider
    has ParaQueue $.waiting;    # queue with unprocessed ParaStats objects
    has IB        $!stats;      # list with processed ParaStats objects
    has uint      $!racing;     # 1 if racing, 0 if hypering
    has uint      $!batch;      # initial / current batch size
    has uint      $!auto;       # 1 if batch size automatically adjust on load
    has uint      $!processed;  # number of items processed so far
    has uint      $!produced;   # number of items produced so far
    has uint      $!nsecs;      # nano seconds wallclock used so far
    has           $!lastsema;   # the last semaphore that should be processed
    has atomicint $!unhandled;  # number of unhandled batches

    method new(
           \parent,
           \pressure,
      uint $racing,
      uint $batch,
      uint $auto,
    ) {
        my $self := nqp::create(self);

        nqp::bindattr($self,ParaIterator,'$!parent',  parent                );
        nqp::bindattr($self,ParaIterator,'$!current', emptyIB               );
        nqp::bindattr($self,ParaIterator,'$!queues',  nqp::create(ParaQueue));
        nqp::bindattr($self,ParaIterator,'$!pressure',pressure              );
        nqp::bindattr($self,ParaIterator,'$!waiting', nqp::create(ParaQueue));
        nqp::bindattr($self,ParaIterator,'$!stats',   nqp::create(IB)       );

        nqp::bindattr_i($self,ParaIterator,'$!racing', $racing);
        nqp::bindattr_i($self,ParaIterator,'$!batch',  $batch );
        nqp::bindattr_i($self,ParaIterator,'$!auto',   $auto  );

        # Set up initial barrier.  Once the final batch has been scheduled
        # for processing, .close-queues will decrement this final value and
        # thus indicate completion
        nqp::bindattr_i($self,ParaIterator,'$!unhandled',1);
        $self
    }

    # Mark another batch for processing
    method mark-batch-for-processing() {
        nqp::atomicinc_i($!unhandled);
    }

    # Handling when the current result buffer is exhausted
    method !next() {
        if nqp::atomicload_i($!unhandled) {
            nqp::atomicdec_i($!unhandled);
            nqp::shift($!queues);  # block until provided/IE on premature end
        }
        else {
            IE                     # delivered all batches
        }
    }

    # Thread-safely handle next batch
    method !next-batch($next is raw) {

        # Next buffer to be processed
        my $buffer := $!racing ?? $next !! nqp::shift($next);

        # Stop delivering after this buffer
        if nqp::eqaddr($next,$!lastsema) {
            nqp::unshift($!queues,IE);
        }

        # Need to handle stats to calculate optimum batch size
        elsif $!auto {
            self!update-stats unless nqp::isnull(nqp::atpos($!waiting,0));
            $!batch = nqp::div_i(nqp::mul_i($!processed,$target-nsecs),$!nsecs)
              if $!nsecs;

            # Initiate more work with possibly updated batch size
            nqp::push($!pressure,$!batch);
        }

        # No need to handle stats here and now
        else {
            # Initiate more work using default batch value calculated
            nqp::push($!pressure,$!batch);
        }

        $buffer
    }

    # Update statistics from the waiting queue
    method !update-stats() {
        # Fetch statistics of batches that have been produced until now
        my uint $processed = $!processed;
        my uint $produced  = $!produced;
        my uint $nsecs     = $!nsecs;
        nqp::until(
          nqp::isnull(nqp::atpos($!waiting,0)),
          nqp::stmts(
            nqp::push($!stats,my $stats := nqp::shift($!waiting)),
            ($processed = nqp::add_i($processed,$stats.processed     )),
            ($produced  = nqp::add_i($produced, $stats.produced      )),
            ($nsecs     = nqp::add_i($nsecs,    $stats.smoothed-nsecs))
          )
        );

        # Update total statistics
        $!processed = $processed;
        $!produced  = $produced;
        $!nsecs     = $nsecs;
    }

    method pull-one(ParaIterator:D:) {
        nqp::if(
          nqp::elems($!current),
          (my $pulled := nqp::shift($!current)),
          nqp::if(
            nqp::eqaddr((my $next := self!next),IE),
            (return IE),
            nqp::stmts(
              ($!current := self!next-batch($next)),
              ($pulled   := self.pull-one)
            )
          )
        );

        $pulled
    }

    method skip-at-least(ParaIterator:D: uint $skipping) {
        my      $current  := $!current;
        my      $queues   := $!queues;
        my uint $toskip    = $skipping;

        nqp::while(
          1,
          nqp::stmts(
            (my uint $elems = nqp::elems($current)),
            nqp::if(
              $elems > $toskip,  # can't ignore whole buffer, only part
              nqp::stmts(
                ($!current := nqp::splice($current, nqp::list, 0, $toskip)),
                (return 1)
              )
            ),
            nqp::if(             # ignored whole buffer, fetch next
              nqp::eqaddr((my $next := self!next),IE),
              nqp::stmts(
                nqp::setelems($!current,0),  # invalidate current contents
                (return 0)                   # really done
              )
            ),
            ($toskip = nqp::sub_i($toskip,$elems)),
            ($current := $!current := self!next-batch($next))
          )
        );
    }

    proto method push-all(|) {*}

    # Optimized version appending to an IterationBuffer
    multi method push-all(ParaIterator:D: IterationBuffer:D \target) {
        my $current := $!current;

        loop {
            nqp::splice(target,$current,nqp::elems(target),0);
            nqp::eqaddr((my $next := self!next),IE)
              ?? (return IE)
              !! ($current := self!next-batch($next));
        }
    }

    # Slower generic version that needs to coerce each buffer to a List
    # to ensure the correct semantics with .append
    multi method push-all(ParaIterator:D: Any:D \target) {
        my $current := $!current;

        loop {
            nqp::while(
              nqp::elems($current),
              target.push(nqp::shift($current))
            );
            nqp::eqaddr((my $next := self!next),IE)
              ?? (return IE)
              !! ($current := self!next-batch($next));
        }
    }

    # Optimized version used for its side-effects
    method sink-all(ParaIterator:D:) {
        loop {
            nqp::eqaddr((my $next := self!next),IE)
              ?? (return IE)
              !! self!next-batch($next);
        }
    }

    method close-queues(ParaIterator:D:) is implementation-detail {
        nqp::atomicdec_i($!unhandled)  # remove the barrier
    }

    method stats(ParaIterator:D:) is raw is implementation-detail {
        self!update-stats;
        $!stats
    }

    # Set the last batch to be produced, unless it was already set:
    # the first one set, wins!
    method set-lastsema($lastsema is raw, $output is raw) {
        $!lastsema := $!racing ?? $output !! $lastsema
          unless nqp::isconcrete($!lastsema);
    }

    # Add a semaphore to the queue.  This is a ParaQueue to which a fully
    # processed IterationBuffer will be pushed when it is ready.  This
    # guarantees that the order of the results will be the order in which
    # this method will be called.  When racing, just return the queues
    # Paraqueue to push to whenever results are ready to be pushed
    method semaphore() is implementation-detail {
        $!racing
          ?? $!queues
          !! nqp::push($!queues,nqp::create(ParaQueue))
    }
}

#- ParaStats -------------------------------------------------------------------
# A class for keeping stats about the production of a single result queue
# from a batch of values

class ParaStats {
    has uint $.threadid;
    has uint $.ordinal;
    has uint $.processed;
    has uint $.produced;
    has uint $.nsecs;

    method new(uint $ordinal, uint $processed, uint $produced, uint $nsecs) {
        my $self := nqp::create(self);
        nqp::bindattr_i($self, ParaStats, '$!threadid',
          nqp::threadid(nqp::currentthread)
        );

        nqp::bindattr_i($self, ParaStats, '$!ordinal',   $ordinal  );
        nqp::bindattr_i($self, ParaStats, '$!processed', $processed);
        nqp::bindattr_i($self, ParaStats, '$!produced',  $produced );
        nqp::bindattr_i($self, ParaStats, '$!nsecs',     $nsecs    );
        $self
    }

    # If the time it took for the batch to be processed is more than twice
    # the target slice time, then most likely there has been a garbage
    # collection run, which halts the world for a tiny bit.  But enough to
    # mess up any calculations.  So we basically do a modulo on the time,
    # but make sure we have at least the target slice time in there as well.
    # This appears to produce the best timing calculations
    method smoothed-nsecs(ParaStats:D:) {
        $!nsecs > $target-nsecs2
          ?? nqp::add_i(nqp::mod_i($!nsecs,$target-nsecs),$target-nsecs)
          !! $!nsecs
    }

    multi method gist(ParaStats:D:) {
        "#$!threadid - $!ordinal: $!processed / $!produced ($!nsecs nsecs)"
    }
}

#- ParaSeq ---------------------------------------------------------------------
# The class containing all of the logic for parallel sequences
class ParaSeq is Seq {
    has           $!source;      # iterator producing source values
    has           $!SCHEDULER;   # $*SCHEDULER to be used
    has           $!snitcher;    # the snitcher to be called before/after
    has uint      $!racing;      # 1 = racing, 0 = hypering
    has uint      $!batch;       # initial batch size, must be > 0
    has uint      $!auto;        # 1 if batch size automatically adjusts
    has uint      $!catch;       # 1 if exceptions should be caught
    has uint      $!degree;      # number of CPUs, must be > 1

#- "last" handling -------------------------------------------------------------

    # A very hacky way to catch the excution of "last".  This requires the
    # the scope of the mapper to be within the scope where the ParaSeq module
    # is imported to
    proto sub last(|) is export {*}
    multi sub last() {
        my $LAST := $*LAST;
        $LAST = 1 unless nqp::istype($LAST,Failure);
        &CORE::last()
    }
    multi sub last(Mu $value) {
        my $LAST := $*LAST;
        $LAST = 1 unless nqp::istype($LAST,Failure);
        &CORE::last($value)
    }

#- private helper methods ------------------------------------------------------

    method !cue(&callable) {
        my $promise;

        # Already living inside an async environment, so use this Promise
        # with its dynamic context
        with $*PROMISE {
            $promise := $_;
        }

        # First time we do an async call: create a promise that can be
        # living inside the async code's scope, to allow access to the
        # dynamic variables living in *this* dynamic context
        else {
            $promise := Promise.new(:scheduler($!SCHEDULER));
            nqp::bindattr($promise,Promise,'$!dynamic_context',nqp::ctx);
        }

        # Set up code to be actually queued, which adds a $*PROMISE to its
        # dynamic scope, so that references to dynamic variables will refer
        # to dynamic variables of the original scope
        my &code := {
            my $*PROMISE := $promise;
            callable()
        }

        # Perform the actual cueing
        $!catch
          ?? $!SCHEDULER.cue: nqp::p6capturelex(&code.clone), :catch({
                 nqp::push($exceptions,$_);
                 try .resume;
             })
          !! $!SCHEDULER.cue: nqp::p6capturelex(&code.clone)
    }

    # Set up object and return it
    method !setup(
           $source,
      uint $racing,
      uint $batch,
      uint $auto,
      uint $catch,
      uint $degree,
    ) is hidden-from-backtrace {
        my $self := nqp::create(self);
        nqp::bindattr($self, ParaSeq, '$!source',    $source    );
        nqp::bindattr($self, Seq,     '$!iter',      nqp::null  );
        nqp::bindattr($self, ParaSeq, '$!SCHEDULER', $*SCHEDULER);
        nqp::bindattr($self, ParaSeq, '$!snitcher',  nqp::null  );

        nqp::bindattr_i($self, ParaSeq, '$!racing', $racing);
        nqp::bindattr_i($self, ParaSeq, '$!batch',  $batch );
        nqp::bindattr_i($self, ParaSeq, '$!auto',   $auto  );
        nqp::bindattr_i($self, ParaSeq, '$!catch',  $catch );
        nqp::bindattr_i($self, ParaSeq, '$!degree', $degree);
        $self
    }

    # Start the async process with the given processor logic and the
    # required granularity for producing values
    method !start(&processor, uint $granularity = 1, uint :$slow) {

        # Logic for making sure batch size has correct granularity
        my &granulize := $granularity == 1
          ?? -> uint $batch { $batch }
          !! -> uint $batch {
                nqp::mod_i($batch,$granularity)
                  ?? nqp::mul_i(nqp::div_i($batch,$granularity),$granularity)
                       || $granularity
                  !! $batch
             }

        # Get copy of first buffer if possible
        my $first;
        if nqp::istype($!source,BufferIterator)
          && ($first := $!source.zap-buffer) {

            # We can get rid of the BufferIterator now
            $!source := $!source.source;
            nqp::until(
              nqp::elems($first) %% $granularity
                || nqp::eqaddr((my $pulled := $!source.pull-one),IE),
              nqp::push($first,$pulled)
            );
        }

        # Batch size allowed to vary, so set up back pressure queue
        # with some simple variation
        my $pressure  := nqp::create(ParaQueue);
        my uint $batch = $!batch;
        if $!auto {
            my uint $half = nqp::div_i($batch,2) || $batch;
            $batch = 0;

            nqp::push($pressure,granulize($batch = $batch + $half))
              for ^$!degree;
        }

        # Fixed batch size, so set up back pressure queue with fixed values
        else {
            $batch = granulize($batch);
            nqp::push($pressure,$batch) for ^$!degree;
        }

        # Set up the result iterator
        nqp::bindattr(
          self, Seq, '$!iter',
          ParaIterator.new(self, $pressure, $!racing, $!batch, $!auto)
        );

        # Batcher logic for fast batching, where no special phaser handling
        # is required
        sub fast() {
            my $source   := $!source;
            my $result   := nqp::getattr(self,Seq,'$!iter');
            my $snitcher := $!snitcher;

            my uint $ordinal;    # ordinal number of batch
            my uint $exhausted;  # flag: 1 if exhausted

            # Queue the first buffer we already filled, and set up the
            # result iterator
            if nqp::isconcrete($first) {
                $snitcher($first) unless nqp::isnull($snitcher);
                $result.mark-batch-for-processing;
                processor($ordinal, $first, $result.semaphore);
            }

            # Until we're halted or have a buffer that's not full
            nqp::until(
              $exhausted,       # nothing left to batch
              nqp::stmts(
                ($exhausted = nqp::eqaddr(
                  $source.push-exactly(
                    (my $buffer := nqp::create(IB)),
                    granulize(nqp::shift($pressure))  # wait for ok to proceed
                  ),
                  IE
                )),
                nqp::if(         # add if something to add
                  nqp::elems($buffer),
                  nqp::stmts(
                    nqp::unless(
                      nqp::isnull($snitcher),
                      $snitcher($buffer.List)
                    ),
                    $result.mark-batch-for-processing;
                    processor(++$ordinal, $buffer, $result.semaphore);
                  )
                )
              )
            );

            # No more result queues will come
            $result.close-queues
        }

        # Batcher logic for slower batching, which tells each process unit
        # whether or not it is the last batch, so that it can make sure the
        # correct phasers are executed
        sub slow() {
            my $source   := $!source;
            my $result   := nqp::getattr(self,Seq,'$!iter');
            my $snitcher := $!snitcher;

            my uint $ordinal; # ordinal number of batch

            # flag: 1 if exhausted
            my uint $exhausted =
              nqp::eqaddr((my $initial := $source.pull-one),IE);

            # Queue the first buffer we already filled, and set up the
            # result iterator