多路复用回调

Question

假设我在一个应用程序中有许多任务，这些任务可以以任何顺序完成。当所有任务都完成后，我需要运行一些代码。如果这很重要，那么应用程序在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

另请参阅my perlmonks question。

有像这样的东西吗？

Answer 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 => );

Answer 2

我当然不是异步方面的专家，但我认为Mojo::IOLoop::Delay ( Mojolicious套件的一部分)也有类似的接口。请注意，Mojo::IOLoop可以与EV and thus AnyEvent一起使用。

下面是来自the 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文档中显示。

编辑

这里有一个简单的例子。请注意，在delay类中刚刚发生了一个小的语法更改，所以这只适用于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

这里我们创建了delay对象，参数是一个finish事件回调。对于每个异步操作，我调用返回end回调的begin。默认情况下，这些回调去掉了它们的第一个参数来删除多余的调用(参见上面的例子)，但是我们没有这些，所以我们传递0来表明不这样做。对于每个操作，我只需使用零等待计时器进行延迟。然后，end回调的参数将按顺序排队等待end事件。塔达！