Please see my update above

I am relieved that things are working up to my projected speed based upon tests. One of the reasons that I moved to MCE from writing my own fork/threads implementation or using other CPAN modules/frameworks is because of the relative speed of IPC with MCE. So I was pretty worried about what I saw. My only wish now is that I can find out what changed and caused a problem before.

Marioroy Thank You for your contribution of MCE! While I agree with Discipulus, zentara, BrowserUK and other monks that developers should understand how to parallelize without the dependency of a framework, and I can and have, I appreciate when I don't have to think so much about this and develop tests to specifically validate that portion of code. As it happens, most of my parallelization needs have patterns that fit pretty cleanly into your framework. So there is minimal need for me to change my approach.

And with your addition of MCE::Flow, I'm actually looking at rewriting one of my existing workflows that currently takes over 4 hours to run. I have hesitated changing it previously because of concerns about how much coordination code I would need to write to pipeline all of the pieces together. I believe that Flow will do much of that for me.

Thanks again!

lbe

Hi again,

Discloser: I tried various things in the event the application is IPC-bound which may not be the case for the OP.

Chunking has the effect of reducing the number of operations for IPC. Thus, relieving pressure from the manager process. Please find the same demonstrations with chunking applied. Basically, chunking requires looping inside the iterator and user_func. In other words, chunking is a way to prevent the application from becoming IPC-bound.

Iterator & Gather demonstration

use strict;
use warnings;

use MCE;
use MCE::Candy;
use Time::HiRes 'time';

# usage: script_gather2.pl [ N ]

open my $fh_out, ">:utf8", "output.txt";

my $mce = MCE->new(
    gather      => MCE::Candy::out_iter_fh($fh_out),
    max_workers => 3,
    chunk_size  => 5,
    user_func   => \&user_func,
)->spawn;

my $start = time;
$mce->process( make_iterator( shift || 1000 ) );
my $duration = time - $start;

$mce->shutdown;
close $fh_out;

printf "duration: %0.3f\n", $duration;

sub make_iterator {
    my $max_id = shift;
    my $nxt_id = 0;

    # closure block receives chunk_size value

    return sub {
        my ( $chunk_size, @data ) = @_;
        return if $nxt_id >= $max_id;

        for ( 1 .. $chunk_size ) {
            push @data, ++$nxt_id;
            last if $nxt_id >= $max_id;
        }

        return @data;
    };
}

sub user_func {
    my ( $mce, $chunk_ref, $chunk_id ) = @_;
    my $output = '';

    # chunking requires looping inside user_func

    for ( 0 .. $#{ $chunk_ref } ) {
        $output .= $chunk_ref->[$_]."\n";
    }

    # gather with chunk_id must be called one time only

    MCE->gather($chunk_id, $output);
}
[download]

Iterator & Relay demonstration

use strict;
use warnings;

use MCE;
use Time::HiRes 'time';

# usage: script_relay2.pl [ N ]

open my $fh_out, ">:utf8", "output.txt";
$fh_out->autoflush(1);

my $mce = MCE->new(
    init_relay  => 0,
    max_workers => 3,
    chunk_size  => 5,
    user_func   => \&user_func,
)->spawn;

my $start = time;
$mce->process( make_iterator( shift || 1000 ) );
my $duration = time - $start;

$mce->shutdown;
close $fh_out;

printf "duration: %0.3f\n", $duration;

sub make_iterator {
    my $max_id = shift;
    my $nxt_id = 0;

    # closure block receives chunk_size value

    return sub {
        my ( $chunk_size, @data ) = @_;
        return if $nxt_id >= $max_id;

        for ( 1 .. $chunk_size ) {
            push @data, ++$nxt_id;
            last if $nxt_id >= $max_id;
        }

        return @data;
    };
}

sub user_func {
    my ( $mce, $chunk_ref, $chunk_id ) = @_;
    my $output = '';

    # chunking requires looping inside user_func

    for ( 0 .. $#{ $chunk_ref } ) {
        $output .= $chunk_ref->[$_]."\n";
    }

    # relay must be called one time only

    MCE::relay { print {$fh_out} $output };
}
[download]

Regards, Mario

Hi again,

Discloser: I also tried chunking with threads and MCE::Hobo using a queue for comparison with MCE (Iterator & Gather/Relay) demonstrations.

Please find demonstrations using threads and MCE::Hobo for testing IPC overhead. On Unix platforms (particular BSD variants), the MCE solution may run up to 2x faster than threads. Increasing CHUNK_SIZE will decrease IPC overhead plus increase the memory footprint for the $output variable. Chunking/batching is helpful. Just be mindful of memory consumption in the output routine.

Regarding the input queue, I limit to 50 items to prevent the worker from populating the queue out of control. Thread::Queue must be recent with limit capability or will fail.

Threads & Thread::Queue->new() demonstration

use strict;
use warnings;

use threads;
use Thread::Queue 3.07;
use Sereal qw( decode_sereal encode_sereal );
use Time::HiRes qw( time );

use constant {
    MAX_WORKERS => 3,
    CHUNK_SIZE  => 5,
};

# usage: script_que_thrs.pl [ N ]

my ( @wrks_in, @wrks_out );
my $que_in  = Thread::Queue->new(); $que_in->limit = 50;
my $que_out = Thread::Queue->new();
my $start   = time;

