Mojolicous：限制 Promises / IOLoop->subprocess 的数量

Question

我正在使用 Mojolicious 非阻塞方法 (Promises) 从外部系统请求数据。 1）我想立即通知用户该过程已经开始； 2) 我想扩展这个程序。

下面的代码适用于一小部分数字（几百个），如果数字更多，我会收到错误[error] Can't create pipe: Too many open files at /path/lib/perl5/Mojo/IOLoop.pm line 156.问题 1）我如何限制我生成的 Promise 的数量（ map 在我下面的代码中）：

#!/usr/bin/env perl

use Mojolicious::Lite;
use Mojolicious::Plugin::TtRenderer;

sub isPrime
{
    my ($n) = @_;
    my $e = sqrt($n);
    for (my $i=2; $i<$e; $i++) {
        return 0 if $n%$i==0;
    }
    return 1;
}

sub makeApromise
{
    my ($number) = @_;

    my $promise = Mojo::Promise->new;
    Mojo::IOLoop->subprocess(
    sub {  # first callback is executed in subprocess
        my %response;
        # Simulate a long computational process
        $response{'number'}  = $number;
        $response{'isPrime'} = isPrime($number);
        return \%response;
    },
        sub {  # second callback resolves promise with subprocess result
            my ($self, $err, @result) = @_;
            return $promise->reject($err) if $err;
            $promise->resolve(@result);
        },
    );
    return $promise;
}

plugin 'tt_renderer'; # automatically render *.html.tt templates

any '/' => sub {
    my ($self) = @_;
    my $lines = $self->param( 'textarea' );

    if ($lines) {
    my @numbers;
    foreach my $number (split(/\r?\n/, $lines)) {
        push(@numbers, $number) if $number =~ /^\d+$/;
    }
    if (@numbers) {
        ####################################
        ### This is the problem below... ###
        my @promises = map { makeApromise($_) } @numbers;
        ####################################
        # MojoPromise Wait
        Mojo::Promise->all(@promises)
        ->then(sub {
            my @values = map { $_->[0] } @_;
            foreach my $response (@values) {
            #print STDERR $response->{'number'}, " => ", $response->{'isPrime'}, "\n";
            # Prepare email...
            }
            # Send an email...
               })
        #->wait # Don't wait? I want to tell the user to wait for an email as quickly as possible...
        if @promises;
    }
    $self->stash(done => "1",);
    }
    $self->render(template => 'index', format => 'html', handler => 'tt');
};

app->start;
__DATA__

@@ index.html.tt
<!DOCTYPE html>
<html lang="en">
  <head>
    <title>Make A Promise</title>
  </head>
  <body>
    [% IF done %]
    <h3>Thank you! You will receive an email shortly with the results.</h3>
    [% ELSE %]
    <h3>Enter numbers...</h3>
    <form role="form" action="/" method="post">
      <textarea name="textarea" rows="5" autofocus required></textarea>
      <button type="submit">Submit</button>
    </form>
    [% END %]
  </body>
</html>

我注释掉了wait;但是，代码似乎仍然处于阻塞状态。 问题 2) 我怎样才能立即通知用户该过程已经开始？ （即当我stash done 变量时）

Answer 1

问题不在于承诺的数量，而在于子流程的数量。限制这种情况的一种方法是简单地限制您在程序逻辑中一次创建的数量。与其在地图中一次生成它们，不如设置一个限制并从@numbers 中检索那么多（可能使用splice）并生成这些子进程；创建一个等待它们的 ->all 承诺，并将 ->then 附加到该承诺以检索你的下一个数字块，依此类推。

另一种选择是使用Future::Utils fmap_concat，它可以通过让您提供一些最大未完成的期货来处理速率限制代码。您的 promise-returning 函数可以应用 Mojo::Promise::Role::Futurify 来链接后续 Future 以以这种方式使用。

#!/usr/bin/env perl

use Mojolicious::Lite;
use Mojo::File 'path';
use Mojo::IOLoop;
use Mojo::Promise;
use Future::Utils 'fmap_concat';

get '/' => sub {
  my $c = shift;
  my $count = $c->param('count') // 0;
  my @numbers = 1..$count;

  if (@numbers) {
    my $result_f = fmap_concat {
      my $number = shift;
      my $p = Mojo::Promise->new;
      Mojo::IOLoop->subprocess(sub {
        sleep 2;
        return $number+1;
      }, sub {
        my ($subprocess, $err, @result) = @_;
        return $p->reject($err) if $err;
        $p->resolve(@result);
      });
      return $p->with_roles('Mojo::Promise::Role::Futurify')->futurify;
    } foreach => \@numbers, concurrent => 20;

    $result_f->on_done(sub {
      my @values = @_;
      foreach my $response (@values) {
        $c->app->log->info($response);
      }
    })->on_fail(sub {
      my $error = shift;
      $c->app->log->fatal($error);
    })->retain;

    $c->stash(done => 1);
  }
  $c->render(text => "Processing $count numbers\n");
};

app->start;

至于等待方法，当事件循环已经在运行时，它什么也不做，如果你在 Mojolicious 守护进程中启动应用程序（而不是 PSGI 或 CGI 服务器不'不支持异步响应）。回调之外的 ->stash 和 ->render 调用将在设置子进程后立即运行。然后响应处理程序将完成，并且事件循环将再次获得控制权，一旦承诺解决，它将触发适当的 ->then 回调。渲染不应该等待子进程设置之外的任何事情；既然你说可能有数百个，那可能是你正在经历的放缓。确保您使用的是 Mojolicious 7.86 或更高版本，因为 Subprocess 已更改，因此在事件循环的下一个滴答声（在您的响应处理程序完成之后）之前不会发生分叉。

我还要注意，子流程并不是真正为此设计的；它们被设计用于执行慢速代码，仍然在响应中向浏览器返回最终结果（Mojolicious::Plugin::Subprocess 非常适合这个用例）。我可以看到的一个问题是，如果您重新启动应用程序，任何仍然挂起的子进程都将被忽略。对于您想要启动并忘记的作业，您可以考虑使用像 Minion 这样的作业队列，它可以很好地集成到 Mojolicious 应用程序中，并通过单独的工作进程运行。