### ----------------------------------------------------
### -- ParaSeq
### -- Licenses: Artistic-2.0
### -- Authors: Elizabeth Mattijsen
### -- File: lib/ParaSeq.rakumod
### ----------------------------------------------------
### -- Chunk 2 of 3

            if nqp::isconcrete($first) {
                $snitcher($first) unless nqp::isnull($snitcher);
                $result.mark-batch-for-processing;
                processor($ordinal, $exhausted, $first, $result.semaphore);
            }

            # Until the source iterator is exhausted or we're halted
            until $exhausted {

                # Setup buffer with initial value
                nqp::push((my $buffer := nqp::create(IB)),$initial);

                # Wait for ok to proceed
                if granulize(nqp::shift($pressure)) -> uint $needed {

                    # Fetch any additional values needed and set flag
                    $exhausted = nqp::eqaddr(
                      $source.push-exactly($buffer,$needed - 1),IE
                    );
                }

                # If not already exhausted attempt to get one value more
                $exhausted = nqp::eqaddr(($initial := $source.pull-one),IE)
                  unless $exhausted;

                # Process if there's something to process
                if nqp::elems($buffer) {

                    # Snitch if so indicated
                    $snitcher($buffer.List) unless nqp::isnull($snitcher);

                    # Schedule the batch
                    $result.mark-batch-for-processing;
                    processor(
                      ++$ordinal, $exhausted, $buffer, $result.semaphore
                    );
                }
            }

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

        # Make sure the scheduling of further batches actually happen in a
        # separate thread, whether they'd be slow or fast
        self!cue: $slow ?? &slow !! &fast;

        # All scheduled now, so let the show begin!
        self
    }

    # If there was no result iterator yet, return invocant: can still use
    # this object.  If there was, clone the invocant and make the result
    # iterator the source for the clone.
    method !re-shape() {
        if nqp::isnull(nqp::getattr(self,Seq,'$!iter')) {
            self
        }
        else {
            my $self := nqp::clone(self);
            nqp::bindattr(
              $self,ParaSeq,'$!source',
              nqp::getattr(self,Seq,'$!iter')
            );
            nqp::bindattr($self,Seq,'$!iter',nqp::null);
            $self
        }
    }

    # Create a clone of the invocant, set the given iterator as source and
    # reset the result iterator
    method !re-source($source is raw) {
        my $self := nqp::clone(self);
        nqp::bindattr($self,ParaSeq,'$!source',$source);
        nqp::bindattr($self,Seq,'$!iter',nqp::null);
        $self
    }

    # Mark the given queue as done
    method !batch-done(
      uint $ordinal,
      uint $then,
           $input is raw,
           $semaphore is raw,
           $output is raw
    ) {
        # Push the statistics into the waiting queue of the iterator
        # so that it has up-to-date information to determine future
        # batch sizes
        nqp::push(
          nqp::getattr(self,Seq,'$!iter').waiting,
          ParaStats.new(
            $ordinal,
            nqp::elems($input),
            nqp::elems($output),
            nqp::time() - $then
          )
        ) if nqp::istype(nqp::getattr(self,Seq,'$!iter'),ParaIterator);

        # Make the produced values available to the result iterator
        nqp::push($semaphore,$output);
    }

#- debugging -------------------------------------------------------------------

    multi method snitch(ParaSeq:D: &snitcher = &note) {
        $!snitcher := &snitcher;
        self
    }

#- where all the magic happens under the hood ----------------------------------

    method iterator(ParaSeq:D:) {
        nqp::ifnull(
          nqp::getattr(self,Seq,'$!iter'),
          $!source
        )
    }

    method sink(ParaSeq:D: --> Nil) { self.iterator.sink-all }

