File Coverage

blib/lib/Net/OpenSSH/Parallel.pm

Criterion	Covered	Total	%
statement	24	504	4.7
branch	0	328	0.0
condition	0	64	0.0
subroutine	8	41	19.5
pod	7	7	100.0
total	39	944	4.1

package Net::OpenSSH::Parallel;

our $VERSION = '0.12';

use strict;
use warnings;
use Carp qw(croak carp);

use Net::OpenSSH;
use Net::OpenSSH::Parallel::Constants qw(:error :on_error);

use POSIX qw(WNOHANG);
use Time::HiRes qw(time);
use Scalar::Util qw(dualvar);

our $debug;

sub new {
    my ($class, %opts) = @_;
    my $max_workers = delete $opts{workers};
    my $max_conns = delete $opts{connections};
    my $reconnections = delete $opts{reconnections};
    my $on_error = delete $opts{on_error};

    if ($max_conns) {
	if ($max_workers) {
	    $max_conns < $max_workers and
		croak "connections ($max_conns) < workers ($max_workers)";
	}
	else {
	    $max_workers = $max_conns;
	}
    }

    %opts and croak "unknonwn option(s): ". join(", ", keys %opts);

    my $self = { joins => {},
		 hosts => {},
		 host_by_pid => {},
		 ssh_master_by_pid => {},
		 in_state => {
			      connecting => {},
			      ready => {},
			      running => {},
			      done => {},
			      waiting => {},
			      suspended => {},
			      join_failed => {},
			     },
		 connected => { suspended => {},
				waiting => {},
				join_failed => {},
			      },
		 joins => {},
		 max_workers => $max_workers,
		 max_conns => $max_conns,
		 num_conns => 0,
		 reconnections => $reconnections,
		 on_error => $on_error,
	       };
    bless $self, $class;
    $self;
}

my %debug_channel = (api => 1, state => 2, select => 4, at => 8,
		     action => 16, join => 32, workers => 64,
		     connect => 128, conns => 256, error => 512);

sub _debug {
    my $channel = shift;
    my $bit = $debug_channel{$channel}
	or die "internal error: bad debug channel $channel";
    if ($debug & $debug_channel{$channel}) {
	print STDERR sprintf("%6.3f", (time - $^T)), "| ",
	    join('', map { defined($_) ? $_ : '' } @_), "\n";
    }
}

sub add_host {
    my $self = shift;
    my $label = shift;
    $label =~ /([,*!()<>\/{}])/ and croak "invalid char '$1' in host label";
    my %opts = (@_ & 1 ? (host => @_) : @_);
    $opts{host} = $label unless defined $opts{host};
    $opts{batch_mode} = 1 unless defined $opts{batch_mode};

    my $on_error = delete $opts{on_error};
    my $reconnections = delete $opts{reconnections};

    my $host = { label => $label,
		 workers => 1,
		 opts => \%opts,
		 ssh => undef,
		 state => 'done',
		 queue => [],
		 on_error => $on_error,
		 reconnections => $reconnections,
	       };

    $self->{hosts}{$label} = $host;
    $debug and _debug(api => "[$label] added ($host)");
    $self->{in_state}{done}{$label} = 1;
    $debug and _debug(state => "[$label] state set to done");
}

sub _set_host_state {
    my ($self, $label, $state) = @_;
    my $host = $self->{hosts}{$label};
    my $old = $host->{state};
    delete $self->{in_state}{$old}{$label}
	or die "internal error: host $label is in state $old but not in such queue";
    delete $self->{connected}{$old}{$label}
	if ($old eq 'suspended' or $old eq 'waiting' or $old eq 'join_failed');

    $self->{in_state}{$state}{$label} = 1;
    $host->{state} = $state;
    $debug and _debug(state => "[$label] state changed $old --> $state");

    if ($host->{ssh} and ($state eq 'suspended' or
			  $state eq 'waiting' or
			  $state eq 'join_failed')) {
	$self->{connected}{$state}{$label} = 1;
	$debug and _debug(state => "[$label] host is connected");
    }
}

