### ----------------------------------------------------
### -- ParaSeq
### -- Licenses: Artistic-2.0
### -- Authors: Elizabeth Mattijsen
### -- File: lib/ParaSeq.rakumod
### ----------------------------------------------------
### -- Chunk 3 of 3
 nqp::time;
                my      $output := nqp::create(IB);

                # Get the seen hash for this thread
                my int $i = nqp::threadid(nqp::currentthread);
                my $seen := $lock.protect: {
                    nqp::ifnull(
                      nqp::atpos($hashes,$i),
                      nqp::bindpos($hashes,$i,nqp::hash)
                    )
                }

                # Process all values
                my int $m = nqp::elems($input);
                $i = 0;
                nqp::until(
                  $i == $m,
                  nqp::unless(
                    nqp::existskey(
                      $seen,
                      (my str $key = as(
                        my $value := nqp::atpos($input,$i++)
                      ).WHICH)
                    ),
                    nqp::stmts(
                      nqp::bindkey($seen,$key,1),
                      (my int $j = 0),
                      (my int $n = nqp::elems($output)),
                      nqp::until(  # check all values for match using op(x,y)
                        $j == $n || op(nqp::atpos($output,$j), $value),
                        $j = nqp::add_i($j,2)
                      ),
                      nqp::if(
                        $j == $n,
                        nqp::stmts(  # not found with op(x,y) either, so add
                          nqp::push($output,$value),
                          nqp::push($output,nqp::clone($key))
                        )
                      )
                    )
                  )
                );

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

        # Start the process and fill buffer with initial result
        my $result := self!start(&processor).IterationBuffer;

        # Get rid of the empty slots, caused by threadids of threads that did
        # *not* run any uniquing logic
        my $seens := nqp::create(IB);
        nqp::while(
          nqp::elems($hashes),
          nqp::unless(
            nqp::isnull(my $hash := nqp::shift($hashes)),
            nqp::push($seens,$hash)
          )
        );

        # If there was only one hash with seen information, there was only
        # one thread doing it, and so we've done all the checking that we
        # need already
        if nqp::elems($seens) == 1 {
            self!re-source: WhichIterator.new($result)
        }

        # Multiple threads processed values
        else {
            # Merge all the seen hashes into a single one, with value 0 for
            # the values that were seen in only a single thread, and 1 and
            # higher for values that were seen in multiple threads
            my $seen := nqp::hash;
            nqp::while(
              nqp::elems($seens),
              nqp::stmts(
                (my $iter := nqp::iterator(nqp::shift($seens))),
                nqp::while(
                  $iter,
                  nqp::stmts(
                    (my str $key = nqp::iterkey_s(nqp::shift($iter))),
                    nqp::bindkey(
                      $seen,
                      $key,
                      nqp::add_i(nqp::ifnull(nqp::atkey($seen,$key),-1),1)
                    )
                  )
                )
              )
            );

            # Now remove all of the values that only were seen in a single
            # thread: we're sure that these only occur once in the result
            $iter := nqp::iterator($seen);
            nqp::while(
              $iter,
              nqp::unless(
                nqp::iterval(nqp::shift($iter)),
                nqp::deletekey($seen,nqp::iterkey_s($iter))
              )
            );

            self!re-source: nqp::elems($seen)
              ?? UniqueIterator.new($seen, $result)  # need further checks
              !! WhichIterator.new($result)          # all ok
        }
    }
    multi method unique(ParaSeq:D: :&as = &identity) {
        self!re-source: self.List.unique(:&as).iterator
    }

