multithreading - 莫霍利奇 : Limiting number of Promises/IOLoop->subprocess-6ren

multithreading - 莫霍利奇 : Limiting number of Promises/IOLoop->subprocess

转载作者：行者123 更新时间：2023-12-02 00:58:32

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

下面的代码适用于一小部分数字(几百个)，如果数字更多，我会得到一个错误 [error] Can't create pipe: Too many open files at/path/lib/perl5/Mojo/IOLoop.pm 第 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 变量时)

最佳答案

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

另一种选择是使用 Future::Utils fmap_concat 可以通过提供最大未完成 future 数量来处理限速代码。您的 promise 返回功能可以应用 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;

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

我还要指出，子流程并不是真正为此而设计的；它们旨在执行缓慢的代码，这些代码仍会在响应中向浏览器返回最终结果(并且 Mojolicious::Plugin::Subprocess 非常适合此用例)。我看到的一个问题是，如果您重新启动应用程序，任何仍在挂起的子进程都将被忽略。对于您想要开始并忘记的工作，您可以考虑像 Minion 这样的工作队列。它与 Mojolicious 应用程序完美集成，并通过单独的工作进程运行。

关于multithreading - 莫霍利奇 : Limiting number of Promises/IOLoop->subprocess，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/52393335/

文章推荐： scala - 理解 Flink - 任务不可序列化

文章推荐： numpy - 如何理解 np.argwhere 函数？

文章推荐： Angular 响应式(Reactive)自定义验证器不起作用

opencv - 奇/偶迭代是什么意思？
http://opencv-code.com/quick-tips/implementation-of-guo-hall-thinning-algorithm/ 在看郭霍尔算法，但我不明白它说的那部分
java - 原始递归偶/奇 - 它到底做了什么？
我知道这个方法检查整数是否为偶数，但是具体如何检查呢？我理解斐波那契或阶乘等例子的递归是如何工作的，但不理解这个。我认为由于语法原因我不明白。 // Assume n >= 0 public stat
java - 奇/偶输出程序仅给出 1 个答案
我被困在一个作业上，需要程序接受多个数字，然后如果是奇数则输出"is"，如果是偶数则输出“否”，并且不知道如何让程序接受超过 1 个 int，然后输出正确的 println。这是我到目前为止的代码。
java - 偶/奇 Java 测试
这个测试行得通吗？: if (testInt/2).ofType(Integer){ //to-do if even } 我假设它会 iff 编译器在 ofType() 之前解析 testIn
html - 交替 div 使图像左(偶)或右(奇)
我正在尝试更好地排列图像，而不仅仅是 1 列中的图像。示例见附件，每篇文章的图片可以在左右。这是我的代码。HTML: Content 1
【ACM算法竞赛日常训练】DAY16【奇♂妙拆分】【区区区间间间】【小AA的数列】数学|位运算|前缀和
DAY16共3题：奇♂妙拆分（简单数学）区区区间间间（单调栈）小AA的数列（位运算dp） 🎈 作者：Eriktse 🎈 简介：19

行者123

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

multithreading - 莫霍利奇 : Limiting number of Promises/IOLoop->subprocess