push @wrks_in,  threads->create(\&task) for 1 .. MAX_WORKERS;
push @wrks_out, threads->create(\&output);

input( shift || 1000 );

$que_in->end;  $_->join for @wrks_in;
$que_out->end; $_->join for @wrks_out;

printf "duration: %0.3f\n", time - $start;

sub input {
    my $max_id = shift;
    my $nxt_id = 0;
    my @data;

    while ( 1 ) {
        last if $nxt_id >= $max_id;
        push @data, ++$nxt_id;
        if ( @data == CHUNK_SIZE ) {
            $que_in->enqueue( encode_sereal(\@data) );
            @data = ();
        }
    }

    $que_in->enqueue( encode_sereal(\@data) ) if @data;
}

sub task {
    while ( defined ( my $frozen = $que_in->dequeue ) ) {
        my $array_ref = decode_sereal($frozen);
        my $output = '';
        for ( 0 .. $#{ $array_ref } ) {
            $output .= $array_ref->[$_]."\n";
        }
        $que_out->enqueue($output);
    }
}

sub output {
    open my $fh_out, ">:utf8", "output.txt";
    while ( defined ( my $output = $que_out->dequeue ) ) {
        print {$fh_out} $output;
    }
    close $fh_out;
}
[download]

MCE::Hobo & MCE::Shared->queue() demonstration

use strict;
use warnings;

use MCE::Hobo;
use MCE::Shared;
use Sereal qw( decode_sereal encode_sereal );
use Time::HiRes qw( time );

use constant {
    MAX_WORKERS => 3,
    CHUNK_SIZE  => 5,
};

# usage: script_que_hobo.pl [ N ]

my ( @wrks_in, @wrks_out );
my $que_in  = MCE::Shared->queue( await => 1 );
my $que_out = MCE::Shared->queue();
my $start   = time;

push @wrks_in,  MCE::Hobo->create(\&task) for 1 .. MAX_WORKERS;
push @wrks_out, MCE::Hobo->create(\&output);

input( shift || 1000 );

$que_in->end;  $_->join for @wrks_in;
$que_out->end; $_->join for @wrks_out;

printf "duration: %0.3f\n", time - $start;

sub input {
    my $max_id = shift;
    my $nxt_id = 0;
    my @data;

    while ( 1 ) {
        last if $nxt_id >= $max_id;
        push @data, ++$nxt_id;
        if ( @data == CHUNK_SIZE ) {
            $que_in->await(50); # limit
            $que_in->enqueue( encode_sereal(\@data) );
            @data = ();
        }
    }

    $que_in->enqueue( encode_sereal(\@data) ) if @data;
}

sub task {
    while ( defined ( my $frozen = $que_in->dequeue ) ) {
        my $array_ref = decode_sereal($frozen);
        my $output = '';
        for ( 0 .. $#{ $array_ref } ) {
            $output .= $array_ref->[$_]."\n";
        }
        $que_out->enqueue($output);
    }
}

sub output {
    open my $fh_out, ">:utf8", "output.txt";
    while ( defined ( my $output = $que_out->dequeue ) ) {
        print {$fh_out} $output;
    }
    close $fh_out;
}
[download]

Update:

Workers may write directly to the output handle, one at a time. This is possible with a mutex. Do not forget to enable autoflush on the file handle. Recent version of MCE and MCE::Shared load IO::Handle automatically in the event the autoflush method is not found.

use strict;
use warnings;

use MCE::Hobo;
use MCE::Mutex;
use MCE::Shared;
use Sereal qw( decode_sereal encode_sereal );
use Time::HiRes qw( time );

use constant {
    MAX_WORKERS => 3,
    CHUNK_SIZE  => 5,
};

# usage: script_que_hobo2.pl [ N ]

my ( @wrks_in, @wrks_out );
my $que_in = MCE::Shared->queue( await => 1 );
my $mutex  = MCE::Mutex->new();
my $start  = time;

open my $fh_out, ">:utf8", "output.txt";
$fh_out->autoflush(1);

push @wrks_in, MCE::Hobo->create(\&task) for 1 .. MAX_WORKERS;
input( shift || 1000 );

$que_in->end; $_->join for @wrks_in;
close $fh_out;

printf "duration: %0.3f\n", time - $start;

sub input {
    my $max_id = shift;
    my $nxt_id = 0;
    my @data;

    while ( 1 ) {
        last if $nxt_id >= $max_id;
        push @data, ++$nxt_id;
        if ( @data == CHUNK_SIZE ) {
            $que_in->await(50);
            $que_in->enqueue( encode_sereal(\@data) );
            @data = ();
        }
    }

    $que_in->enqueue( encode_sereal(\@data) ) if @data;
}

sub task {
    while ( defined ( my $frozen = $que_in->dequeue ) ) {
        my $array_ref = decode_sereal($frozen);
        my $output = '';

        for ( 0 .. $#{ $array_ref } ) {
            $output .= $array_ref->[$_]."\n";
        }

        $mutex->enter(sub {
            print {$fh_out} $output;
        });
    }
}
[download]

Regards, Mario

Please see my update 4 above. chunking definitely helped

lbe

Please see my update 4 above. chunking definitely helped

lbe