#- interfaces that are just infectious -----------------------------------------

    proto method collate(|) {*}
    multi method collate(ParaSeq:D:) {
        self!re-source: self.List.collate.iterator
    }

    proto method combinations(|) {*}
    multi method combinations(ParaSeq:D:) {
        self!re-source: self.List.combinations.iterator
    }
    multi method combinations(ParaSeq:D: $of) {
        self!re-source: self.List.combinations($of).iterator
    }

    multi method flat(ParaSeq:D:) {
        self!re-source: self.List.flat.iterator
    }

    multi method head(ParaSeq:D: $what) {
        self!re-source: self.Seq.head($what).iterator
    }

    multi method invert(ParaSeq:D:) {
        self!re-source: self.Seq.invert.iterator
    }

    multi method keys(ParaSeq:D:) {
        self!re-source: self.Seq.keys.iterator
    }

    multi method pairup(ParaSeq:D:) {
        self!re-source: self.List.pairup.iterator
    }

    multi method permutations(ParaSeq:D:) {
        self!re-source: self.List.permutations.iterator
    }

    multi method pick(ParaSeq:D: $what) {
        self!re-source: self.List.pick($what).iterator
    }

    multi method produce(ParaSeq:D: Callable:D $producer) {
        self!re-source: self.Seq.produce($producer).iterator
    }

    proto method repeated(|) {*}
    multi method repeated(ParaSeq:D: |c) {
        self!re-source: self.Seq.repeated(|c).iterator
    }

    multi method reverse(ParaSeq:D:) {
        self!re-source:
          Rakudo::Iterator.ReifiedReverse(self.IterationBuffer, Mu)
    }

    multi method roll(ParaSeq:D: $what) {
        self!re-source: self.List.roll($what).iterator
    }

    multi method rotate(ParaSeq:D: Int(Cool) $rotate) {
        $rotate
          ?? self!re-source(
               Rakudo::Iterator.ReifiedRotate($rotate, self.IterationBuffer, Mu)
             )
          !! self
    }

    proto method rotor(|) {*}
    multi method rotor(ParaSeq:D: Int:D $size, :$partial) {
        self!batch($size, $partial.Bool)                   # can be faster
    }
    multi method rotor(ParaSeq:D: |c) {
        self!re-source: self.List.rotor(|c).iterator  # nah, too difficult
    }

    proto method skip(|) {*}
    multi method skip(ParaSeq:D:) {
        self.iterator.skip-one;
        self
    }
    multi method skip(ParaSeq:D: Int(Cool) $what) {
        self.iterator.skip-at-least($what);
        self
    }
    multi method skip(ParaSeq:D: Whatever) {
        self!re-source: Rakudo::Iterator.Empty
    }
    multi method skip(ParaSeq:D: |c) {
        self!re-source: self.Seq.skip(|c).iterator
    }

    proto method slice(|) {*}
    multi method slice(ParaSeq:D: |c) {
        self!re-source: self.List.slice(|c).iterator
    }

    proto method snip(|) {*}
    multi method snip(ParaSeq:D: |c) {
        self!re-source: self.Seq.snip(|c).iterator
    }

    proto method sort(|) {*}
    multi method sort(ParaSeq:D:) {
        self!re-source: self.List.sort.iterator
    }
    multi method sort(ParaSeq:D: Callable:D $how) {
        self!re-source: self.List.sort($how).iterator
    }

    multi method tail(ParaSeq:D: $what) {
        self!re-source: self.List.tail($what).iterator
    }

    proto method toggle(|) {*}
    multi method toggle(ParaSeq:D: |c) {
        self!re-source: self.Seq.toggle(|c).iterator
    }

    multi method values(ParaSeq:D:) { self }

