CHANGELOG

head

Migrate to build-tools and zef ecosystem


root

para

L<Examples| file:../examples/index.podlite>


L<Download |file:../download/index.podlite>


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


HEADER

FOOTER

ChangeLog

code

item

Html

Workers

C<Async::Workers> - Asynchronous threaded workers


my $wm = Async::Workers.new( :max-workers(5) );


for 1..10 -> $n {
    $wm.do-async: {
        sleep 1.rand;
        say "Worker #$n";
    }
}


This module provides an easy way to execute a number of jobs in parallel while allowing to keep the resources consumed
by your code under control.


Both OO and procedural interfaces are provided.


This module has been tested by running 20k repetitions of it's test suite using 56 parallel processes. This doesn't
prove there're no bugs, but what can be told for certain is that within the tested scenarios the robustness is high
enough.


The goal is achieved by combining a I<queue of jobs> and a number of pre-spawned threads controlled by I<workers>. A
job is picked from the queue by a currently unoccupied worker and the code object associated with it is invoked. Since
the number of workers is limited it is easier to predict and plan CPU usage.


Yet, it is still possible to over-use memory and cause heavy swapping or even overflows in cases when there is too many
jobs are produced and an average one takes longer to complete than it is needed to generate a new one. To provide some
control over this situation one can define I<hi> and I<lo> thresholds on the queue size. When the queue contains as many
jobs, as defined  by the I<hi> threshold, C<do-async> method blocks upon receiving a new job request and unblocks only
when the queue shortens down to its I<lo> threshold.


The worker manager doesn't start workers until the first job is sent to the queue. It is also possible to shutdown all
workers if they're no longer needed.


Some internal events are reported with messages from C<Async::Workers::Msg>. See the C<on_msg> method below.


Maximum number of workers. Defaults to C<$*KERNEL.cpu-cores>.


Set the maximum number of jobs a worker should process before stopping and letting the manager to spawn a new one. The
functionality is not activated if the attribute is left undefined.


Low and high thresholds of the queue size.


Current queue size. If the queue has been blocked due to reaching C<hi-threshold> then jobs awaiting for unblock are not
counted toward this value.


The number of currently occupied workers.


A C<Promise> which is kept when manager completes all jobs after transitioning into I<shutdown> state. When this happens
the job queue is closed and all workers are requested to stop. Submission of a new job with C<do-async> at this point will
re-vivify the queue and return the manager into working state.


In case of an internal failure the promise will be broken with an exception.


A C<Promise> which is kept each time the queue gets emptied. Note that the initially empty queue is not reflected with
this attribute. Only when the queue contained at least one element and then went down to zero length this promise is
kept. In other words, it happens when C<Async::Workers::Msg::Queue::Empty> is emitted.


Immediately after being kept the attribute gets replaced with a fresh C<Promise>. So that the following example will
finish only if the queue has been emptied twice:


If set to I<True> then no exceptions thrown by jobs are reported. In this case it is recommended to monitor messages
for C<Async::Workers::Msg::Job::Died>.


C<<do-async( &code, |params --> Async::Workers::Job )>>


Takes a C<&code> object and wraps it into a job object. C<params> are passed to C<&code> when it gets executed.


This method blocks if C<hi-threshold> is defined and the queue size has reached the limit.


If no error happens then the method returns an C<Async::Workers::Job> instance. Otherwise it may throw either
C<X::Manager::Down> if the manager is in C<shutdown> or C<completed> status; or it may throw C<X::Manager::NoQueue> if
somehow the job queue has not been initialized properly.


Switches the manager into I<shutdown> state and closes the job queue. Since the queue might still contain some
incomplete jobs it is likely to take some time until the C<completed> promise gets kept. Normally it'd be helpful
to C<await> for the manager:


my $wm = Async::Workers.new(...);
...
$wm.shutdown;
await $wm;


In this case the execution blocks until the job queue is emptied. Note that at this point C<completed> might still not
been fulfilled because workers are being shutting down in the meanwhile.


The Raku Knowledge Base


=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


ChangeLog

CHANGELOG

v0.2.2

Async::Workers v0.2.2

Authors

License

Dependencies

Test Dependencies

Provides

Documentation