my %sel2re_cache;

sub _selector_to_re {
    my ($self, $part) = @_;
    $sel2re_cache{$part} ||= do {
	$part = quotemeta $part;
	$part =~ s/\\\*/.*/g;
	qr/^$part$/;
    }
}

sub _select_labels {
    my ($self, $selector) = @_;
    my %sel;
    my @parts = split /\s*,\s*/, $selector;
    for (@parts) {
	my $re = $self->_selector_to_re($_);
	$sel{$_} = 1 for grep $_ =~ $re, keys %{$self->{hosts}};
    }
    my @labels = keys %sel;
    $debug and _debug(select => "selector($selector) --> [", join(', ', @labels), "]");
    return @labels;
}

sub all { shift->push('*', @_) }

my %push_action_alias = (get  => 'scp_get',
                         put  => 'scp_put',
                         psub => 'parsub',
                         cmd  => 'command');

my %push_min_args = ( here      => 1,
                      goto      => 1,
                      stop      => 0,
                      sub       => 1,
                      parsub    => 1,
                      scp_get   => 2,
                      scp_put   => 2,
                      rsync_get => 2,
                      rsync_put => 2 );

my %push_max_args =  ( here     => 1,
                       goto     => 1,
                       stop     => 0 );

sub push {
    my $self = shift;
    my $selector = shift;
    my $action = shift;
    my $in_state = $self->{in_state};

    if (ref $action eq 'CODE') {
	unshift @_, $action;
	$action = 'sub';
    }

    my $alias = $push_action_alias{$action};
    $action = $alias if defined $alias;

    $action =~ /^(?:command|(?:(?:rsync|scp)_(?:get|put))|join|sub|parsub|here|stop|goto|_notify)$/
	or croak "bad action '$action'";

    my %opts = (($action ne 'sub' and ref $_[0] eq 'HASH') ? %{shift()} : ());
    %opts and grep($action eq $_, qw(join here))
        and croak "unsupported option(s) '" . join("', '", keys %opts) . "' in $action action";

    my @labels = $self->_select_labels($selector);

    my $max = $push_max_args{$action};
    croak "too many parameters for action $action"
        if (defined $max and $max < @_);

    my $min = $push_min_args{$action};
    croak "too few parameters for action $action"
        if (defined $min and $min > @_);

    if ($action eq 'join') {
	my $notify_selector = shift @_;
	my $join = { id => '#' . $self->{join_seq}++,
		     depends => {},
		     notify => [] };
	my @depends = $self->push($notify_selector, _notify => {}, $join)
	    or do {
		$join->_debug(join => "join $join->{id} does not depend on anything, ignoring!");
		return ();
	    };
	$join->{depends}{$_} = 1 for @depends;

	for my $label (@labels) {
	    my $host = $self->{hosts}{$label};
	    push @{$host->{queue}}, [join => {}, $join];
	    $debug and _debug(api => "[$label] join $join->{id} queued");
	    $self->_set_host_state($label, 'ready')
		if $in_state->{done}{$label};
	}
    }
    else {
	for my $label (@labels) {
	    my $host = $self->{hosts}{$label};
	    push @{$host->{queue}}, [$action, \%opts, @_];
	    $debug and _debug(api => "[$label] action $action queued");
	    $self->_set_host_state($label, 'ready')
		if $in_state->{done}{$label};
	}
    }
    @labels;
}

sub _audit_conns {
    my $self = shift;
    my $hosts = $self->{hosts};
    my $num = 0;
    $num++ for grep $_->{ssh}, values %$hosts;
    $debug and _debug(conns => "audit_conns counted: $num, saved: $self->{num_conns}");
    $num == $self->{num_conns}
	or die "internal error: wrong number of connections, counted: $num, saved: $self->{num_conns}";
    my $in_state = $self->{in_state};
    for my $state (keys %$in_state) {
	my $num = 0;
	$num++ for grep $hosts->{$_}{ssh}, keys %{$in_state->{$state}};
	my $total = keys %{$in_state->{$state}};
	print STDERR "conns in state $state: $num of $total\n";
    }
}