#- introspection ---------------------------------------------------------------

    method auto(ParaSeq:D:) { nqp::hllbool($!auto) }

    method batch-sizes(ParaSeq:D:) {
        self.stats.map(*.processed).minmax
    }

    method catch(ParaSeq:D:) { nqp::hllbool($!catch) }

    method default-batch()  is raw { $default-batch  }
    method default-degree() is raw { $default-degree }

    method exceptions() { bag-the-exceptions }

    multi method is-lazy(ParaSeq:D:) { $!source.is-lazy }

    method is-deterministic(ParaSeq:D:) {
        nqp::hllbool($!source.is-deterministic)
    }
    method is-monotonically-increasing(ParaSeq:D:) {
        nqp::hllbool($!source.is-monotonically-increasing)
    }

    method racing(ParaSeq:D:) { nqp::hllbool($!racing) }

    method stats(ParaSeq:D:) {
        nqp::isnull(nqp::getattr(self,Seq,'$!iter'))
          ?? ()
          !! nqp::getattr(self,Seq,'$!iter').stats.List
    }

    method processed(ParaSeq:D:) { self.stats.map(*.processed).sum }
    method produced(ParaSeq:D:)  { self.stats.map(*.produced ).sum }

    method threads(ParaSeq:D:) {
        self.stats.map(*.threadid).unique.List
    }

