complex-futures.pl

#!/usr/bin/env perl
# Perl 5 version of https://gist.github.com/benjchristensen/4671081
# Using IO::Async and Future
use strict;
use warnings;
use feature 'say';
use curry;

use Future;
use IO::Async::Function;
use IO::Async::Loop;
use Time::HiRes;

my $loop = IO::Async::Loop->new;

for my $sub ( \&run, \&run2, \&run3, \&run3f, \&run4, \&run5 ) {
    my $t0 = [ Time::HiRes::gettimeofday() ];
    $sub->();
    say "Finished in @{[ Time::HiRes::tv_interval $t0 ]} seconds";
}

# demo how futures can easily become blocking and prevent other work
# from being performed asynchronously
sub run {
    my $f1 = call_a();
    my $f3 = call_c( $f1->get );

    # these are blocked until $f1->get resolves
    my $f2 = call_b();
    my $f4 = call_d( $f2->get );
    my $f5 = call_e( $f2->get );

    say $f3->get, " => ", $f4->get * $f5->get;
}

# demo how reordering of futures can improve the situation but still
# does not address differing response latencies of $f1 and $f2
sub run2 {
    my $f1 = call_a();
    my $f2 = call_b();

    # blocks until $f1->get resolves
    my $f3 = call_c( $f1->get );

    # blocks until $f2->get resolves
    my $f4 = call_d( $f2->get );
    my $f5 = call_e( $f2->get );

    say $f3->get, " => ", $f4->get * $f5->get;
}

# demo how changing where processes are injected to solve the issue of
# run2() (at the cost of some complexity via future chaining)
sub run3 {
    my $f1 = call_a();
    my $f2 = call_b();

    # spawn another process so waiting on $f1->get for $f3 doesn't block
    # $f4/$f5.  Note that this shares $f1 across processes, which is not
    # recommended.
    my $fn = IO::Async::Function->new( code => sub { $f1->get } );
    $loop->add($fn);
    my $f3 = $fn->call( args => [] )->then( sub { call_c(shift) } );

    # blocks until $f2->get resolves
    my $f4 = call_d( $f2->get );
    my $f5 = call_e( $f2->get );

    say $f3->get, " => ", $f4->get * $f5->get;
}

# demo fluent use of Future methods to provide a better example than all
# the above, as well as reducing blocking get() calls
sub run3f {
    my $f1 = call_a();
    my $f2 = call_b();

    # sequence resolving $f1 then $f3 without blocking $f4/$5; no need
    # for spawning a process!
    my $f3 = $f1->then( sub { call_c(shift) } );

    # $f2 will resolve first then $f4/$f5 follows
    my $f4 = $f2->then( sub { call_d(shift) } );
    my $f5 = $f2->then( sub { call_e(shift) } );

    # converge all futures into a single future to get() once
    Future->needs_all( $f3, $f4, $f5 )->then(
        sub {
            my ( $r3, $r4, $r5 ) = @_;
            say $r3, " => ", $r4 * $r5;
            Future->done;
        }
    )->get;
}

# demo typical handling of futures as they complete: execute multiple
# calls but synchronously handle each response in the order they are put
# in the list, rather than the order they complete.
sub run4 {
    my @futures = (
        call_a(),       #
        call_b(),       #
        call_c('A'),    #
        call_c('B'),    #
        call_c('C'),    #
        call_d(1),      #
        call_e(2),      #
        call_e(3),      #
        call_e(4),      #
        call_e(5),      #
    );

    map { say "do more work => ", $_->get } @futures;
}

# demo polling/callback approach to futures as they complete, handling
# responses as soon as the future is done.
sub run5 {
    my @futures = (
        call_a(),       #
        call_b(),       #
        call_c('A'),    #
        call_c('B'),    #
        call_c('C'),    #
        call_d(1),      #
        call_e(2),      #
        call_e(3),      #
        call_e(4),      #
        call_e(5),      #
    );

    Future->needs_all(
        map {
            $_->on_done( sub { say "do more work => ", +shift } )
        } @futures
    )->get;
}

sub call_a {
    $loop->delay_future( after => 10 )->then_done('responseA');
}

sub call_b {
    $loop->delay_future( after => 4 )->then_done(100);

}

sub call_c {
    my $b = shift;
    $loop->delay_future( after => 6 )->then_done("responseB_$b");
}

sub call_d {
    my $b = shift;
    $loop->delay_future( after => 14 )->then_done( 40 + $b );
}

sub call_e {
    my $b = shift;
    $loop->delay_future( after => 5 )->then_done( 5000 + $b );
}