#- coercers --------------------------------------------------------------------

    multi method Array(  ParaSeq:D:) { self.List.Array   }
    multi method Bag(    ParaSeq:D:) { self.List.Bag     }
    multi method BagHash(ParaSeq:D:) { self.List.BagHash }

    multi method Bool(ParaSeq:D:) {
        nqp::hllbool(
          nqp::not_i(nqp::eqaddr(self.iterator.pull-one,IterationEnd))
        )
    }

    multi method eager(ParaSeq:D:) { self.List      }
    multi method Hash(ParaSeq:D: ) { self.List.Hash }

    multi method IterationBuffer(ParaSeq:D:) {
        self.iterator.push-all(my $buffer := nqp::create(IB));
        $buffer
    }

    multi method list(ParaSeq:D:) { List.from-iterator(self.iterator) }
    multi method List(ParaSeq:D:) { self.IterationBuffer.List         }

    multi method Map(    ParaSeq:D:) { self.List.Map     }
    multi method Mix(    ParaSeq:D:) { self.List.Mix     }
    multi method MixHash(ParaSeq:D:) { self.List.MixHash }

    multi method Seq(    ParaSeq:D:) { Seq.new: self.iterator    }
    multi method serial( ParaSeq:D:) { self.Seq                  }
    multi method Set(    ParaSeq:D:) { self.List.Set             }
    multi method SetHash(ParaSeq:D:) { self.List.SetHash         }
    multi method Slip(   ParaSeq:D:) { self.IterationBuffer.Slip }
    multi method Str(    ParaSeq:D:) { self.join(" ")            }
    multi method Supply( ParaSeq:D:) { self.List.Supply          }

#- hyper -----------------------------------------------------------------------

    proto method hyper(|) {*}

    # If the degree is 1, then just return the iterable, nothing to parallelize
    multi method hyper(ParaSeq: \iterable,$,$,$,$, 1) is implementation-detail {
        iterable
    }

    # Entry point from the subs
    multi method hyper(ParaSeq:U:
      \iterable,
      $racing,
      $batch  is copy,
      $auto   is copy,
      $catch  is copy,
      $degree is copy,
    ) is implementation-detail {
        $batch := ($batch // $default-batch).Int;
        X::Invalid::Value.new(
          :method<hyper>, :name<batch>,  :value($batch)
        ).throw if $batch <= 0;

        $degree := ($degree // $default-degree).Int;
        X::Invalid::Value.new(
          :method<hyper>, :name<degree>, :value($degree)
        ).throw if $degree <= 1;

        my $iterator := iterable.iterator;
        my $buffer   := nqp::create(IB);

        # First batch already exhausted iterator, so work with buffer in
        # a non-parallelized way
        if nqp::eqaddr($iterator.push-exactly($buffer, $batch),IE) {
            $buffer.Seq
        }

        # Need to actually parallelize, set up ParaSeq object
        else {
            $auto  := ($auto  // True).Bool;
            $catch := ($catch // True).Bool;
            self!setup(
              BufferIterator.new($buffer, $iterator),
              $racing, $batch, $auto, $catch, $degree
            )
        }
    }

    # Change hypering settings along the way
    multi method hyper(ParaSeq:D:
      $batch?, $degree?, :$auto, :$catch, :$racing
    ) {
        $degree && $degree == 1
          ?? self.serial     # re-hypering with degree == 1 -> serialize
          !! ParaSeq.hyper(  # restart, taking over defaults
               self,
               $racing // $!racing,
               $batch  // $!batch,
               $auto   // $!auto,
               $catch  // $!catch,
               $degree // $!degree
             )
    }
}

#- actual interface ------------------------------------------------------------

proto sub hyperize(|) is export {*}
multi sub hyperize(\iterable, $batch?, $degree?, :$auto, :$catch) {
    ParaSeq.hyper(iterable, 0, $batch, $auto, $catch, $degree)
}
multi sub hyperize(List:D $list, $size?, $degree?, :$auto, :$catch) {
    my uint $batch = $size // $default-batch;
    $list.is-lazy || $list.elems > $batch
      ?? ParaSeq.hyper($list, 0, $batch, $auto, $catch, $degree)
      !! $list
}

proto sub racify(|) is export {*}
multi sub racify(\iterable, $batch?, $degree?, :$auto, :$catch) {
    ParaSeq.hyper(iterable, 1, $batch, $auto, $catch, $degree)
}
multi sub racify(List:D $list, $size?, $degree?, :$auto, :$catch) {
    my uint $batch = $size // $default-batch;
    $list.is-lazy || $list.elems > $batch
      ?? ParaSeq.hyper($list, 1, $batch, $auto, $catch, $degree)
      !! $list
}

# vim: expandtab shiftwidth=4


