Advice on object interface

Random_Walk has asked for the wisdom of the Perl Monks concerning the following question:

Oh Exalted and Glorious Monks I have been in the wilderness and no longer see my way through these barren lands. Please give me a gently push towards the true path.

I am writing a module to run one or more system commands in parallel for a limited time on both *nix and Win32. It all started with this thread Win32 capturing output from a process that may hang and then grew here delayed die and open3 now I have the inklings of a useful module but would love to get some feedback on what I am doing right, what is abhorrent and any suggestions on how the return the results from the object and any coding style improvements

Using the module goes something like this

use run_limited;          # I need a better name
my $runner = Runner->new; # got to get the names straight 
$runner->load('true');
$runner->load('sleep 2');
$runner->run("$timeout");
[download]

Then it gets ugly, I end up with all the results in the Runner object

$VAR1 = \bless( {
              'sleep 999' => {
                               'pid' => '38864',
                               'stderr' => bless( [], 'Thread::Queue' 
+),
                               'status' => 2,
                               'signaling' => bless( [
                                                       'command comple
+te'
                                                     ], 'Thread::Queue
+' ),
                               'error' => [],
                               'output' => [],
                               'stdout' => bless( [], 'Thread::Queue' 
+)
                             },
              'echo STDERR 1>&2' => {
                               'stderr' => bless( [], 'Thread::Queue' 
+),
                               'time' => 1,
                               'status' => 0,
                               'output' => [],
                               'outcome' => 'command complete',
                               'pid' => '57234',
                               'signaling' => bless( [], 'Thread::Queu
+e' ),
                               'error' => [
                                            'STDERR
'
                                          ],
                               'stdout' => bless( [], 'Thread::Queue' 
+)
                               },
              'no_such_command' => {
                               'stderr' => bless( [], 'Thread::Queue' 
+),
                               'time' => 1,
                               'status' => 0,
                               'output' => [],
                               'outcome' => 'command complete',
                               'pid' => '57236',
                               'signaling' => bless( [], 'Thread::Queu
+e' ),
                               'error' => [],
                               'stdout' => bless( [], 'Thread::Queue' 
+)
                               },
                   'true' => {
                               'stderr' => bless( [], 'Thread::Queue' 
+),
                               'time' => 1,
                               'status' => 0,
                               'output' => [],
                               'outcome' => 'command complete',
                               'pid' => '50508',
                               'signaling' => bless( [], 'Thread::Queu
+e' ),
                               'error' => [],
                               'stdout' => bless( [], 'Thread::Queue' 
+)
                             },
                   'echo STDOUT' => {
                               'stderr' => bless( [], 'Thread::Queue' 
+),
                               'time' => 1,
                               'status' => 0,
                               'output' => [
                                             'STDOUT
'
                                           ],
                               'outcome' => 'command complete',
                               'pid' => '20080',
                               'signaling' => bless( [], 'Thread::Queu
+e' ),
                               'error' => [],
                               'stdout' => bless( [], 'Thread::Queue' 
+)
                               },
                   'sleep 2' => {
                               'stderr' => bless( [], 'Thread::Queue' 
+),
                               'time' => 3,
                               'status' => 0,
                               'output' => [],
                               'outcome' => 'command complete',
                               'pid' => '43846',
                               'signaling' => bless( [], 'Thread::Queu
+e' ),  
                               'error' => [],
                               'stdout' => bless( [], 'Thread::Queue' 
+)
                               }
                 }, 'Runner' );
[download]

What is the best way to get at all this ? Should I just clean up the object (undef the thread queues for instance) and return %$self ? to pass a useful structure back or would it be better to define methods like

$stdout=$runner->stdout($command);
$stderr=$runner->stderr($command);
... etc ...
[download]

and then I can also use want array to return them in a similar way to [$|@]results=`cmd` does.

Here is the code for the module:

#!/usr/bin/perl
#

{
package Runner;

use strict;
use warnings;
use threads;
use Thread::Queue;
use IPC::Open3;

sub new {
    my $class = shift;
    return bless {}, $class;
}

sub load {
    my $self=shift;
    for (@_) {
        $self->{$_}={};
    }
}

sub run {
    my $self=shift;
    my $timeout=shift;
    my $running_commands = 0;
    my @threads;
    local $SIG{CHLD}='IGNORE'; # incantation against zombies is this v
+oodoo ?

    while (my ($command, $info) = each %$self) {
        $info->{stdout}    = new Thread::Queue;
        $info->{stderr}    = new Thread::Queue;
        $info->{signaling} = new Thread::Queue;
        push @threads, threads->create(
            \&RunInThread,
            $command,
            $info->{stdout},
            $info->{stderr},
            $info->{signaling},
        );
        $info->{output} = [];
        $info->{error}  = [];
        $info->{status} = 2;
        $info->{pid}    = $info->{signaling}->dequeue;
        $running_commands++;
    }

    for my $time (1..$timeout) {
        last unless $running_commands;
        sleep 1;
        for my $info (values %$self) {
            next unless ($info->{status})>>1;
            while (my $result = $info->{signaling}->dequeue_nb) {
                $info->{outcome}=$result;
                if ($result eq "command complete") {
                    $info->{status}=0;
                    $info->{time}=$time;
                    $running_commands--;
                    next;
                } else {
                    # something bad is being signalled
                    $info->{status}=1;
                }
            }
            while (my $result = $info->{stdout}->dequeue_nb) {
                push @{$info->{output}}, $result;
            }
            while (my $error  = $info->{stderr}->dequeue_nb) {
                push @{$info->{error}}, $error;
            }
        }
    }
    # time is up
    # kill those that did not complete

    my $killed=0;
    if ($running_commands) {
        for my $info (values %$self) {
            next unless $info->{status} == 2;
            $killed++;
            # politeness costs nothing ?
            kill 15, $info->{pid};
            while (my $result = $info->{stdout}->dequeue_nb) {
                if ($result eq "command complete") {
                    $info->{status}=0;
                    $info->{time}=$timeout;
                    next;
                }
                push @{$info->{output}}, $result;
            }
            kill 9, $info->{pid};
            waitpid $info->{pid}, 0;
        }
    }
    $_->join for @threads;
    return $killed;
}

sub RunInThread {
    my ( $cmd, $stdout, $stderr, $signaling) = @_;
    my ( $in, $out, $err, $pid);
    # duplicate stderr and stdout
    open( $out, ">&STDOUT" ) or
        die "Can't dup STDOUT to OUTPUT: $!\n";
    open( $err, ">&STDERR" ) or
        die "Can't dup STDERR to OUTERR: $!\n";
    eval { $pid = open3($in, $out, $err, $cmd) };
    if ($@) {
        $signaling->enqueue($@);
        exit 1; # this will be in a forked child
    } else {
        $signaling->enqueue($pid);
        $stdout->enqueue( $_ ) while <$out>;
        $stderr->enqueue( $_ ) while <$err>;
        $signaling->enqueue("command complete");
    }
}

# the following right bracket closes the package
}
1;
[download]

And this is my testing code

#!/usr/bin/perl
use strict;
use warnings;
use Data::Dumper;
use Test::More tests => 4;

my $timeout = (shift or 5);

push @INC, '.';

BEGIN {use_ok('run_limited')}

my $runner = Runner->new;
isa_ok($runner, 'Runner');
can_ok($runner, qw(load run));

$runner->load('true');
$runner->load('sleep 2');
$runner->load('sleep 999');
$runner->load('echo STDOUT');
$runner->load('echo STDERR 1>&2');
$runner->load('no_such_command');


is ($runner->run("$timeout"), 1, "Check correct number had to be kille
+d");

print Dumper(\$runner);
[download]

The other problem I am having is capturing complete failure of the child (ie when I try to run no_such_command). The open3 docs talk about catching SIGPIPE but I tried

our $sigpipe=0;
$SIG{PIPE}=sub{$sigpipe++};
[download]

But $sigpipe resolutely remained zero.

Cheers,
R.

Pereant, qui ante nos nostra dixerunt!

Comment on Advice on object interface Select or Download Code

Replies are listed 'Best First'.
Re: Advice on object interface by Random_Walk (Prior) on Apr 04, 2005 at 15:26 UTC
Nothing like posting to Perlmonks to clarify the mind. Of course I do not need to read from the non blocking queues in the poll loop, I can leave all the output in them until all the threads are finished or killed. If I make a nice OO interface to get the results they can just be pulled direct from the queues when requested. Cheers, R. Pereant, qui ante nos nostra dixerunt!	[reply]