sub _hash_chain_get {
    my $name = shift;
    for (@_) {
	if (defined $_) {
	    my $v = $_->{$name};
	    return $v if defined $v;
	}
    }
    undef;
}

sub _at_error {
    my ($self, $label, $error) = @_;
    my $host = $self->{hosts}{$label};
    my $task = delete $host->{current_task};
    my $queue = $host->{queue};

    $debug and _debug(error => "_at_error label: $label, error: $error");

    my $opts;
    $opts = $task->[1] if $task;

    my $on_error;
    if ($error == OSSH_MASTER_FAILED) {
        if ($host->{state} eq 'connecting') {
            # task is not set in state connecting!
            $task and die "internal error: task is defined in state connecting";
            $opts = $queue->[0][1] if @$queue;
        }
	my $max_reconnections = _hash_chain_get(reconnections => $opts, $host, $self) || 0;
	my $reconnections = $host->{current_task_reconnections}++ || 0;
	$debug and _debug(error => "[$label] reconnection: $reconnections, max: $max_reconnections");
	if ($reconnections < $max_reconnections) {
	    $debug and _debug(error => "[$label] will reconnect!");
	    $on_error = OSSH_ON_ERROR_RETRY;
	}
    }
    $on_error ||= _hash_chain_get(on_error => $opts, $host, $self);

    if (ref $on_error eq 'CODE') {
	if ($error == OSSH_JOIN_FAILED) {
	    $on_error = $on_error->($self, $label, $error);
	}
	else {
	    $on_error = $on_error->($self, $label, $error, $task);
	}
    }

    $on_error = OSSH_ON_ERROR_ABORT if (not defined $on_error or
                                        $error == OSSH_ABORTED);

    $debug and _debug(error => "[$label] on_error (final): $on_error, error: $error (".($error+0).")");

    if ($on_error == OSSH_ON_ERROR_RETRY) {
	if ($error == OSSH_MASTER_FAILED) {
	    $self->_set_host_state($label, 'suspended');
	    $self->_disconnect_host($label);
	    $self->_set_host_state($label, 'ready');
	}
	elsif ($error == OSSH_GOTO_FAILED) {
            # No way to retry after a GOTO error!
            # That should probably croak, but that would leave unmanaged
            # processes running
            $on_error = OSSH_ON_ERROR_ABORT;
        }
        else {
	    unshift @$queue, $task;
	}
	return;
    }

    delete $host->{current_task_reconnections};

    if ($on_error == OSSH_ON_ERROR_IGNORE) {
	if ($error == OSSH_MASTER_FAILED) {
	    # stablishing a new connection failed, what we should do?
	    # currently we remove the current task from the queue and
	    # continue.
	    shift @$queue unless $task;
	    $self->_set_host_state($label, 'suspended');
	    $self->_disconnect_host($label);
	    $self->_set_host_state($label, 'ready');
	}
        else {
	    $self->_set_host_state($label, 'ready');
        }
	# else do nothing
    }
    else {
        unless ($on_error == OSSH_ON_ERROR_DONE or
                $on_error == OSSH_ON_ERROR_ABORT or
                $on_error == OSSH_ON_ERROR_ABORT_ALL) {
            carp "bad on_error code $on_error";
            $on_error = OSSH_ON_ERROR_ABORT;
        }
	my $queue = $host->{queue};
	my $failed = ($on_error != OSSH_ON_ERROR_DONE);
	$debug and _debug(error => "[$label] dropping queue, ", scalar(@$queue), " jobs");
	while (my $task = shift @$queue) {
	    my ($action, undef, $join) = @$task;
	    $debug and _debug(error => "[$label] remove action $action from queue");
	    $self->_join_notify($label, $join, $failed)
		if $action eq '_notify';
	}

 	$on_error == OSSH_ON_ERROR_ABORT_ALL
 	    and $self->{abort_all} = 1;

	$self->_set_host_state($label, 'done');
	$self->_disconnect_host($label);
	$host->{error} = $error;
    }
}

