Hi monks! need your help in finding the problem in the code.

For a set of hosts : for each hosts

1. I'm forking a child process, creating a pipe

2. writing end of the pipe (file handle) is passed to child

3. Reading end of pipe is added to select object in parent

In parent, I'm waiting for number of child processes to fall below max processes allowed and also reading from the file handles which are readable ( by accessing select objects' can_read method ).

Data read is merged in a hash ref.

But, after all childs are finished and data is dumped from hash ref which should have all child's data, sometimes it has data from all the hosts but sometimes data from some hosts are missed.

Need your help in finding what I'm doing wrong. Here's the code.

#!/usr/bin/env perl

use warnings;
use strict;
use Getopt::Long;
use POSIX ":sys_wait_h";
use IO::Select;
use IO::Pipe;
use Fcntl;
use Data::Dumper;

my $h_r;
my $c_max;
my $sel = IO::Select->new();

sub sel_can_read {
    # args : timeout
    my ($t) = @_;

    # select-object($sel), hashref($h_r)
    
    # @rh, ready handles. $h, handle
    while ( my @rh = $sel->can_read($t) ) {
        foreach my $h (@rh) {
            my $o = join ('' , $h->getlines());
            my $_h = eval($o);
            $h_r = _merge_hash($_h, $h_r);
            # only one read per handle, so close
            $sel->remove($h);
            $h->close;
        }
    }
}

sub wait_num_child_below_max {

        my ( $child_aref, $c_m ) = @_;
        
        while ( scalar(@{$child_aref}) > $c_m ) {
            foreach my $c ( @{$child_aref} ) {
                @{$child_aref} = grep { $_ ne $c } @{$child_aref} if (
+ waitpid($c, WNOHANG) < 0 );
                # read child's fh
                sel_can_read(0);        # non-blocking
            }
        }
        return;
}


sub _merge_hash {
    # args : array of hashref1, hashref2 to be merged,
    # returns merge hashref
    
    my $h_r;
    foreach (@_) {
        if ( defined($h_r) ) {
            $h_r = { %$h_r, %$_ };
        } else {
            $h_r = $_;
        }
    }
    return $h_r;
}

sub do_some_stuff_for_host {
    my ($fh, $host) = @_;
    my %h;
    # do some stuff, populate (hash) %h 
    print $fh Data::Dumper->new([\%h])->Terse(1)->Dump;
}

sub print_report {
    print STDOUT Data::Dumper->new([$h_r])->Terse(1)->Dump;
}


sub main {
    my @a = ('host1', 'host2');
    my @c_p;
    
    foreach my $i (@a) {
        wait_num_child_below_max(\@c_p, $c_max);
        
        my $pipe = IO::Pipe->new();
        
        my $pid = fork();

        if ( $pid ) {
            # parent
            push @c_p, $pid;
            my $fh = $pipe->reader();
            $fh->blocking(0);    # non-blocking read
            $sel->add($fh);
        } elsif ( $pid == 0) {
            #child
            my $cfh = $pipe->writer();
            do_some_stuff_for_host(\*$cfh,$i);
            exit (0);
        }
    }    
    
    # leave no child behind
    wait_num_child_below_max(\@c_p, 0);
    
    print_report();
}

main();
[download]