File Coverage

blib/lib/Proc/JobQueue/DependencyQueue.pm

Criterion	Covered	Total	%
statement	18	48	37.5
branch	0	14	0.0
condition	0	9	0.0
subroutine	6	11	54.5
pod	1	3	33.3
total	25	85	29.4

package Proc::JobQueue::DependencyQueue;

use strict;
use warnings;
use Carp qw(confess);
require Proc::JobQueue;
use Time::HiRes qw(time);
use Object::Dependency;
require POSIX;

our @ISA = qw(Proc::JobQueue);

our $timer_interval = 6;
my $debug = 0;

sub new
{
	my ($pkg, %params) = @_;

	$params{dependency_graph} ||= Object::Dependency->new();

	my $queue = $pkg->SUPER::new(
		startmore_in_progress	=> 0,
		on_failure		=> \&on_failure,
		%params
	);

	if (defined(&IO::Event::unloop_all)) {
		my $last_dump = time;

		my $timer = IO::Event->timer(
			interval	=> $params{timer_interval} || $timer_interval,
			cb		=> sub {
				print STDERR "beep!\n" if $debug;
				eval {
					$queue->startmore;
				};
				if ($@) {
					print STDERR "DIE DIE DIE DIE DIE (DT1): $@";
					# exit 1; hangs
					POSIX::_exit(1);
				};
				if ($debug && time > $last_dump + $timer_interval) {
					$params{dependency_graph}->dump_graph();
					$last_dump = time;
				}
				use POSIX ":sys_wait_h";
				my $k;
				do { $k = waitpid(-1, WNOHANG) } while $k > 0;
			},
		);

		$Event::DIED = sub {
			Event::verbose_exception_handler(@_);
			IO::Event::unloop_all();
		};
	}

	return $queue;
}

sub unloop
{
	my ($queue) = @_;
	if (defined(&IO::Event::unloop_all)) {
		IO::Event::unloop_all();
	}
}

sub on_failure
{
	my ($queue, $job, @exit_code) = @_;
	if ($job->{on_failure}) {
		$job->{on_failure}->(@exit_code);
	} elsif ($job->{errors}) {
		$job->{errors}->("FAILED: $job->{desc}", @exit_code);
	} else {
		print STDERR "JOB $job->{desc} FAILED\nexit @exit_code\n";
	}
}


1;

__END__