use v6.d;
unit class Async::Workers:ver<0.2.2>:auth<zef:vrurg>:api<0.2.0>;
also does Awaitable;

use Async::Workers::Msg;
use Async::Workers::Job;
use Async::Workers::Worker;
use Async::Workers::X;
use Async::Workers::CX;
use AttrX::Mooish;

my $singleton;

has Channel $!queue is mooish( :lazy, :clearer );
has Lock::Async $!ql .= new;
# Queue lock
has Promise:D $!queue-unblock .= kept;

has UInt $.max-workers = $*KERNEL.cpu-cores;
has UInt $.lo-threshold;
has UInt $.hi-threshold;
has atomicint $.queued = 0;
has atomicint $.running = 0;
has UInt $.max-jobs where { !.defined || $_ > 0 };

has Bool:D $.quiet = False;

has %!workers;
has Lock::Async $!wl .= new;
# Workers lock

has Promise $.completed;
has $!completion-vow;
has Promise $.empty .= new;
has $!empty-vow = $!empty.vow;
has Promise $!shutdown .= new;

has Supplier $!messages .= new;

has atomicint $!active-workers = 0;

has $.debug = ?%*ENV<ASYNC_WORKERS_DEBUG>;

sub stop-worker( $rc?, Bool:D :$soft = False ) is export {
    with $*AW-WORKER {
        .stop: $rc, :$soft
    }
    else {
        X::NoWorker.new(:helper<stop-worker>).throw
    }
}

sub async-workers( |c ) is export {
    $singleton //= Async::Workers.new(|c)
}

sub worker-manager( |c ) is export {
    $singleton //= Async::Workers.new(|c)
}

sub do-async( |c ) is export {
    worker-manager.do-async(|c)
}

sub shutdown-workers is export {
    .shutdown with $singleton;
}

