My brother some of what you describe is exactly why I do not use those modules. PLEASE check this out and adapt as needed. The communication and timer aspects of this are mine, however the threading efficiency belongs to BrowserUk... and that for that I must give thanks... I hope it helps.

You will need to adjust the TIMEOUT (in seconds) to the value that you need. Also, adjust the THREADS to the number of threads to the value that you want to run. Populate the @SERVERS variable however you choose. If you need an understanding of why I use the SSH_OPTs like I do, just let me know. Also, these options will work for SCP. Finally, you will need a key for the SSH_KEY constant. Keep in mind the %RESULTS is a simple shared hash, you can populate this as you see fit. However, if you are going to expand the RESULTS beyond the simple example, you will probably need to review shared_clone from the threads::shared module.

If you need any assistance I will help where I can...

#!/usr/bin/perl -l
use strict;
use threads;
use threads::shared;
use Thread::Queue;
use Time::HiRes qw( usleep time );
use IPC::Open3;
use POSIX qw(:errno_h :sys_wait_h);
use File::Basename;
use FindBin qw( $RealBin $RealScript );
use FileHandle;

BEGIN
{
  our $SCRIPT_NAME  = $RealScript;
  our $SCRIPT_DIR   = File::Basename::dirname( $RealBin );
  # if you have a custom lib, add it.
  push ( @INC, ( q{.}, qq{$SCRIPT_DIR/lib} ) );
}

use constant {
  RET_SUCCESS   => 1,
  RET_FAILURE   => 0,
  EXIT_SUCCESS  => 0,
  EXIT_FAILURE  => 1,

  TIMEOUT       => 4,  ## SET YOUR TIMEOUT in seconds
  THREADS       => 2,  ## SET the number of threads you want to run.

  SSH_USER  =>  q{root},  ## Remote user
  SSH_KEY   =>  qq{$main::SCRIPT_DIR/keys/YOUR_DSS_KEY},  ## authorize
+d keys
  SSH_CMD   =>  q{/usr/bin/ssh},
  SSH_OPT   =>  q{-q -o UserKnownHostsFile=/dev/null } .
                q{-o StrictHostKeyChecking=no } .
                q{-o BatchMode=yes } .
                q{-o ConnectTimeout=10 } .
                q{-o NoHostAuthenticationForLocalhost=yes } .
                q{-o PreferredAuthentications=publickey } .
                q{-o ServerAliveInterval=15 } .
                q{-o ServerAliveCountMax=4 } .
                q{-o TCPKeepAlive=no},

};
my %RESULTS :shared;        # Stupid example of collective
                            # returned information.
my %PROCESS_WATCH :shared;  # watch external procs.

# pretend like we have obtained some list of servers.
my @SERVERS = qw( server01.your.domain.com
                  server02.your.domain.com
                 );

my $semSTD :shared;
sub tprint {
  my $tid = threads->tid;
  lock $semSTD;
  print STDOUT  q{[} . timestamp() . q{][} .
                $tid . q{]: }, @_;
  return RET_SUCCESS;
}

sub timestamp {
  return localtime time;
} ## end timestamp.

my $die_early :shared = 0;
$SIG{ INT } = sub {
  tprint q{Early termination requested};
  $die_early = 1;
};

sub check_process_signal {
  my $sig = shift;
  if ( WIFEXITED($sig) ) {
    tprint q{process normal exit};
    return RET_SUCCESS;
  } elsif ( WIFSIGNALED($sig) ) {
    tprint q{process terminated because of signal};
    return RET_FAILURE;
  }  elsif ( WIFSTOPPED($sig) ) {
    tprint q{process is stopped};
    return RET_FAILURE;
  }
  return RET_SUCCESS;
}

sub add_to_process_watch {
  my $pid = shift;
  lock %PROCESS_WATCH;
  $PROCESS_WATCH{$pid} = time;
  return RET_SUCCESS;
}

