File Coverage

blib/lib/Proc/JobQueue/Job.pm

Criterion	Covered	Total	%
statement	98	126	77.7
branch	43	80	53.7
condition	3	9	33.3
subroutine	19	22	86.3
pod	9	13	69.2
total	172	250	68.8

package Proc::JobQueue::Job;

use strict;
use warnings;
use Hash::Util qw(lock_keys);
use Carp qw(confess);
use Tie::Function::Examples qw(%q_shell);
use Callback;
use Proc::Background;
use Proc::JobQueue qw(is_remote_host);
use Scalar::Util qw(weaken);

our $debug = $Proc::JobQueue::debug;

sub new
{
	my ($pkg, %params) = @_;
	my $job = bless {
		desc			=> '',
		priority		=> 100,
		procbg			=> undef,
		run			=> undef,
		command			=> undef,
		queue			=> undef,
		jobnum			=> undef,
		postcb			=> undef,
		generate_command	=> undef,
		callback		=> undef,
		host			=> undef,
		jobnum			=> undef,
		on_failure		=> undef,
		errors			=> undef,
		status			=> 'queued',
		dependency_graph	=> undef,
		force_host		=> undef,
		%params
	}, $pkg;
	lock_keys(%$job);
	if ($job->{queue}) {
		$job->{queue}->add($job);
	}
	unless ($job->can_command || $job->can_callback) {
		confess "$pkg job needs a command or a callback";
	}
	unless ($job->{desc}) {
		if ($job->{command}) {
			$job->{desc} = $job->{command};
		} else {
			$job->{desc} = "$job"; # stringify
		}
	}
	return $job;
}

sub can_command
{
	my ($job) = @_;
	return 1 if $job->{command};
	return 1 if $job->{generate_command};
	return 1 if $job->can('command');
	return 0;
}

sub can_callback
{
	my ($job) = @_;
	return 1 if $job->{callback};
	return 1 if $job->can('startup');
	return 0;
}

sub start
{
	my ($job) = @_;

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

	$job->{status} = 'started';

	my $command = $job->{command}
		|| ($job->{generate_command} && $job->{generate_command}->($job))
		|| ($job->can('command') && $job->command())
		;

	if ($command) {
		$job->{desc} = $command
			unless $job->{desc};
		if (is_remote_host($host)) {
			$command = "ssh $host -o BatchMode=yes -o StrictHostKeyChecking=no $q_shell{$command}";
		}
		$job->{run} = $command
			unless $job->{run};
		print "+ $command\n";
		$job->{procbg} = Proc::Background->new($command);
	} elsif ($job->{callback}) {
		if (ref($job->{callback}) eq 'Callback') {
			$job->{callback}->call($job);
		} else {
			$job->{callback}->($job, $host, $jobnum, $queue);
		}
	} elsif ($job->can('startup')) {
		$job->startup($job, $host, $jobnum, $queue);
	} else {
		die "don't know how to start $job";
	}
}

sub host
{
	my ($job, $host) = @_;
	$job->{host} = $host if defined $host;
	return $job->{host};
}

sub jobnum
{
	my ($job, $jobnum) = @_;
	$job->{jobnum} = $jobnum if defined $jobnum;
	return $job->{jobnum};
}

sub queue
{
	my ($job, $queue) = @_;
	if ($queue) {
		$job->{queue} = $queue;
		weaken $job->{queue};
	}
	return $job->{queue};
}

sub runnable
{
	return 1;
}

sub checkjob
{
	my ($job) = @_;
	print STDERR "# checking up on $job->{jobnum} $job->{desc} on $job->{host}\n" if $debug > 6;
	unless ($job->{procbg}) {
		print STDERR "# $job->{jobnum} is not a Proc::Background job\n" if $debug > 9;
		return undef;
	}
	if ($job->{procbg}->alive) {
		print STDERR "# $job->{jobnum} $job->{desc} is still alive\n" if $debug > 6;
		return undef;
	}
	my $queue = $job->{queue};
	my $e = $job->{procbg}->wait;
	$e >>= 8;
	print "# $job->{desc} on $job->{host} finished\n";
	$job->finished($e);
	return $e;
}

sub finished
{
	my ($job, @exit_code) = @_;
	return if $job->{status} eq 'finished';
	$job->{status} = 'finished';
	die "NO JOBNUM FOR $job" unless $job->{jobnum};
	print STDERR "# FINISHED $job->{jobnum} $job->{desc} on $job->{host}\n" if $debug > 7;
	if ($job->{postcb}) {
		$_->call($job, @exit_code)
			for @{$job->{postcb}};
		delete $job->{postcb};  # may clean circular references
	}
	my $queue = $job->{queue};
	undef $job->{queue};
	if ($queue) {
		if ($job->{jobnum}) {
			print STDERR "# calling JOBDONE for $job->{jobnum} $job->{desc} ($job->{status})\n" if $debug > 5;
			$queue->jobdone($job, 0, @exit_code);   # not re-entrant since startmore == 0
		} else {
			print STDERR "# NOT calling JOBDONE for $job->{jobnum} $job->{desc} ($job->{status})\n" if $debug;
		}
	}
	if ($exit_code[0]) {
		print STDERR "# calling failed(@exit_code) for $job->{jobnum} $job->{desc}\n" if $debug > 6;
		$job->failed(@exit_code);
	} else {
		print STDERR "# calling success() for $job->{jobnum} $job->{desc}\n" if $debug > 7;
		$job->success();
		print STDERR "# done calling success() for $job->{jobnum} $job->{desc}\n" if $debug > 9;
	}
	$queue->startmore if $queue;	# can be re-entrant
}


sub success
{
	my ($job) = @_;
	print STDERR "# Empty success on $job->{jobnum} $job->{desc}\n" if $debug > 8;
}

sub addpostcb
{
	my ($job, $cb1, @more) = @_;
	my $cb = new Callback($cb1, @more);
	$job->{postcb} = []
		unless $job->{postcb};
	push(@{$job->{postcb}}, $cb);
}

sub failed
{
	my ($job, @exit_code) = @_;
	if ($job->{queue} && $job->{queue}{on_failure}) {
		$job->{queue}{on_failure}->($job, @exit_code);
	} else {
		die "job $job->{desc} failed with @exit_code";
	}
}

1;

__END__