#- endpoints -------------------------------------------------------------------

    proto method first(|) {*}
    multi method first(ParaSeq:D:) { self.head }
    multi method first(ParaSeq:D: Callable:D $matcher) {
        self.Seq.first($matcher, |%_)
    }
    multi method first(ParaSeq:D: $matcher) {
        self.Seq.first($matcher, |%_)
    }
    multi method first(ParaSeq:D: $matcher, :$end!) {
        $end
          ?? self.List.first($matcher, :end, |%_)
          !! self.first($matcher, |%_)
    }

    proto method head(|) {*}
    multi method head(ParaSeq:D:) {
        my $value := self.iterator.pull-one;
        nqp::eqaddr($value,IE) ?? Nil !! $value
    }

    multi method join(ParaSeq:D: Str:D $joiner = '') {
        self.fail-iterator-cannot-be-lazy('.join') if self.is-lazy;

        # Logic for joining a buffer
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            self!cue: {
                my uint $then = nqp::time;

                nqp::push(
                  (my $output := nqp::create(IB)),
                  $input.List.join($joiner)
                );

                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Finish off by joining the joined strings
        self!start(&processor).List.join($joiner)
    }

    proto method minmax(|) {*}
    multi method minmax(ParaSeq:D: &by = &[cmp]) {
        self.fail-iterator-cannot-be-lazy('.minmax') if self.is-lazy;

        # Logic for joining a buffer
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            self!cue: {
                my uint $then = nqp::time;

                my $min := my $max := nqp::atpos($input,0);

                my uint $elems = nqp::elems($input);
                my uint $i;
                nqp::while(
                  nqp::islt_i(++$i,$elems),  # intentionally skip first
                  nqp::stmts(
                    my $value := nqp::atpos($input,$i);
                    nqp::if(
                      nqp::islt_i(by($value,$min),0),
                      $min := $value
                    ),
                    nqp::if(
                      nqp::isgt_i(by($value,$max),0),
                      $max := $value
                    )
                  )
                );

                my $output := nqp::create(IB);
                nqp::push($output,$min);
                nqp::push($output,$max);

                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Start processing and return result of batches
        my $batches := self!start(&processor).IterationBuffer;

        # Post process the result
        my $min := nqp::shift($batches);
        my $max := nqp::shift($batches);
        my $value;

        nqp::while(
          nqp::elems($batches),
          nqp::stmts(
            nqp::if(
              nqp::islt_i(by(($value := nqp::shift($batches)),$min),0),
              ($min := $value)
            ),
            nqp::if(
              nqp::isgt_i(by(($value := nqp::shift($batches)),$max),0),
              ($max := $value)
            )
          )
        );

        Range.new($min, $max)
    }

    proto method pick(|) {*}
    multi method pick(ParaSeq:D:) { self.List.pick }

    multi method reduce(ParaSeq:D: Callable:D $reducer) {

        # Logic for reducing a buffer
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            self!cue: {
                my uint $then = nqp::time;

                self!batch-done(
                  $ordinal, $then, $input, $semaphore, $input.reduce($reducer)
                );
            }
        }

        # Finish off by reducing the reductions
        self!start(&processor).Seq.reduce($reducer)
    }

    proto method roll(|) {*}
    multi method roll(ParaSeq:D:) { self.List.roll }

    multi method sum(ParaSeq:D:) {
        self.fail-iterator-cannot-be-lazy('.sum') if self.is-lazy;

        # Logic for summing a buffer
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            self!cue: {
                my uint $then = nqp::time;

                nqp::push((my $output := nqp::create(IB)),$input.List.sum);

                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Finish off by summing the sums
        self!start(&processor).List.sum
    }

    proto method tail(|) {*}
    multi method tail(ParaSeq:D:) { self.List.tail }

#-- antipairs ------------------------------------------------------------------

    multi method antipairs(ParaSeq:D:) {
        my uint $base;  # base offset

        # Logic for queuing a buffer for .antipairs
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            my uint $elems  = nqp::elems($input);
            $base = $base + $elems;

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                # Store the Pairs as fast as possible
                my uint $i;
                nqp::while(
                  $i < $elems,
                  nqp::push(
                    $output,
                    Pair.new(nqp::atpos($input,$i),$offset + $i++)
                  )
                );

                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Let's go!
        self!re-shape!start(&processor)
    }

#- batch -----------------------------------------------------------------------

    method !batch(uint $size, uint $partial) {

        # Logic for queuing a buffer for batches
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                # Ultra-fast batching
                my uint $elems = nqp::elems($input);
                my uint $start;
                my uint $end = nqp::sub_i($size,1);
                nqp::while(
                  $start < $elems,
                  nqp::stmts(
                    nqp::push($output,nqp::slice($input,$start,$end).List),
                    ($start = nqp::add_i($start,$size)),
                    ($end   = nqp::add_i($end,  $size)),
                    nqp::if(
                      $end >= $elems,
                      nqp::if(                          # incomplete final batch
                        $partial,
                        ($end = nqp::sub_i($elems,1)),  # take final partially
                        ($start = $elems)               # stop iterating
                      )
                    )
                  )
                );
                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Let's go!
        self!re-shape!start(&processor, $size)
    }

    proto method batch(|) {*}
    multi method batch(ParaSeq:D: Int:D :$elems!) { self!batch($elems,1) }
    multi method batch(ParaSeq:D: Int:D  $elems ) { self!batch($elems,1) }

#- deepmap ---------------------------------------------------------------------

    multi method deepmap(ParaSeq:D: Block:D $mapper) {
        my str @phasers;
        nqp::push(@phasers,$_) if $mapper.has-phaser($_) for <FIRST NEXT LAST>;
        warn "@phasers.join(", ") phaser(s) will be ignored" if @phasers;

        my uint $granularity = granularity($mapper);
        $granularity == 1
          ?? self!mapBlock($mapper, 'deepmap', :no-first, :no-last)
          !! die "Can only handle blocks with one parameter, got $granularity"
    }
    multi method deepmap(ParaSeq:D: Callable:D $mapper) {
        self!mapCallable($mapper, 'deepmap')
    }

#- duckmap ---------------------------------------------------------------------

    multi method duckmap(ParaSeq:D: Block:D $mapper) {
        my str @phasers;
        nqp::push(@phasers,$_) if $mapper.has-phaser($_) for <FIRST NEXT LAST>;
        warn "@phasers.join(", ") phaser(s) will be ignored" if @phasers;

        my uint $granularity = granularity($mapper);
        $granularity == 1
          ?? self!mapBlock($mapper, 'duckmap', :no-first, :no-last)
          !! die "Can only handle blocks with one parameter, got $granularity"
    }
    multi method duckmap(ParaSeq:D: Callable:D $mapper) {
        self!mapCallable($mapper, 'duckmap')
    }

#- grep ------------------------------------------------------------------------

    # Handling Callables that can have phasers
    multi method grep(ParaSeq:D: Block:D $matcher, :$k, :$kv, :$p) {

        # Need to have postprocessing
        if $k || $kv || $p { 
            my uint $granularity = granularity($matcher);
            my uint $base;  # base offset to be added for each batch

            my %nameds := Map.new( $k ?? :k !! :kv );
            my $mapper := (nqp::clone($matcher) but BlockMapper)
              .setup($granularity, :no-first, :no-last);

            sub processor(
              uint $ordinal,
              uint $last,
                   $input     is raw,
                   $semaphore is raw,
            ) {
                my uint $offset = $base;
                $base = $base + nqp::elems($input);

                # Set up post-processor depend on named argument.  Needs
                # to be done here because of scoping of $offset
                my $postprocess := $k
                  ?? -> $, $output is raw, $key {
                         nqp::push($output,$offset + $key)
                     }
                  !! $kv
                    ?? -> $iterator, $output is raw, $key {
                           nqp::push($output,$offset + $key),     # key
                           nqp::push($output,$iterator.pull-one)  # value
                       }
                    !! -> $iterator, $output is raw, $key {
                           nqp::push(
                             $output,
                             Pair.new($offset + $key, $iterator.pull-one)
                           )
                       }

                # Cue the actual work
                self!cue: {
                    my uint $then = nqp::time;

                    self!batch-done(
                      $ordinal, $then, $input, $semaphore,
                      $mapper.run(
                        $ordinal == 0, $last,
                        'grep', $input, nqp::getattr(self,Seq,'$!iter'),
                        $semaphore, $postprocess, |%nameds
                      )
                    );
                }
            }

            # Let's go using the slower path
            self!re-shape!start(&processor, $granularity, :slow)
        }       

        # Use standard "map-like" handling
        else {
            self!mapBlock($matcher, 'grep', :no-first, :no-last)
        }
    }

    # Handling other Callables that cannot have phasers
    multi method grep(ParaSeq:D: Callable:D $matcher, :$k, :$kv, :$p) {
        my uint $base;  # base offset for :k, :kv, :p

        # Logic for queuing a buffer for bare grep { }, producing values
        sub v(uint $ordinal, uint $last, $input is raw, $semaphore is raw) {
            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                $input.Seq.grep($matcher).iterator.push-all($output);

                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Logic for queuing a buffer for grep { } :k
        sub k(uint $ordinal, uint $last, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            $base = $base + nqp::elems($input);

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                my $iterator := $input.Seq.grep($matcher, :k).iterator;
                nqp::until(
                  nqp::eqaddr((my $key := $iterator.pull-one),IE),
                  nqp::push($output,$offset + $key)
                );
                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Logic for queuing a buffer for grep { } :kv
        sub kv(uint $ordinal, uint $last, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            $base = $base + nqp::elems($input);

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                my $iterator := $input.Seq.grep($matcher, :kv).iterator;
                nqp::until(
                  nqp::eqaddr((my $key := $iterator.pull-one),IE),
                  nqp::stmts(
                    nqp::push($output,$offset + $key),     # key
                    nqp::push($output,$iterator.pull-one)  # value
                  )
                );
                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Logic for queuing a buffer for grep { } :p
        sub p(uint $ordinal, uint $last, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            $base = $base + nqp::elems($input);

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                my $iterator := $input.Seq.grep($matcher, :kv).iterator;
                nqp::until(
                  nqp::eqaddr((my $key := $iterator.pull-one),IE),
                  nqp::push(
                    $output,
                    Pair.new($offset + $key, $iterator.pull-one)
                  )
                );
                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Let's go!
        self!re-shape!start(
          $k ?? &k !! $kv ?? &kv !! $p ?? &p !! &v,
          granularity($matcher),
          :slow
        )
    }

    # Handling smart-matching logic
    multi method grep(ParaSeq:D: $matcher, :$k, :$kv, :$p) {
        my uint $base;  # base offset for :k, :kv, :p

        # Logic for queuing a buffer for grep /.../
        sub v(uint $ordinal, $input is raw, $semaphore is raw) {
            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                $input.Seq.grep($matcher).iterator.push-all($output);

                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Logic for queuing a buffer for grep /.../ :k
        sub k(uint $ordinal, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            $base = $base + nqp::elems($input);

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                my $iterator := $input.Seq.grep($matcher, :k).iterator;
                nqp::until(
                  nqp::eqaddr((my $key := $iterator.pull-one),IE),
                  nqp::push($output,$offset + $key)
                );
                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Logic for queuing a buffer for grep /.../ :kv
        sub kv(uint $ordinal, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            $base = $base + nqp::elems($input);

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                my $iterator := $input.Seq.grep($matcher, :kv).iterator;
                nqp::until(
                  nqp::eqaddr((my $key := $iterator.pull-one),IE),
                  nqp::stmts(
                    nqp::push($output,$offset + $key),     # key
                    nqp::push($output,$iterator.pull-one)  # value
                  )
                );
                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Logic for queuing a buffer for grep /.../ :p
        sub p(uint $ordinal, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            $base = $base + nqp::elems($input);

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                my $iterator := $input.Seq.grep($matcher, :kv).iterator;
                nqp::until(
                  nqp::eqaddr((my $key := $iterator.pull-one),IE),
                  nqp::push(
                    $output,
                    Pair.new($offset + $key, $iterator.pull-one)
                  )
                );
                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Let's go!
        self!re-shape!start($k ?? &k !! $kv ?? &kv !! $p ?? &p !! &v)
    }

#- kv --------------------------------------------------------------------------

    multi method kv(ParaSeq:D:) {
        my uint $base;  # base offset

        # Logic for queuing a buffer for .keys
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            my uint $elems  = nqp::elems($input);
            $base = $base + $elems;

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                # Store the key values as fast as possible
                my uint $i;
                nqp::while(
                  $i < $elems,
                  nqp::stmts(
                    nqp::push($output,$offset + $i),
                    nqp::push($output,nqp::atpos($input,$i++))
                  )
                );

                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Let's go!
        self!re-shape!start(&processor)
    }

#- map -------------------------------------------------------------------------

    # Handling Callables that can have phasers
    method !mapBlock($Mapper, $method) {
        my uint $granularity = granularity($Mapper);

        my $mapper := (nqp::clone($Mapper) but BlockMapper)
          .setup($granularity, |%_);

        sub processor (
          uint $ordinal,
          uint $last,
               $input     is raw,
               $semaphore is raw,
        ) {
            self!cue: {
                my uint $then = nqp::time;

                self!batch-done(
                  $ordinal, $then, $input, $semaphore,
                  $mapper.run(
                    $ordinal == 0, $last,
                    $method, $input, nqp::getattr(self,Seq,'$!iter'),
                    $semaphore
                  )
                );
            }
        }

        # Let's go using the slower path
        self!re-shape!start(&processor, $granularity, :slow)
    }

    # Handling Callables that cannot have phasers
    method !mapCallable($mapper, str $method) {

        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            self!cue: {
                 my uint $then    = nqp::time;

                $input.List."$method"($mapper).iterator.push-all(
                  my $output := nqp::create(IB)
                );

                self!batch-done($ordinal, $then, $input, $semaphore, $output);
             }
        }

        # Let's go!
        self!re-shape!start(&processor, granularity($mapper))
    }

    multi method map(ParaSeq:D: Block:D $mapper) {
        self!mapBlock($mapper, 'map')
    }
    multi method map(ParaSeq:D: Callable:D $mapper) {
        self!mapCallable($mapper, 'map')
    }

#- max -------------------------------------------------------------------------

    # Helper method for max / min / maxpairs / minpairs
    method !limiter(
       int $direction,  # direction for limiting: 1=max, -1=min
           &by,         # comparison op
      uint $p,          # 1 = want pairs
      uint $kv?,        # 1 = want keys and values
      uint $k?,         # 1 = want keys
      uint $v?          # 1 = want values
    ) {

        # Sorry, we're lazy!
        self.fail-iterator-cannot-be-lazy('.max') if self.is-lazy;

        # Keeper for key values at start of a batch
        my uint $base;

        # Always produce .kv internally for convenience
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            my uint $elems  = nqp::elems($input);
            $base = $base + $elems;

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                # Initial limited value
                nqp::push($output,nqp::clone($offset));
                nqp::push($output,my $limit := nqp::atpos($input,0));

                my uint $i;
                my  int $cmp;
                nqp::while(
                  nqp::islt_i(++$i,$elems),  # intentionally skip first
                  nqp::if(
                    ($cmp = by(nqp::atpos($input,$i), $limit)),
                    nqp::if(                          # not Same as limit
                      nqp::iseq_i($cmp,$direction),
                      nqp::stmts(                     # right direction
                        nqp::setelems($output,0),     # clean slate
                        nqp::push($output,++$offset),
                        nqp::push($output,$limit := nqp::atpos($input,$i))
                      ),
                      ++$offset                       # wrong direction
                    ),
                    nqp::stmts(                       # Same as limit
                      nqp::push($output,++$offset),
                      nqp::push($output,nqp::atpos($input,$i))
                    )
                  )
                );

                self!batch-done(
                  $ordinal, $then, $input, $semaphore, $output
                );
            }
        }

        # Get result batches as a single result
        my $batches := self!start(&processor).IterationBuffer;
        my $final   := nqp::create(IB);

        my $key   := nqp::shift($batches);
        my $value := nqp::shift($batches);

        # Set up result handling and store first result
        my &resulter = $p
          ?? { nqp::push($final,Pair.new($key,$value)) }
          !! $kv
            ?? { nqp::push($final,$key); nqp::push($final,$value) }
            !! $k
              ?? { nqp::push($final,$key)   }
              !! { nqp::push($final,$value) }
        resulter();

        # Create final result by running over batch results
        my $limit := $value;
        my int $cmp;
        nqp::while(
          nqp::elems($batches),
          nqp::stmts(
            ($key   := nqp::shift($batches)),
            ($value := nqp::shift($batches)),
            ($cmp    = by($value, $limit)),
            nqp::if(
              $cmp,
              nqp::if(                # not Same as limit
                nqp::iseq_i($cmp,$direction),
                nqp::stmts(           # right direction
                  nqp::setelems($final,0),
                  resulter()
                )
              ),
              resulter()              # Same as limit
            )
          )
        );

        self!re-source: $final.iterator
    }

    proto method max(|) {*}
    multi method max(ParaSeq:D: &by = &[cmp], :$p, :$kv, :$k, :$v) {
        $p || $kv || $k || $v
          # Not an endpoint, use our own logic
          ?? self!limiter(1, &by, $p.Bool, $kv.Bool, $k.Bool, $v.Bool)
          # Standard endpoint .max, nothing to optimize here
          !! self.Seq.max(&by)
    }

#- maxpairs --------------------------------------------------------------------

    proto method maxpairs(|) {*}
    multi method maxpairs(ParaSeq:D: &by = &[cmp]) { self!limiter(1, &by, 1) }

#- min -------------------------------------------------------------------------

    proto method min(|) {*}
    multi method min(ParaSeq:D: &by = &[cmp], :$p, :$kv, :$k, :$v) {
        $p || $kv || $k || $v
          # Not an endpoint, use our own logic
          ?? self!limiter(-1, &by, $p.Bool, $kv.Bool, $k.Bool, $v.Bool)
          # Standard endpoint .max, nothing to optimize here
          !! self.Seq.min(&by)
    }

#- minpairs --------------------------------------------------------------------

    proto method minpairs(|) {*}
    multi method minpairs(ParaSeq:D: &by = &[cmp]) { self!limiter(-1, &by, 1) }

#- nodemap ---------------------------------------------------------------------

    multi method nodemap(ParaSeq:D: Block:D $mapper) {
        my str @phasers;
        nqp::push(@phasers,$_) if $mapper.has-phaser($_) for <FIRST NEXT LAST>;
        warn "@phasers.join(", ") phaser(s) will be ignored" if @phasers;

        my uint $granularity = granularity($mapper);
        $granularity == 1
          ?? self!mapBlock($mapper, 'nodemap', :no-first, :no-last)
          !! die "Can only handle blocks with one parameter, got $granularity"
    }
    multi method nodemap(ParaSeq:D: Callable:D $mapper) {
        self!mapCallable($mapper, 'nodemap')
    }

#- pairs -----------------------------------------------------------------------

    multi method pairs(ParaSeq:D:) {
        my uint $base;  # base offset

        # Logic for queuing a buffer for .pairs
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
            my uint $offset = $base;
            my uint $elems  = nqp::elems($input);
            $base = $base + $elems;

            self!cue: {
                my uint $then    = nqp::time;
                my      $output := nqp::create(IB);

                # Store the Pairs as fast as possible
                my uint $i;
                nqp::while(
                  $i < $elems,
                  nqp::push(
                    $output,
                    Pair.new($offset + $i,nqp::atpos($input,$i++))
                  )
                );

                self!batch-done($ordinal, $then, $input, $semaphore, $output);
            }
        }

        # Let's go!
        self!re-shape!start(&processor)
    }

#- squish ----------------------------------------------------------------------
# The squish method only uses a hypered approach if it is a non-standard one
# as the standard one is too optimized already to make the hypering overhead
# make sense

    proto method squish(|) {*}
    multi method squish(ParaSeq:D: :&as = &identity, :&with = &[===]) {

        # Set iterator to be used, default if nothing special
        my $iterator := do if nqp::eqaddr(&as,&identity)
          && nqp::eqaddr(&with,&[===]) {
            self.List.squish.iterator
        }

        # Either a non-standard "as" or "with", hypering then *does make sense
        else {

            sub processor(uint $ordinal, $input is raw, $semaphore is raw) {
                self!cue: {
                    my uint $then = nqp::time;

                    $input.Seq.squish(:&as, :&with).iterator.push-all(
                      my $batch := nqp::create(IB)
                    );

                    # Change each batch into a single element
                    nqp::push((my $output := nqp::create(IB)),$batch);

                    self!batch-done(
                      $ordinal, $then, $input, $semaphore, $output
                    );
                }
            }

            # Let's go!
            SquishIterator.new: self!start(&processor).iterator, &as, &with
        }

        self!re-source: $iterator
    }

#- unique ----------------------------------------------------------------------
# The unique method uses a 2 stage approach: the first stage is having all
# the threads produce "their" unique values, with a "seen" hash for each
# thread that was being used.  The second stage then post-processes the
# entire result, making sure that a value that was seen by more than one
# thread *still* would only be produced once, with the first value seen
# (when using a specific :as modifier)

    proto method unique(|) {*}
    multi method unique(ParaSeq:D: :with(&op)!, :&as = &identity) {
        return self.unique(:&as) if nqp::eqaddr(&op,&[===]);

        my $hashes    := nqp::create(IB);  # list of hashes, keyed on threadid
        my $lock      := Lock.new;

        # Logic for queuing a buffer for unique WITH an infix op
        sub processor(uint $ordinal, $input is raw, $semaphore is raw) {

            self!cue: {
                my uint $then =