sub _at_connect {
    my ($self, $label) = @_;
    my $host = $self->{hosts}{$label};
    $debug and _debug(connect => "[$label] _connect, starting SSH connection");
    $host->{ssh} and die "internal error: connecting host is already connected";
    my $ssh = $host->{ssh} = Net::OpenSSH->new(expand_vars => 1,
					       %{$host->{opts}},
					       async => 1);
    $ssh->set_var(LABEL => $label);
    my $master_pid = $ssh->get_master_pid;
    $host->{master_pid} = $master_pid;
    $self->{ssh_master_by_pid}{$master_pid} = $label;
    $self->{num_conns}++;
    $self->_set_host_state($label, 'connecting');
    if ($ssh->error) {
	$self->_at_error($label, $ssh->error);
    }
}

sub _at_connecting {
    my ($self, $label) = @_;
    my $host = $self->{hosts}{$label};
    $debug and _debug(at => "[$label] at_connecting, waiting for master");
    my $ssh = $host->{ssh};
    if ($ssh->wait_for_master(1)) {
	$debug and _debug(at => "[$label] at_connecting, master connected");
	$self->_set_host_state($label, 'ready');
    }
    elsif ($ssh->error) {
	$self->_at_error($label, $ssh->error);
    }
}

sub _join_notify {
    my ($self, $label, $join, $failed) = @_;
    # use Data::Dumper;
    # print STDERR Dumper $join;
    delete $join->{depends}{$label}
	or die "internal error: $join->{id} notified for non dependent label $label";
    $debug and _debug(join => "removing dependent $label from join $join->{id}");
    $join->{failed} = 1 if $failed;
    if (not %{$join->{depends}}) {
	$debug and _debug(join => "join $join->{id} done");
	$join->{done} = 1;
	my $failed = $join->{failed};
	for my $label (@{$join->{notify}}) {
	    $debug and _debug(join => "notifying $label about join $join->{id} done");
	    $self->_set_host_state($label, $failed ? 'join_failed' : 'ready');
	}
    }
    # print STDERR Dumper $join;
}

sub _num_workers {
    my $in_state = shift->{in_state};
    ( keys(%{$in_state->{ready}}) +
      keys(%{$in_state->{running}}) +
      keys(%{$in_state->{connecting}}) );
}

sub _disconnect_host {
    my ($self, $label) = @_;
    my $host = $self->{hosts}{$label};
    my $state = $host->{state};
    $state =~ /^(?:waiting|suspended|done|connecting)$/
	or die "internal error: disconnecting $label in state $state";
    if ($host->{ssh}) {
	$debug and _debug(connect => "[$label] disconnecting host");
	my $master_pid = delete $host->{master_pid};
	delete $self->{ssh_master_by_pid}{$master_pid}
	    if defined $master_pid;
	undef $host->{ssh};
	$self->{num_conns}--;
	$self->_set_host_state($label, $state);
    }
}

sub _disconnect_any_host {
    my $self = shift;
    my $connected = $self->{connected};
    $debug and _debug(conns => "disconnect any host");
    # $self->_audit_conns;
    my $label;
    for my $state (qw(suspended join_failed waiting)) {
	# use Data::Dumper;
	# print Dumper $connected;
	$debug and _debug(conns => "looking for connected host in state $state");
	($label) = each %{$connected->{$state}};
	keys %{$connected->{$state}}; # reset iterator
	last if defined $label;
    }
    $debug and _debug(conns => "[$label] disconnecting");
    defined $label or die "internal error: unable to disconnect any host";
    $self->_disconnect_host($label);
}

my @private_opts = qw(on_error or_goto reconnections);