sub remove_from_process_watch {
  my $pid = shift;
  lock %PROCESS_WATCH;
  if ( defined $PROCESS_WATCH{$pid} ) { delete $PROCESS_WATCH{$pid}; }
  return RET_SUCCESS;
}

sub set_results {
  lock %RESULTS;
  ($_[0]) ? $RESULTS{'success'}++ : $RESULTS{'failure'}++;
  return RET_SUCCESS
}

sub is_pid_alive {
  my $pid = shift;
  my $status = RET_SUCCESS;
  if ( kill(0, $pid) ) {      ## Still alive.
    $status =  RET_SUCCESS;
  } elsif ( $! == EPERM ) {   ## Changed UID.
    $status =  RET_SUCCESS;
  } elsif ( $! == ESRCH ) {   ## Died or zombied.
    $status =  RET_FAILURE;
  } else {                    ## Could not locate.
    $status =  RET_FAILURE;
  }
  return $status;
} ## end is_pid_alive.

sub run_command {
  my $o = {
    'debug'     => q{},
    'host'      => q{},
    'ssh_user'  => q{},
    'ssh_key'   => q{},
    'cmd'       => q{},
    'opt'       => q{},
    @_,
  };

  my $cmd = q{};

  if ( (defined $o->{'host'}) and
       ($o->{'host'} ne q{}) ) {
    # Remote command.
    my $ssh_user  = ( $o->{'ssh_user'} ne q{} )
                      ? $o->{'ssh_user'}
                      : SSH_USER;
    my $ssh_key   = ( $o->{'ssh_key'} ne q{} )
                      ? $o->{'ssh_key'}
                      : SSH_KEY;
    $cmd =  SSH_CMD . qq{ -i $ssh_key } .
            SSH_OPT .
            qq{ $ssh_user\@$o->{'host'} '$o->{'cmd'} $o->{'opt'}'};
  } else {
    # Local command.
    $cmd = qq{$o->{'cmd'} $o->{'opt'}}
  }

  my $hdl = {
    'stdin'   => FileHandle->new,
    'stdout'  => FileHandle->new,
    'stderr'  => FileHandle->new,
  };
  my $pid = eval { open3( $hdl->{'stdin'},
                          $hdl->{'stdout'},
                          $hdl->{'stderr'}, qq{$cmd} ) or die $!; };

  add_to_process_watch( $pid );
  tprint qq{waiting for external process: $pid};
  waitpid( $pid, 0 );
  my $exit_status = check_process_signal($?);

  my $h_ret = {
    'stdout'  =>  [$hdl->{'stdout'}->getlines()],
    'stderr'  =>  [$hdl->{'stderr'}->getlines()],
    'status'  => $exit_status
  };

  return $h_ret;
}

sub get_disk_information {
  my $o = {
    'debug'       => 0,
    'host'        => q{},
    'df_opt'      => q{-Pk},
    @_
  };

  my $ref_cmd = run_command(  'debug'  => $o->{'debug'},
                              'cmd'    => q{/bin/df},
                              'opt'    => $o->{'df_opt'},
                              'host'   => $o->{'host'} );
  # one could parse the data as needed
  # but I will just return the info...
  return $ref_cmd;
}

sub get_uname_information {
  my $o = {
    'debug'       => 0,
    'host'        => q{},
    'uname_opt'   => q{},
    @_
  };

  my $ref_cmd = run_command(  'debug'  => $o->{'debug'},
                              'cmd'    => q{/bin/uname},
                              'opt'    => $o->{'uname_opt'},
                              'host'   => $o->{'host'} );
  # one could parse the data as needed
  # but I will just return the info...
  return $ref_cmd;
}

