File Coverage

blib/lib/Schedule/Load.pm

Criterion	Covered	Total	%
statement	108	125	86.4
branch	16	40	40.0
condition	7	18	38.8
subroutine	22	25	88.0
pod			n/a
total	153	208	73.5

# Load.pm -- Schedule load management
# See copyright, etc in below POD section.
######################################################################

require 5.005;
package Schedule::Load;
require Exporter;
@ISA = ('Exporter');
@EXPORT = qw( );
@EXPORT_OK = qw(_min _max _nfreeze _nthaw _pfreeze _pthaw);
%EXPORT_TAGS = (_utils => \@EXPORT_OK);

use vars qw($VERSION $Debug %Machines %_Default_Params $Default_Port @Default_Hosts);

use IO::Pipe;
use IO::File;
use IO::Socket;
use Sys::Hostname;
use Storable qw();
use Socket;
require Exporter;
BEGIN { eval 'use Data::Dumper; $Data::Dumper::Indent=1;';}	#Ok if doesn't exist: debugging only
use POSIX qw (EWOULDBLOCK BUFSIZ);
use strict;
use Carp;

######################################################################
#### Configuration Section

$VERSION = '3.064';
$Debug = 0;

%_Default_Params = (
		    min_pctcpu=>3,
		    port=>((defined $ENV{SLCHOOSED_PORT})	# Debugging
			   ? $ENV{SLCHOOSED_PORT}
			   : (getservbyname ('slchoosed',"")
			      ? 'slchoosed' : 1752)),
		    dhost=> [(defined $ENV{SLCHOOSED_HOST})
			     ? split ':', $ENV{SLCHOOSED_HOST}
			     : qw(localhost)],
		    req_retries=>3,		# Number of tries before we bail out
		    req_retry_delay=>10,	# After being closed, delay for failover server to come up
		    );

######################################################################
#### Internal utilities

sub _subprocesses {
    my $parent = shift || $$;
    my $pt = shift;  # Generally undef, unless happen to have process table already
    # All pids under the given parent
    # Used by testing module
    # Same function in Parallel::Forker::_subprocesses
    use Proc::ProcessTable;
    if (!$pt) {
	$pt = new Proc::ProcessTable( 'cache_ttys' => 1);
    }
    my %parent_pids;
    foreach my $p (@{$pt->table}) {
	$parent_pids{$p->pid} = $p->ppid;
    }
    my @out;
    my @search = ($parent);
    while ($#search > -1) {
	my $pid = shift @search;
	push @out, $pid if $pid ne $parent;
	foreach (keys %parent_pids) {
	    push @search, $_ if $parent_pids{$_} == $pid;
	}
    }
    return @out;
}

sub _min {
    return $_[0] if (!defined $_[1]);
    return $_[1] if (!defined $_[0]);
    return $_[0] if ($_[0] <= $_[1]);
    return $_[1];
}
sub _max {
    return $_[0] if (!defined $_[1]);
    return $_[1] if (!defined $_[0]);
    return $_[0] if ($_[0] >= $_[1]);
    return $_[1];
}

sub _pfreeze {
    my $cmd = shift;
    my $ref = shift;
    my $debug = shift;

    my $serialized = $cmd . " " . unpack ("h*", Storable::nfreeze $ref) . "\n";
    if ($debug) {
	printf "AFREEZE $cmd: %s\n", Data::Dumper::Dumper($ref);
    }
    return $serialized;
}

sub _pthaw {
    my $line = shift;
    my $debug = shift;

    $line =~ /^(\S+)\s*(\S*)/;
    my $cmd = $1; my $serialized = $2;

    my $ref;
    if ($serialized) {
	eval {
	    # Tolerate storable version mismatches.  The error will look like
	    # "Storable binary image v2.7 more recent then I am"
	    $ref = Storable::thaw(pack ("h*", $serialized));
	};
	if (!$ref && $@) {
	    $cmd = "THAW_ERROR-".$@;
	}
    }
    if ($debug) {
	print "$cmd: ", Data::Dumper::Dumper($ref);
    }
    return ($cmd, $ref);
}

######################################################################
######################################################################
######################################################################
#### Internal socket class, so we can override NEW

package Schedule::Load::Socket;
use IO::Socket;
use Time::HiRes qw(usleep gettimeofday);

use strict;
use vars qw(@ISA);
@ISA = qw(IO::Socket::INET);

sub new {
    my $class = shift;

    my %params = (@_);
    # There is a bug in the socket that it requires untainted peer address
    # it will just silently fail if you give it a tainted host name
    if ($params{PeerAddr}) {
	$params{PeerAddr} =~ /([a-z0-9A-Z._-]*)/; $params{PeerAddr}=$1;	# Untaint
    }
    if ($params{PeerPort}) {
	$params{PeerPort} =~ /([a-z0-9A-Z._-]*)/; $params{PeerPort}=$1;	# Untaint
    }

    my $fh;
    $? = 0;
    {
	local $SIG{__WARN__} = sub {
	    return if $_[0] =~ /Connection refused/;
	    warn @_;
	};
	$fh = $class->SUPER::new(
				 Proto => 'tcp',
				 %params);
    }
    $fh = undef if $?;
    return $fh;
}

sub send_and_check {
    my $fh = shift;
    my $out = join "", @_;
    # Send any arguments to the filehandle
    # Returns 0 if failed, else 1
    while ($out ne "") {
	if (!$fh || !$fh->connected()) {
	    return 0;
	}
	my $rv = eval { return $fh->syswrite($out); };
	# Node ->connected call does a system getpeeraddr() call
	if (!$fh || !$fh->connected() || ($! && $! != POSIX::EWOULDBLOCK)) {
	    return 0;
	}
	if (!defined $rv) {  usleep 1000; next; }  # Couldn't write: very rare
	# Truncate what did get out
	$out = substr ($out, $rv);
    }
    return 1;
}

package Schedule::Load;

######################################################################
######################################################################
######################################################################
#### Package return
1;

######################################################################
__END__