sub _at_ready {
    my ($self, $label) = @_;
    if (my $max_workers = $self->{max_workers}) {
	my $in_state = $self->{in_state};
	my $num_workers = $self->_num_workers;
	$debug and _debug(workers => "num workers: $num_workers, maximun: $max_workers");
	if ($num_workers > $max_workers) {
	    $debug and _debug(workers => "[$label] suspending");
	    $self->_set_host_state($label, 'suspended');
	    return;
	}
    }

    my $host = $self->{hosts}{$label};
    $debug and _debug(at => "[$label] at_ready");

    my $queue = $host->{queue};

    if ($self->{abort_all}) {
        $self->_at_error($label, OSSH_ABORTED);
        return;
    }

    while (defined (my $task = shift @$queue)) {
	my $action = shift @$task;
	$debug and _debug(at => "[$label] at_ready, starting new action $action");
	if ($action eq 'join') {
	    my (undef, $join) = @$task;
	    if ($join->{done}) {
		$debug and _debug(join => "join[$join->{id}] is done");
		if ($join->{failed}) {
		    $self->_at_error($label, OSSH_JOIN_FAILED);
		    return;
		}
		$debug and _debug(action => "[$label] join $join->{id} already done");
		next;
	    }
	    CORE::push @{$join->{notify}}, $label;
	    $self->_set_host_state($label, 'waiting');
	    return;
	}
        elsif ($action eq 'here') {
            next;
        }
        elsif ($action eq 'stop') {
            $self->_skip($label, 'END');
            next;
        }
        elsif ($action eq 'goto') {
            my (undef, $target) = @$task;
            $self->_skip($label, $target);
            next;
        }
	elsif ($action eq '_notify') {
	    my (undef, $join) = @$task;
	    $self->_join_notify($label, $join);
	    next;
	}
	elsif ($action eq 'sub') {
            # use Data::Dumper;
            # _debug (action => Dumper(@$task));
	    shift @$task;
	    my $sub = shift @$task;
	    $debug and _debug(action => "[$label] calling sub $sub");
	    $sub->($self, $label, @$task);
	    next;
	}
	else {
            my $ssh = $host->{ssh};
            unless ($action eq 'parsub' and $task->[0]{no_ssh}) {
                unless ($ssh) {
                    # unshift the task we have just removed and connect first:
                    unshift @$task, $action;
                    unshift @$queue, $task;
                    if (my $max_conns = $self->{max_conns}) {
                        $self->_disconnect_any_host() if $self->{num_conns} >= $max_conns;
                    }
                    $debug and _debug(at => "[$label] host is not connected, connecting...");
                    $self->_at_connect($label);
                    return;
                }

                if (my $error = $ssh->error) {
                    $self->_at_error($label, $error);
                    return;
                }
            }

            $host->{current_task} = [$action, @$task];
            my %opts = %{shift @$task};
            delete @opts{@private_opts};
            my $method = $self->can("_start_$action")
                or die "internal error: method _start_$action not found";
            my $pid = $method->($self, $label, \%opts, @$task);
            $debug and _debug(action => "[$label] action pid: ", $pid);
            unless (defined $pid) {
                my $error = (($ssh && $ssh->error) ||
                             dualvar(($action eq 'parsub'
                                      ? "Unable to fork parsub"
                                      : "Action $action failed to start"), OSSH_SLAVE_FAILED));
                $self->_at_error($label, $error);
                return;
            }
            $self->{host_by_pid}{$pid} = $label;
            $self->_set_host_state($label, 'running');
            return;
        }
    }
    $debug and _debug(at => "[$label] at_ready, queue_is_empty, we are done!");
    $self->_set_host_state($label, 'done');
    $self->_disconnect_host($label);
}

sub _start_parsub {
    my $self = shift;
    my $label = shift;
    my $opts = shift;
    my $sub = shift;
    my $ssh = $self->{hosts}{$label}{ssh};
    $debug and _debug(action => "[$label] start parsub action [@_]");
    my $pid = fork;
    unless ($pid) {
        defined $pid or return;
        eval { $sub->($label, $ssh, @_) };
        $@ and $debug and _debug(error => "slave died on parsub: $@");
        POSIX::_exit($@ ? 1 : 0);
    }
    $pid;
}

