Mojolicous: Limiting number of Promises / IOLoop->subprocess

user_2035207 0
Viewed 354

I'm using Mojolicious non-blocking methods (Promises) to request data from external systems. 1) I'd like to notify the user immediately that the process has started; 2) I'd like to scale this program.

The code below works for a small set of numbers (few hundreds), with more numbers, I get an error [error] Can't create pipe: Too many open files at /path/lib/perl5/Mojo/IOLoop.pm line 156. Question 1) How can I limit the number of Promises I spawn (map in my code below):

#!/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>

I commented out the wait; however, it appears the code is still blocking. Question 2) How can I notify the user immediately that the process has already started? (i.e. when I stash the done variable)

mojolicious promise subprocess perl multithreading
1 Answers

The problem isn't the number of promises but the number of subprocesses. One way to limit this is to simply limit how many you create at a time in your program logic. Instead of spawning them all at once in a map, set a limit and retrieve that many from @numbers (perhaps using splice) and spawn those subprocesses; create an ->all promise that waits on those and attach a ->then to that promise to retrieve your next chunk of numbers, and so on.

Another option is to use Future::Utils fmap_concat which can take care of the rate-limiting code by have you provide a number of the maximum outstanding Futures. Your promise-returning function can apply Mojo::Promise::Role::Futurify to chain a following Future to use in this manner.

#!/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;

As for the wait method, this does nothing when the event loop is already running, which in a webapp response handler it will be, if you started the application in a Mojolicious daemon (as opposed to a PSGI or CGI server which don't support asynchronous responses). The ->stash and ->render calls outside of the callbacks will be run immediately after setting up the subprocesses. Then the response handler will complete, and the event loop will have control again, which will fire the appropriate ->then callbacks once the promises resolve. The render should not be waiting for anything beyond the setting up of subprocesses; since you said there may be hundreds, that could be the slowdown you're experiencing. Make sure you are using Mojolicious 7.86 or newer as Subprocess was changed so the fork will not happen until the next tick of the event loop (after your response handler has completed).

I'll also note that Subprocesses aren't really designed for this; they're designed for executing slow code that still returns an eventual result to the browser in a response (and Mojolicious::Plugin::Subprocess is nice for this use case). One problem I can see is that if you restart the application, any still pending subprocesses will just be ignored. For jobs that you want to set off and forget, you might consider a job queue like Minion which has great integration into Mojolicious apps, and runs via a separate worker process.

user_58482000
Related