use strict;
use warnings;
 
use Gearman::XS qw(:constants);
use Gearman::XS::Client;
use FreezeThaw qw(freeze thaw cmpStr safeFreeze cmpStrHard);
 
# set up the mapper
print "Connect to the Mapper gearman servers.\n";
my $mapper = new Gearman::XS::Client;
if ($mapper->add_server('localhost', 4730) != GEARMAN_SUCCESS) {
        printf(STDERR "%s\n", $mapper->error());
        exit(1);
}
 
# set up the reducer, generate a sortable unique id
print "Connect to the Reducer gearman servers.\n";
my $reducer = new Gearman::XS::Client;
if ($reducer->add_server('localhost', 4731) != GEARMAN_SUCCESS) {
        printf(STDERR "%s\n", $reducer->error());
        exit(1);
}
my $reducer_id = time . '_' . join "", map { ("a".."z", 0..9)[rand 36] } (1..4);

# submit jobs with each split to the mappers
my ($ret, $job_handle);
my $jobs = {};
foreach my $split (@splits) {
        # submit a mapper job to be performed by gearman workers
        ($ret, $job_handle) = $mapper->do_background( 'mapper',
                freeze ({ # workload
                        'split' => int($split)
                })
        );
 
        # add this to the jobs to be monitored
        if ($ret == GEARMAN_SUCCESS) {
                print "> Begin mapping $split with job_handle=$job_handle.\n";
                $jobs->{$job_handle} = {
                        mapper => 1, split => int($split), gearman_client => $mapper
                };
 
        } else {
                printf(STDERR "%s\n", $mapper->error()) and die;
        }
 
        # sleep for a tenth of a sec
        select(undef, undef, undef, 0.10);
}

while (1) {   
    # stop if there are no more jobs
    last unless (keys %$jobs);
 
    # check each job, run reducer when a mapper is done
    foreach $job_handle (sort keys %$jobs) {
 
        # get the job status from this job's gearman client
        ($return_value, $is_status_known, $running_status, $status_num, $status_denom) = 
                $jobs->{$job_handle}->{gearman_client}->job_status($job_handle);
 
        # this job is done
        unless ($running_status) {
            # this is a complete mapper job... run its reducer
            if ($jobs->{$job_handle}->{mapper}) {
                 ....
            ｝
            delete $jobs->{$job_handle};
         ｝
           # sleep for a tenth of a sec
           select(undef, undef, undef, 0.10);
      }
}

if ($jobs->{$job_handle}->{mapper}) {
 
    warn("< Done mapping " . $jobs->{$job_handle}->{split} . " with job_handle=$job_handle.\n");
 
    # submit a reducer job to be performed by gearman workers
    ($ret, $re_job_handle) = $reducer->do_background( 'reducer',
            freeze ({ # workload
                'split' => int($jobs->{$job_handle}->{split}), 
                'reducer_id' => $reducer_id 
                })  
            );  
 
    # add this to the jobs to be monitored
    if ($ret == GEARMAN_SUCCESS) {
        print ">> Begin reducing " . $jobs->{$job_handle}->{split} . " with job_handle=$re_job_handle\n";
        $jobs->{$re_job_handle} = { reducer => 1, split => $jobs->{$job_handle}->{split}, gearman_client => $reducer };
    } else {
        printf(STDERR "%s\n", $mapper->error()) and die;
    }   
 
}