Обратные вызовы мультиплексирования

Предположим, что у меня есть несколько задач в одном приложении, которое может быть завершено в любом порядке. И мне нужно запустить код, когда все задачи будут завершены. Если это имеет значение, приложение работает под AnyEvent, но без Coro.

В некоторой степени AnyEvent $cv->begin/$cv->end разрешить то, что я хочу. Тем не менее, я хотел бы иметь более мелкомасштабный контроль. Например, я бы не смог дважды завершить задачу. Также было бы полезно собрать данные из всех задач.

Конечно, это можно сделать. Настройте множество обратных вызовов, которые разделяют хэш; удалять ключи из этого хеша всякий раз, когда заканчивается задание; вызовите megacallback, когда хеш пуст. Интересно, есть ли более цивилизованный способ сделать это, может быть, какой-то модуль CPAN?

Например, вот мнимый API, который заполнит мою потребность.

#!/usr/bin/perl -w 
use strict;

use Some::Module;

# Set goals
my $cb = Some::Module->new( sub { say 'BOOM!' } );
$cb->begin( qw(foo bar) );

# Much later, as tasks start getting done
$cb->end( foo => 42 );       # "return" value from task 'foo'
$cb->begin( 'baz' );         # can add more tasks, why not
$cb->end( 'bar' );           # just finish task 'bar'
# still waiting for 'baz' to finish at this point

# Finally, last hanging task is done
$cb->end( baz => 137 );      # BOOM!
# at this point, sub {}->( { foo=>42, bar=>undef, baz=>137 } ) 
#     has been called

См. также мой вопрос perlmonks.

Это что-то вроде этого?

Ответы

Ответ 1

Возможно, вы захотите рассмотреть Future.

В частности, для ожидания завершения многих действий вы можете использовать Future->needs_all или подобное:

my @things = ... # generate Futures to represent each thing

Future->needs_all( @things )
  ->on_done( sub {
     # some code here when all the things are done 
  });

В качестве альтернативы вы также можете попробовать Async:: MergePoint, который дает API намного ближе к тому, что вы имели в виду:

my $mp = Async::MergePoint->new( needs => [qw( foo bar )] );
$mp->close( on_done => sub {
   # some code here when all the things are done
});

$mp->done( foo => $value );
$mp->done( bar => );

Ответ 2

Я, конечно, не эксперт в асинхронном материале, но я думаю, что Mojo:: IOLoop:: Delay (часть Mojolicious suite) имеет аналогичный интерфейс. Обратите внимание, что Mojo:: IOLoop можно использовать с EV и, следовательно, AnyEvent.

Вот пример из cookbook:

use Mojo::UserAgent;
use Mojo::IOLoop;

# Synchronize non-blocking requests portably
my $ua    = Mojo::UserAgent->new;
my $delay = Mojo::IOLoop->delay(sub {
  my ($delay, $tx, $tx2) = @_;
  ...
});
$ua->get('http://mojolicio.us'         => $delay->begin);
$ua->get('http://mojolicio.us/perldoc' => $delay->begin);
$delay->wait unless Mojo::IOLoop->is_running;

Также обратите внимание, что $delay->begin возвращает обратный вызов, который по существу является методом end.

Другие примеры, такие как классная концепция "шагов", показаны в документации:: Delay.

ИЗМЕНИТЬ

Вот краткий пример. Обратите внимание, что небольшое изменение синтаксиса JUST произошло в классе задержки, поэтому это работает только с Mojolicious 3.93+, а не с тем, что это было невозможно раньше, но синтаксис несколько отличался.

#!/usr/bin/env perl

use strict;
use warnings;
use v5.10;

use Mojo::IOLoop;

my $delay = Mojo::IOLoop->delay(sub{shift; say for @_});

my $end_foo = $delay->begin(0);
Mojo::IOLoop->timer( 0 => sub { $end_foo->('hi') } );

my $end_bar = $delay->begin(0);
Mojo::IOLoop->timer( 0 => sub { $end_bar->('bye') } );

$delay->wait unless $delay->ioloop->is_running; #start loop if necessary

Здесь мы создаем объект задержки, аргументом является обратный вызов события finish. Для каждого асинхронного действия я вызываю begin, который возвращает обратный вызов end. По умолчанию эти обратные вызовы делят свой первый аргумент на удаление избыточного invocant (см. Пример выше), но у нас их нет, поэтому мы передаем 0, чтобы указать на то, чтобы этого не делать. Для каждого действия я просто откладываю таймер с нулевым ожиданием. Аргументы до конца обратного вызова затем ставятся в очередь для конечного события по порядку. Тада!