AttackCoro.pm

package AttackCoro;
use 5.10.1;
use strict;
use warnings;
use base "Exporter";
our @EXPORT = qw/
watcher coro pool sleep
/;
our @EXPORT_OK = @EXPORT;
use Coro;
use Coro::AnyEvent;
use Coro::LWP;
use AnyEvent;
sub sleep($)
{
my($s) = @_;
Coro::AnyEvent::sleep($s);
return;
}
sub watcher()
{
return AnyEvent->timer(
interval => 2,
cb => sub {
my $time = time;
map {
say "Attack: thread #$_->{id} ('$_->{desc}') terminated."
if $_->{debug};
$_->cancel(undef);
} grep {
$_->{timeout_at} && $time >= $_->{timeout_at}
} reverse Coro::State::list;
},
);
}
sub coro($$;$)
{
my($sub, $arg, $options_) = @_;
$options_ ||= {};
my $options = {
debug => 0,
desc => "anon",
timeout => 0,
eval => 0,
ready => 0,
join => 0,
%$options_,
};
state $id = 1;
my $coro; $coro = Coro->new(sub {
$coro->{id} = $id++;
$coro->{desc} = $options->{desc};
$coro->{timeout_at} = time + $options->{timeout}
if $options->{timeout};
$coro->{debug} = $options->{debug};
say "Attack: thread #$coro->{id} ('$coro->{desc}') started."
if $options->{debug};
my $result;
if($options->{eval}) {
eval { $result = $sub->($arg) };
if($@ && $options->{debug}) {
chomp $@;
say "Attack: thread #$coro->{id} ('$coro->{desc}') died: $@";
}
} else {
$result = $sub->($arg);
}
say "Attack: thread #$coro->{id} ('$coro->{desc}') finished."
if $options->{debug};
return $result // undef;
});
$coro->ready if $options->{ready} || $options->{join};
return $options->{join} ? $coro->join : $coro;
}
sub pool($$;$)
{
my($sub, $args, $options_) = @_;
$options_ ||= {};
my $options = {
debug => 0,
desc => "anon",
ready => 0,
join => 0,
limit => 0,
%$options_,
};
say "Attack: create pool '$options->{desc}' with ".@$args." threads."
if $options->{debug};
my $pool = async
{
my @results;
while(my @args_ = ($options->{limit} > 0 ? splice @$args, 0, $options->{limit} : splice @$args))
{
my @coros = map {
coro($sub, $_, {
%$options,
desc => "$options->{desc} pool",
ready => 0,
join => 0,
});
} @args_;
push @results, map { $_->join } map { $_->ready; $_ } @coros;
}
return @results;
};
$pool->ready if $options->{ready} || $options->{join};
return $options->{join} ? $pool->join : $pool;
}
2;