sub _start_command {
    my $self = shift;
    my $label = shift;
    my $opts = shift;
    my $ssh = $self->{hosts}{$label}{ssh};
    $debug and _debug(action => "[$label] start command action [@_]");
    $ssh->spawn($opts, @_);
}

sub _start_scp_get {
    my $self = shift;
    my $label = shift;
    my $opts = shift;
    my $ssh = $self->{hosts}{$label}{ssh};
    $debug and _debug(action => "[$label] start scp_get action");
    $opts->{async} = 1;
    $ssh->scp_get($opts, @_);
}

sub _start_scp_put {
    my $self = shift;
    my $label = shift;
    my $opts = shift;
    my $ssh = $self->{hosts}{$label}{ssh};
    $debug and _debug(action => "[$label] start scp_put action");
    $opts->{async} = 1;
    $ssh->scp_put($opts, @_);
}

sub _start_rsync_get {
    my $self = shift;
    my $label = shift;
    my $opts = shift;
    my $ssh = $self->{hosts}{$label}{ssh};
    $debug and _debug(action => "[$label] start rsync_get action");
    $opts->{async} = 1;
    $ssh->rsync_get($opts, @_);
}

sub _start_rsync_put {
    my $self = shift;
    my $label = shift;
    my $opts = shift;
    my $ssh = $self->{hosts}{$label}{ssh};
    $debug and _debug(action => "[$label] start rsync_put action");
    $opts->{async} = 1;
    $ssh->rsync_put($opts, @_);
}

# FIXME: dead code?
sub _start_join {
    my $self = shift;
    my $label = shift;
    warn "internal mismatch: this shouldn't be happening!";
}

sub _skip {
    my ($self, $label, $target) = @_;
    $debug and _debug(action => "skipping until $target");
    my $host = $self->{hosts}{$label};
    my $queue = $host->{queue};
    my ($ix, $task);
    for ($ix = 0; defined($task = $queue->[$ix]); $ix++) {
        last if ($task->[0] eq 'here' and $task->[2] eq $target);
    }
    if ($task or $target eq 'END') {
        for (1..$ix) {
            my $task = shift @$queue;
            $self->_join_notify($label, $task->[2])
                if $task->[0] eq '_notify';
        }
        return;
    }
    $debug and _debug(action => "here label $target not found");
    $self->_at_error($label, OSSH_GOTO_FAILED);
}

sub _finish_task {
    my ($self, $pid) = @_;
    my $label = delete $self->{host_by_pid}{$pid};

    if (defined $label) {
	$debug and _debug(action => "[$label] action finished pid: $pid, rc: $?");
	my $host = $self->{hosts}{$label};
        my $or_goto;
        if ($?) {
            my ($action) = @{$host->{current_task}};
            my $rc = ($? >> 8);
            my $ssh = $host->{ssh} or die "internal error: $label is not connected";
            my $error = $ssh->error;
            $or_goto = $host->{current_task}[1]{or_goto} unless ($error or $rc == 255);
            if (defined $or_goto) {
                $debug and _debug(action => "[$label] skipping to $or_goto with rc = $rc");
            }
            else {
                $error ||= dualvar(OSSH_SLAVE_FAILED,
                                   "child exited with non-zero return code ($rc)");
                $self->_at_error($label, $error);
                return
            }
        }
        $self->_set_host_state($label, 'ready');
        $self->_skip($label, $or_goto) if defined $or_goto;
        delete $host->{current_task};
        delete $host->{current_task_reconnections};
    }
    else {
	my $label = delete $self->{ssh_master_by_pid}{$pid};
	defined $label or carp "spurious child exit (pid: $pid)";

        $debug and _debug(action => "[$label] master ssh exited");
        my $host = $self->{hosts}{$label};
        my $ssh = $host->{ssh}
            or die ("internal error: master ssh process exited but ".
                    "there is no ssh object associated to host $label");
        $ssh->master_exited;
        my $state = $host->{state};
        # do error handler later...
    }
}

