[Expectperl-discuss] Yet Another Threading Example

[Bryan B. <br...@bu...>]

Ok, still playing around with threading on top of Expect, and I've
designed what I think is a pretty good solution.  However there is much I
dont understand and I'm looking for feedback to make sure i'm not doing
something inherently wrong.

The way I've been able to do this is by pre-allocating pty's for jobs I'm
going to need, in this case i'm ssh'ing so i need one pty per host.

After the pty's are set up, i fork off one or more worker process who
reads in a list of command from a pipe.  When the worker gets a hostname
it then forks another process who ties STDIN/OUT/ERR to the pty for that
host and exec's the ssh command.

The main process controls the threading and manages the workers.  In my
example i dont do much error checking at all, but in the even the thread
returned a bad result, i could probably HUP the worker process and have it
destroy its child.  That way i dont loose a worker when its child process
hangs.

Anyway, here is my code, please let me know what you think.


Thanks

Bryan Bueter

-->

#!/usr/bin/perl

use strict;
use threads;
use IO::Pty;
use Expect;
$Expect::Log_Stdout = 0;

my $prompt = '[>#\$] $';
my @hosts = ("host1", "host2", "host3", "host4", "host5", "host6",
"host7", "host8", "host9"); # Aliases to localhost
my @commands = ("uptime", "who", "who am i", "pwd");

# Pre-allocate ptys
my %ptys = ();
foreach (@hosts) { $ptys{$_} = new IO::Pty; }

# Start worker processes
my @workers = ();
my $workers = 4;
for (my $i=0; $i<$workers; $i++) {
  pipe(my $from_parent, my $to_child);
  unless(fork()) {
    close($to_child);
    worker($from_parent);
  }
  close($from_parent);
  push(@workers, $to_child);
}

# Scan and feed worker processes hosts to spawn ssh sessions
my @threads;
while (@hosts) {
  for (my $i=0; $i<@workers; $i++) {
    if ( !defined($threads[$i]) || $threads[$i]->is_joinable() ) {
      $threads[$i]->join() if defined($threads[$i]);

      my $host = shift @hosts;
      print { $workers[$i] } $host."\n";
      $ptys{$host}->close_slave();
      $ptys{$host}->set_raw();

      $threads[$i] = threads->new("run_threaded", $host);
      last;
    }
  }
}

# Out of hosts join everyone back in and clean up
foreach (@threads) { $_->join(); }
foreach (@workers) { close $_ ; }
wait();
## END OF MAIN ###

# Expect on the pty and run the @commands
sub run_threaded {
  my $host = shift;

  my $exp = Expect->exp_init($ptys{$host});
  $exp->log_file("$host.log");
  $exp->expect(5, '-re', $prompt) or return 0;

  foreach (@commands) {
    print "Executing $_ on $host\n";
    $exp->send($_."\n");
    $exp->expect(5, '-re', $prompt) or return 0;
  }

  $exp->send("exit;\nexit;\n;exit;\n");
  $exp->hard_close();
  return 1;
}

# Worker waits for hostnames and spawns the ssh command
sub worker {
  my $pipe = shift;

  while (<$pipe>) {
    chomp;
    my $host = $_;

    unless(fork()) {
      my $pty = $ptys{$host};

      $pty->make_slave_controlling_terminal();
      my $slave = $pty->slave();
      close($pty);
      $slave->clone_winsize_from(\*STDIN);
      $slave->set_raw();
      open(STDIN, "<&". $slave->fileno());
      open(STDOUT, ">&". $slave->fileno());
      open(STDERR, ">&". $slave->fileno());
      close($slave);

      { exec("ssh -o StrictHostKeyChecking=no -x $host"); }
      die "Cannot exec ssh: $!";
    }
    wait();
  }

  exit(0);
}

<--