my proto as-soon( | ) {*}
multi as-soon( Promise:D $promise, Pair:D $event ) {
    my $mgr = CALLER::LEXICAL::<self>;
    my $what = $event.key;
    my &code = $event.value;
    $promise.then: {
        CATCH {
            CATCH { note "==

=== Died in as-soon CATCH:\n", .message, "\n", .backtrace.Str }
            $mgr.message: Async::Workers::Msg::Exception, exception => $_;
            $mgr.x-sorry($_, $what ~ " promise handler");
            $mgr.bail-out: $_;
            .rethrow
        }
        &code( $_ )
    }
}


INTERNAL


multi as-soon( Promise:D $promise, &code ) {
    as-soon($promise, "A" => &code)
}

submethod TWEAK( | ) {
    $!lo-threshold //= $!max-workers;
    die "High queue threshold ($!hi-threshold) can't be lower than the low ($!lo-threshold)"
    if $!lo-threshold > ( $!hi-threshold // Inf );
}

method !build-queue {
    # For Attrx::Mooish
    $!completion-vow = ( $!completed = Promise.new ).vow;
    as-soon $!completed, "Manager completion" => { self.message: Async::Workers::Msg::Complete };
    $!shutdown = Promise.new;
    Channel.new
}

method do-async ( &code, |args --> Async::Workers::Job ) {
    with $!queue {
        if $!completed.status == Planned && $!shutdown.status == Planned {
            self!start-workers;

            $!ql.lock;
            LEAVE $!ql.unlock;

            my $size = ++⚛$!queued;
            self.message: Async::Workers::Msg::Queue::Inc, :$size;

            with $.hi-threshold -> $ht {
                if $size == $ht {
                    self.message: Msg::Queue::Full, :$size;
                }
                if $size > $ht {
                    my $qu = $!queue-unblock;
                    cas($!queue-unblock, $qu, Promise.new);
                    await $!queue-unblock;
                }
            }
            self!queue-job: &code, args
        }
        else {
            self!throw: X::Manager::Down
        }
    }
    else {
        self!throw: X::Manager::NoQueue
    }
}

method !queue-job( &code, Capture:D $args --> Async::Workers::Job ) {
    my $job = Async::Workers::Job.new(:&code, :$args, :manager( self ));
    $!queue.send: $job;
    $job
}

method !job-invoke( Async::Workers::Job:D $job ) {
    ++⚛$!running;
    self.message: Async::Workers::Msg::Job::Enter, :$job;
}

method !job-complete( Async::Workers::Job:D $job ) {
    --⚛$!running;
    if $job.completed.status == Broken {
        my $exception = $job.completed.cause;
        self.message: Async::Workers::Msg::Job::Died, :$job, :$exception;
        self.x-sorry: $exception, "JOB #" ~ $job.id unless $!quiet;
    }
    self.message: Async::Workers::Msg::Job::Complete, :$job;
    self!dec-queue;
}

proto method message( | ) {*}
multi method message( Msg:U \msg, *%profile ) {
    self.message: msg.new(|%profile)
}
multi method message( Msg:D $msg ) {
    $!messages.emit: $msg
}

method !throw( Exception \ex, |c ) {
    ex.new(|c).throw
}

method !start-workers {
    return unless $!shutdown.status == Planned;
    while $!active-workers < $!max-workers {
        my $need-one = False;
        cas $!active-workers, {
            ( $need-one = $_ < $!max-workers ) ?? $_ + 1 !! $_
        }

        if $need-one {
            # Worker callback, simply redispatches to a private method.
            my sub wcb( Str:D $method, |c ) {
                self!"$method"(|c)
            }

            my $worker = Async::Workers::Worker.new: :manager( self ), :$!queue, :callback( &wcb );

            # Whatever is the reason for a worker to stop – try starting a new one again. Even if it's a part of
            # reducing $!max-workers number procedure, then the start-workers method would just skip and do nothing.
            as-soon $worker.run, "Worker #" ~ $worker.id ~ " completion" => {
                if .status == Broken {
                    self.x-sorry(.cause, "Worker #" ~ $worker.id);
                    self!deregister-worker($worker)
                }
            };
        }
    }
}

method !register-worker( Async::Workers::Worker:D $worker ) {
    self.message: Async::Workers::Msg::Worker::Started, :$worker;
    $!wl.protect: { %!workers{$worker.id} = $worker };
}

method !deregister-worker( Async::Workers::Worker:D $worker ) {
    my $active = --⚛$!active-workers;
    self.message: Async::Workers::Msg::Worker::Complete, :$worker;
    $!wl.protect: { %!workers{$worker.id}:delete };
    if $!shutdown.status != Planned {
        if $active == 0 && $!completed.status == Planned {
            $!completion-vow.keep(True);
            self!clear-queue;
        }
    }
    else {
        self!start-workers
    }
}

method shutdown {
    $!shutdown.keep;
    $!queue.close;
    self.message: Async::Workers::Msg::Shutdown;
}

method bail-out( Exception:D $exception ) {
    return if $!queue.closed;
    $!shutdown.keep;
    $!queue.close;
    $!completion-vow.break($exception);
}

my class AsyncWorkersAwaitHandle does Awaitable::Handle {
    has &!add-subscriber;
    method not-ready ( &add-subscriber ) {
        use nqp;
        nqp::create(self)!not-ready(&add-subscriber);
    }
    method !not-ready ( &add-subscriber ) {
        $!already = False;
        &!add-subscriber := &add-subscriber;
        self
    }
    method subscribe-awaiter ( &subscriber --> Nil ) {
        &!add-subscriber( &subscriber );
    }
}

method get-await-handle ( --> Awaitable::Handle:D ) {
    if $!queued || $!running {
        AsyncWorkersAwaitHandle.not-ready: -> &on-ready {
            my $tap = $!messages.Supply.tap: -> $msg {
                if $msg ~~ Async::Workers::Msg::Queue::Empty {
                    $tap.close;
                    on-ready(True, Nil);
                }
            }
            # Re-check queue status because we could've missed a message while preparing the tap.
            unless $!queued || $!running {
                $tap.close;
                on-ready(True, Nil);
            }
        }
    }
    else {
        AsyncWorkersAwaitHandle.already-success(Nil);
    }
}

multi method set-threshold( UInt $lo, Num $hi ) {
    self.set-threshold(:$lo, :$hi)
}
multi method set-threshold( UInt :$lo, Num :$hi ) {
    $!lo-threshold = $_ with $lo;
    $!hi-threshold = !.defined || $_ == Inf ?? Nil !! .Int with $hi;
    self.throw: Async::Workers::X::Threshold, lo => $!lo-threshold, hi => $!hi-threshold
    if defined($!lo-threshold & $!hi-threshold) && $!lo-threshold > $!hi-threshold;
}

multi method workers( UInt:D $max where * > 0 ) {
    my $old-max;
    loop {
        $old-max = $!max-workers;
        last if cas($!max-workers, $old-max, $max) == $old-max;
    }
    return if $max == $old-max;

    if $max < $old-max {
        for ^( $old-max - $max ) {
            # This would stop the necessary number of workers
            self.do-async: { stop-worker }
        }
    }
    else {
        self!start-workers;
    }
}

multi method workers ( --> UInt ) {
    $!active-workers
}

method on_msg ( &code ) {
    $!messages.Supply.tap: &code;
}

method x-sorry( Exception:D $exception, Str $what ) {
    my @sorry = ( '==

!===' );
    @sorry.push: $_ with $what;
    #    @sorry.push: "thrown " ~ $exception.^name ~ ":";
    given $exception {
        note
            @sorry.join(" "), "\n",
            ( "[" ~ .^name ~ "] " ~ .message ).indent(2), "\n",
            .backtrace.Str.indent(2)
    }
}


SORRY


method !dec-queue {
    my $size = --⚛$!queued;
    if $size == $!lo-threshold {
        self.message: Async::Workers::Msg::Queue::Low, :$size;
        $!queue-unblock.keep if $!queue-unblock.status == Planned;
    }
    if $size == 0 {
        self.message: Async::Workers::Msg::Queue::Empty, :$size;
        my $oempty = $!empty;
        if cas($!empty, $oempty, Promise.new) === $oempty {
            $!empty-vow.keep: True;
            $!empty-vow = $!empty.vow;
        }
    }
}

our sub META6 {
	use META6;
    name           => 'Async::Workers',
    description    => 'Asynchronous threaded workers',
    version        => Async::Workers.^ver,
	api			   => Async::Workers.^api,
	auth		   => Async::Workers.^auth,
    perl-version   => Version.new('6.d'),
    rake-version   => Version.new('6.d'),
    depends        => [<AttrX::Mooish>],
    test-depends   => <Test Test::Async Test::META Test::When>,
    tags           => <threads async>,
    authors        => ['Vadim Belman'],
    source-url     => 'https://github.com/vrurg/raku-Async-Workers.git',
    support        => META6::Support.new(
        source => 'https://github.com/vrurg/raku-Async-Workers.git',
    ),
    provides => {
        'Async::Workers' => 'lib/Async/Workers.rakumod',
        'Async::Workers::CX' => 'lib/Async/Workers/CX.rakumod',
        'Async::Workers::Job' => 'lib/Async/Workers/Job.rakumod',
        'Async::Workers::Msg' => 'lib/Async/Workers/Msg.rakumod',
        'Async::Workers::Worker' => 'lib/Async/Workers/Worker.rakumod',
        'Async::Workers::X' => 'lib/Async/Workers/X.rakumod',
    },
    license        => 'Artistic-2.0',
    production     => True,
}

# Copyright (c) 2019, Vadim Belman <vrurg@cpan.org>
# vim: ft=perl6


root

README

Workers

React

=begin React :component<HeaderCol> :id<menu> 
=para L<Documentation|/doc/introduction>
=para L<Modules| /mods>
=para L<Examples| file:../examples/index.podlite>
=para L<Download |file:../download/index.podlite>
=para L<About|file:../about/index.podlite>
=para 🔍 K<⌘K>/K<ctrl-K>
=end React


=begin React :component<HeaderCol> :id<footer>  
=begin nested :!nested
=item1 B<Language>
    =item2 L<Get started|file:../getting-started/getting-started.podlite>
    =item2 L<Why Raku?|/doc/language/faq#Why-should-I-learn-Raku-What's-so-great-about-it>
    =item2 L<Try Raku|https://glot.io/new/raku>
    =item2 L<Raku cheat sheet|https://github.com/Raku/mu/blob/master/docs/Perl6/Cheatsheet/cheatsheet.txt>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<Raku on Exercism|https://exercism.org/tracks/raku>
    =item2 L<Wikipedia|https://en.wikipedia.org/wiki/Raku_(programming_language)>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>

=end nested
=begin nested :!nested

=item1 B<L<Documentation|/doc>>
    =item2 L<Getting started, Migration guides from other languages, & Tutorials | file:../doc/introduction.podlite>
    =item2 L<Language References|file:../doc/reference.podlite>
    =item2 L<Type Reference| file:../doc/types.podlite>
    =item2 L<Miscellaneous| file:../doc/miscellaneous.podlite>
    =item2 L<FAQs (Frequently Asked Questions)|/doc/language/faq>
    =item2 L<Community|/doc/language/community>
    =item2 L<The list of all documents|file:../doc/index.podlite>

=item1 B<Resources>
    =item2 L<The Raku Guide|https://raku.guide/>
    =item2 L<Books |https://perl6book.com/>
    =item2 L<Rosetta Code | https://www.raku.org/community/rosettacode>

=end nested
=begin nested :!nested

=item1 B<Resoures>
    =item2 L<Download | file:../download/index.podlite>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>

=item1 B<Learning>
    =item2 L<The Raku Guide|https://raku.guide/>
    =item2 L<Wikibook |https://en.wikibooks.org/wiki/Raku_Programming>
    =item2 L<Books |https://perl6book.com/>
    =item2 L<Rosetta Code | https://www.raku.org/community/rosettacode>
    =item2 L<Learn Raku in Y minutes|https://learnxinyminutes.com/docs/raku/>
    =item2 L<Golfing |https://github.com/AlexDaniel/raku-golf-cheatsheet>
=end nested
=begin nested :!nested

=item1 B<Explore>
    =item2 L<Raku Blog Aggregator|https://planet.raku.org/>
    =item2 L<Rakudo Weekly|https://rakudoweekly.blog/>
    =item2 L<The Weekly Challenge |https://perlweeklychallenge.org/>
    =item2 L<Raku Advent Calendar|https://raku-advent.blog/>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>
    =item2 L<empty|#>

=end nested
=end React


=begin React :component<CookieConsent> :id<CookieConsent> :buttonCaption("Got it!")
=para
This website uses cookies for analytics.
=end React


Workers

INTERNAL

SORRY

Async::Workers v0.2.2

Authors

License

Dependencies

Test Dependencies

Provides

Documentation