sub worker {
  my( $Q ) = @_;
  tprint q{worker started};

  while( !$die_early and defined( my $job = $Q->dequeue ) ) {
    tprint qq{processing job: $job};

    my $ref_di = get_disk_information( 'debug'  => 0,
                                       'host'   => $job,
                                       'df_opt' => q{-Pk /opt} );

    # process the return
    ( $ref_di->{'status'} )
      ? tprint q{this thread might continue}
      : tprint q{this thread might move on to the next job};

    # one might just print the data
    tprint qq{stdout: $job\n}, @{$ref_di->{'stdout'}}
      if ( $#{$ref_di->{'stdout'}} > 0 );
    tprint qq{stderr: $job\n}, @{$ref_di->{'stderr'}}
      if ( $#{$ref_di->{'stderr'}} > 0 );

    # one might just want to know overall status.
    set_results( $ref_di->{'status'} );

    # OBVIOUSLY I could call get_uname_information
    # and collect and report on that information as
    # well. Just an example.

  } ## end while
  tprint q{Worker ending};
  return RET_SUCCESS;
} ## end worker.

my $semPW :shared;
sub process_watcher {
  lock $semPW;

  while ( !$die_early ) {
    usleep( 250_000 );
    {
      lock %PROCESS_WATCH;
      foreach ( keys %PROCESS_WATCH ) {
        unless ( ( defined $PROCESS_WATCH{$_} ) and
                 ( is_pid_alive( $_ ) ) and
                 ( ( time - $PROCESS_WATCH{$_} ) > TIMEOUT ) ) { next;
+ }
        tprint qq{process $_ exceeded timeout } .
                ( time - $PROCESS_WATCH{$_} ) ;
        kill( 9, $_ );
        $PROCESS_WATCH{$_} = undef;
      } ## end foreach.
    } ## end lock.
  } ## end while.
  tprint q{process_watcher is terminating};
  return RET_SUCCESS;
}

sub main {
  %RESULTS = ( 'success' => 0, 'failure' => 0 );

  my $Q = new Thread::Queue;

  $Q->enqueue( @SERVERS );
  $Q->enqueue( (undef) x THREADS );

  tprint q{queue populated};
  my $thr_pw = threads->create( \&process_watcher )->detach;
  my @threads = map threads->new( \&worker, $Q ), 1 .. THREADS;

  tprint q{workers started; waiting...};
  $_->join for @threads;

  print STDOUT q{*} x60;
  print STDOUT q{  Success: }, $RESULTS{'success'};
  print STDOUT q{  Failure: }, $RESULTS{'failure'};
  print STDOUT q{*} x60;
  print STDOUT q{Program complete};
  return RET_SUCCESS;
}

main();

exit EXIT_SUCCESS;
__END__

[download]

In reply to Re^2: making NET:SSH quiet by DeadPoet
in thread making NET:SSH quiet by Stoomy

Title:
Use:  <p> text here (a paragraph) </p>
and:  <code> code here </code>
to format your post, it's "PerlMonks-approved HTML":


  • Posts are HTML formatted. Put <p> </p> tags around your paragraphs. Put <code> </code> tags around your code and data!
  • Titles consisting of a single word are discouraged, and in most cases are disallowed outright.
  • Read Where should I post X? if you're not absolutely sure you're posting in the right place.
  • Please read these before you post! —
  • Posts may use any of the Perl Monks Approved HTML tags:
    a, abbr, b, big, blockquote, br, caption, center, col, colgroup, dd, del, details, div, dl, dt, em, font, h1, h2, h3, h4, h5, h6, hr, i, ins, li, ol, p, pre, readmore, small, span, spoiler, strike, strong, sub, summary, sup, table, tbody, td, tfoot, th, thead, tr, tt, u, ul, wbr
  • You may need to use entities for some characters, as follows. (Exception: Within code tags, you can put the characters literally.)
            For:     Use:
    & &amp;
    < &lt;
    > &gt;
    [ &#91;
    ] &#93;
  • Link using PerlMonks shortcuts! What shortcuts can I use for linking?
  • See Writeup Formatting Tips and other pages linked from there for more info.