Geo::GDAL + MCE parallel demonstration

geo-minmax.pl

#!/usr/bin/env perl
# http://osgeo-org.1560.x6.nabble.com/gdal-dev-Experiments-with-multiprocessing-td5311397.html

# Demonstration - August 25, 2017
# Use MCE 1.830 or later

use strict;
use warnings;

use MCE;
use Geo::GDAL;
use Time::HiRes qw( time );
use ProgressBar::Stack;

# Usage: geo-minmax.pl image1.tif [ image2.tif ... ]

use List::Util qw( min max );

# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# input iterator
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

my ( $band, $pc, $type, $minval, $maxval );

sub _min {
    $_[0] < $_[1] ? $_[0] : $_[1];
}

sub make_iterator {
    my ( $file ) = @_;

    # open geotiff file
    $band = Geo::GDAL::Open($file, 'ReadOnly')->Band();
    $pc   = Geo::GDAL::PackCharacter($band->{DataType});
    $type = $band->{DataType};

    my ( $W, $H ) = $band->Size;
    my ( $w, $h ) = $band->GetBlockSize;
    my ( $xoff, $yoff, $count, $done ) = ( 0,0,0,0 );

    # reset progress, values
    init_progress(); ( $minval, $maxval ) = ( 255, 0 );

    return sub {
        return if $done;

        if ( $xoff >= $W ) {
            update_progress( sprintf "%0.1f", $yoff / $H * 100 )
                if ( ++$count % 5 == 0 );

            $xoff = 0, $yoff += $h;

            if ( $yoff >= $H ) {
                update_progress('100.0');
                $done = 1, print "\n\n";
                return;
            }
        }

        my @args = ( $xoff, $yoff, _min($W-$xoff,$w), _min($H-$yoff,$h) );
        my $buf  = $band->ReadRaster( @args );

        $xoff += $w;

        return [ $buf, @args ];
    };
}

# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# construct MCE, spawn workers, and run
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

sub update_vals {
    my ( $min, $max ) = @_;
    # update min-max values
    $minval = $min if ( $min < $minval );
    $maxval = $max if ( $max > $maxval );
}

my $mce = MCE->new(
    max_workers => MCE::Util::get_ncpu(),

    user_begin => sub {
        # my ( $mce, $task_id, $task_name ) = @_;

        # reset worker's copy
        ( $band, $pc, $type ) = @{ MCE->user_args() };
        ( $minval, $maxval  ) = ( 255, 0 );
    },

    user_func => sub {
        # fyi: $_ is $chunk_ref->[0]
        # my ( $mce, $chunk_ref, $chunk_id ) = @_;
        # my ( $buf, $xoff, $yoff, $xsize, $ysize ) = @{ $_ };

        my @data = unpack("$pc*", $_->[0]);
        my $min  = min @data;
        my $max  = max @data;

        # update worker's copy
        $minval  = $min if ( $min < $minval );
        $maxval  = $max if ( $max > $maxval );
    },

    user_end => sub {
        # my ( $mce, $task_id, $task_name ) = @_;

        # notify manager process computed min-max values
        MCE->do('update_vals', $minval, $maxval);
    }

)->spawn();

# process one or more files, workers persist between runs

for my $file ( @ARGV ) {
    my $start = time;
    print "\nComputing minmax vals: $file\n";

    # run parallel
    $mce->process({
        input_data => make_iterator($file),
        user_args  => [ $band, $pc, $type ]
    });

    printf "Duration  : %0.03f seconds\n", time - $start;
    print  "Min value : $minval\n";
    print  "Max value : $maxval\n";
}

print "\n" if @ARGV;

# shutdown MCE workers

$mce->shutdown;