MCE shared hash demonstrations

demo_flow_and_shared.pl

##
# Article:
#   https://blog.celogeek.com/201401/426/perl-benchmark-cache-with-expires-and-max-size/
#
#   Parallel demonstration via MCE::Flow and MCE::Shared->hash()
#   Ensure the Sereal module is installed for extra performance
##

use strict;
use warnings;

use Digest::MD5 qw/md5_base64/;
use Time::HiRes qw/time/;
use feature 'say', 'state';
use Proc::ProcessTable;

use MCE::Flow;
use MCE::Shared Sereal => 1;

sub get_current_process_memory {
    state $pt = Proc::ProcessTable->new;
    my %info = map { $_->pid => $_ } @{$pt->table};
    return $info{$$}->rss;
}

$|=1;
my $c = MCE::Shared->hash();

say "Mapping";
my @todo = map { md5_base64($_) } (1..600_000);
say "Starting";

my $mem = get_current_process_memory();

my ($read, $write, $found);
{
    my $s = time;

    mce_flow_s {
       bounds_only => 1, max_workers => 8
    }, sub {
       for my $i ($_->[0] .. $_->[1]) {
          $c->set($todo[$i], {md5 => $todo[$i]});
          print "Write: $i\r" if $i % 1000 == 0;
       }
    }, 0, $#todo;

    MCE::Flow::finish;

    $write = time - $s;
}
say "Write: ", scalar(@todo) / $write;
{
    my $s = time;

    my @f = mce_flow_s {
       bounds_only => 1, max_workers => 8,
       user_end => sub { MCE->gather($found) }
    }, sub {
       for my $i ($_->[0] .. $_->[1]) {
          $found++ if ref $c->get($todo[$i]) eq 'HASH';
          print "Read : $i\r" if $i % 1000 == 0;
       }
    }, 0, $#todo;

    MCE::Flow::finish;

    $found += $_ for @f;
    $read = time - $s;
}

say "Read : ", scalar(@todo) / $read;
say "Found: ", $found;
say "Mem  : ", get_current_process_memory() - $mem;

demo_hobo_and_shared.pl

##
# Article:
#   https://blog.celogeek.com/201401/426/perl-benchmark-cache-with-expires-and-max-size/
#
#   Parallel demonstration via MCE::Hobo and MCE::Shared->hash()
#   Ensure the Sereal module is installed for extra performance
##

use strict;
use warnings;

use Digest::MD5 qw/md5_base64/;
use Time::HiRes qw/time/;
use feature 'say', 'state';
use Proc::ProcessTable;

use MCE::Hobo;
use MCE::Shared Sereal => 1;

sub get_current_process_memory {
    state $pt = Proc::ProcessTable->new;
    my %info = map { $_->pid => $_ } @{$pt->table};
    return $info{$$}->rss;
}

$|=1;
my $c = MCE::Shared->hash();

say "Mapping";
my @todo = map { md5_base64($_) } (1..600_000);
say "Starting";

my $mem = get_current_process_memory();

my $seq = MCE::Shared->sequence(
   { bounds_only => 1, chunk_size => 500 },
   0, $#todo
);

my $found = MCE::Shared->scalar(0);

my ($read, $write);
{
    my $s = time;

    my $func = sub {
       while ( my ( $beg, $end ) = $seq->next ) {
          for my $i ( $beg .. $end ) {
             $c->set($todo[$i], {md5 => $todo[$i]});
             print "Write: $i\r" if $i % 1000 == 0;
          }
       }
    };

    MCE::Hobo->create($func) for ( 1 .. 8 );

    # Do other stuff

    $_->join for MCE::Hobo->list;

    $write = time - $s;
}
say "Write: ", scalar(@todo) / $write;
{
    my $s = time;

    # Rewind the sequence iterator
    $seq->rewind;

    my $func = sub {
       my $_found = 0;
       while ( my ( $beg, $end ) = $seq->next ) {
          for my $i ( $beg .. $end ) {
             $_found++ if ref $c->get($todo[$i]) eq 'HASH';
             print "Read : $i\r" if $i % 1000 == 0;
          }
       }
       $found->incrby($_found);
    };

    MCE::Hobo->create($func) for ( 1 .. 8 );

    # Do other stuff

    $_->join for MCE::Hobo->list;

    $read = time - $s;
}

say "Read : ", scalar(@todo) / $read;
say "Found: ", $found->get;
say "Mem  : ", get_current_process_memory() - $mem;