sub _wait_for_jobs {
    my ($self, $time) = @_;
    my $dontwait = ($time == 0);
    $debug and _debug(at => "_wait_for_jobs time: $time");
    # This loop is here because we want to call waitpit before and
    # after the select. If we find some child has exited in the first
    # round we don't call select at all and return immediately
    while (1) {
	if (%{$self->{in_state}{running}}) {
	    $debug and _debug(at => "_wait_for_jobs reaping children");
	    while (1) {
		my $pid = waitpid(-1, WNOHANG);
		last if $pid <= 0;
		$debug and _debug(action => "waitpid caught pid: $pid, rc: $?");
		$dontwait = 1;
		$self->_finish_task($pid);
	    }
	}
	$dontwait and return 1;
	$debug and _debug(at => "_wait_for_jobs calling select");
	{
	    # This is a hack to make select finish as soon as we get a
	    # CHLD signal.
	    local $SIG{CHLD} = sub {};
	    select(undef, undef, undef, $time);
	}
	$dontwait = 1;
    }
}

sub _clean_errors {
    my $self = shift;
    delete $_->{error} for values %{$self->{hosts}};
}

sub run {
    my ($self, $time) = @_;

    $self->_clean_errors;

    my $hosts = $self->{hosts};
    my $max_workers = $self->{max_workers};
    my ($connecting, $ready, $running, $waiting, $suspended, $join_failed, $done) =
	@{$self->{in_state}}{qw(connecting ready running waiting suspended join_failed done)};
    my $connected_suspended = $self->{connected}{suspended};
    while (1) {
	# use Data::Dumper;
	# print STDERR Dumper $self;
	$debug and _debug(api => "run: iterating...");

	$debug and _debug(at => "run: hosts at connecting: ", scalar(keys %$connecting));
	$self->_at_connecting($_) for keys %$connecting;

	$debug and _debug(at => "run: hosts at ready: ", scalar(keys %$ready));

	# $self->_audit_conns;
	$self->_at_ready($_) for keys %$ready;
	# $self->_audit_conns;

	$debug and _debug(at => 'run: hosts at join_failed: ', scalar(keys %$join_failed));
	$self->_at_error($_, OSSH_JOIN_FAILED) for keys %$join_failed;

	if ($max_workers) {
	    $debug and _debug(at => "run: hosts at suspended:", scalar(keys %$suspended));
	    if (%$suspended) {
		my $awake = $max_workers - $self->_num_workers;
		my @labels;
		for my $hash ($connected_suspended, $suspended) {
		    while ($awake > 0) {
			my ($label) = each %$hash or last;
			CORE::push @labels, $label;
			$awake--;
		    }
		    for my $label (@labels) {
			$debug and _debug(workers => "[$label] awaking");
			$self->_set_host_state($label, 'ready');
		    }
		    keys %$hash; # do we really need to reset the each iterator?
		}
	    }
	}

	$debug and _debug(at => "run: hosts at waiting: ", scalar(keys %$waiting));
	$debug and _debug(at => "run: hosts at running: ", scalar(keys %$running));
	$debug and _debug(at => "run: hosts at done: ", scalar(keys %$done), " of ", scalar(keys %$hosts));

	last if keys(%$hosts) == keys(%$done);

	my $time = ( %$ready      ? 0   :
		     %$connecting ? 0.3 :
		                    5.0 );
	$self->_wait_for_jobs($time);
    }

    delete $self->{abort_all};

    my $error;
    for my $label (sort keys %$hosts) {
	$hosts->{$label}{error} and $error = 1;
	$debug and _debug(error => "[$label] error: ", $hosts->{$label}{error});
    }
    !$error;
}

sub get_error {
    my ($self, $label) = @_;
    my $host = $self->{hosts}{$label}
	or croak "no such host $label has been added";
    $host->{error}
}

sub get_errors {
    my $self = shift;
    if (wantarray) {
        return map {
            my $error = $self->get_error($_);
            defined $error ? ($_ => $error) : ()
        } sort keys %{$self->{hosts}}
    }
    else {
        return grep defined($self->get_error($_)), keys %{$self->{hosts}}
    }
}

1;

__END__