Как я могу использовать IO :: Async с массивом в качестве входных данных?

Question

У меня есть этот цикл:

foreach my $element ( @array ) {
     my $result = doSomething($element);
}

Поскольку не имеет значения, что массив обрабатывается по порядку, а скрипт выполняется долго, я бы хотел запустить doSomething() асинхронно.

Я смотрю на IO :: Async для этого, но я не могу найти пример, где вход в цикл представляет собой простой массив, как указано выше.Кажется, что пример фокусируется на открытых сокетах, STDIN и т. Д.

Вот пример, показывающий подачу данных в цикл через STDIN:

$loop->add( IO::Async::Stream->new_for_stdin(
   on_read => sub {
      my ( $self, $buffref, $eof ) = @_;

      while( $$buffref =~ s/^(.*)\n// ) {
          print "You typed a line $1\n";
      }

      return 0;
   },
) );

Как я могу передать его массивуэлементы вместо?

Answer 1

Как прокомментировал @StefanBecker, самый простой способ справиться с этим с помощью IO :: Async - это использовать IO :: Async :: Function .

Из документов:

Этот подкласс IO :: Async :: Notifier заключает тело функции в коллекцию рабочих процессов, позволяя ему выполняться независимо от основного процесса..

В инфраструктуре IO::Async типичный вариант использования IO::Async::Function - это когда блокирующий процесс должен выполняться асинхронно.

Отказ от ответственности: обратите внимание, что, как прокомментированотакже @zdim, IO::Async может не подходить для вашего случая использования.Чистый параллелизатор процесса, такой как Parallel :: ForkManager , вероятно, был бы вашим лучшим вариантом здесь, поскольку он в основном реализует ту же функциональность (разветвление и выполнение параллельно), но в гораздо более прямой манере.Один из главных отличительных факторов IO::Async связан с его возможностями мультиплексирования ввода / вывода, которые вы, по-видимому, здесь не используете.

Но, поскольку вы именно запросили IO::Async, вот пример такой реализацииЯ превратил doSomething в фиктивный метод, который просто ожидает количество времени, указанное в качестве аргумента.Это позволяет вам наблюдать эффект асинхронного выполнения.

use strict;
use warnings;

use IO::Async::Function;
use IO::Async::Loop;
use Future;

# dummy sub
sub doSomething { 
    my ( $delay ) = @_;
    print "start waiting $delay second(s)\n";
    sleep $delay;
    print "done sleeping $delay second(s)\n"; 
    return $delay;
}

# prepare the function for execution
my $loop = IO::Async::Loop->new;
my $function = IO::Async::Function->new( code => sub { return doSomething($_[0]) } );
$loop->add($function);

# trigger asynchronous processing
my @array = qw/5 2 4 0/;
my @futures = map { $function->call( args => [ $_ ] ) } @array;

# safely wait for all ops to complete
Future->wait_all(@futures)->await;
print "all done !\n";

Это дает:

start waiting 5 second(s)
start waiting 2 second(s)
start waiting 4 second(s)
start waiting 0 second(s)
done sleeping 0 second(s)
done sleeping 2 second(s)
done sleeping 4 second(s)
done sleeping 5 second(s)
all done !

NB1: Future->wait_all(@futures)->await также можно записать $_->get for @futures, однако первое выражение,который использует конвергентные фьючерсы, имеет преимущества в том, что он никогда не потерпит неудачу, даже если основной вызов действительно умирает.

NB2: в IO::Async::Function и Future доступно много опций для обработки ошибок, управления количествомрабочие и их поведение и тд.Проверьте документы для более подробной информации ...