File Coverage

blib/lib/Proc/JobQueue.pm

Criterion	Covered	Total	%
statement	195	247	78.9
branch	76	132	57.5
condition	11	25	44.0
subroutine	26	30	86.6
pod	13	21	61.9
total	321	455	70.5

package Proc::JobQueue;

use strict;
use warnings;

use Time::HiRes qw(sleep);
use Sys::Hostname;
use Carp qw(confess);
use Hash::Util qw(lock_keys unlock_keys);
use Time::HiRes qw(time);
use Module::Load;
use Object::Dependency;
require Exporter;

our $VERSION = 0.903;
our $debug ||= 0;
our $status_frequency ||= 2;
our $host_canonicalizer ||= 'File::Slurp::Remote::CanonicalHostnames';

our @ISA = qw(Exporter);
our @EXPORT_OK = qw(is_remote_host canonicalize my_hostname);

sub configure
{
	my ($queue, %params) = shift;
	@$queue{keys %params} = values %params;
}

sub addhost
{
	my ($queue, $host, %params) = @_;
	my $hr;
	if ($hr = $queue->{status}{$host}) {
		@$hr{keys %params} = values %params;
	} else {
		$hr = $queue->{status}{$host} = {
			name		=> $host,
			jobs_per_host	=> $queue->{jobs_per_host},
			in_startmore	=> 0,
			%params,
			running		=> {},
			queue		=> {},
		};
	}
	$queue->set_readiness($host);
}

sub set_readiness
{
	my ($queue, $host) = @_;
	my $hr = $queue->{status}{$host};
	if ($hr->{jobs_per_host} and $hr->{jobs_per_host} > keys %{$hr->{running}}) {
		$queue->{ready_hosts}{$host} = $hr;
	} elsif (! keys %{$hr->{running}}) {
		$queue->{ready_hosts}{$host} = $hr;
	} else {
		delete $queue->{ready_hosts}{$host};
	}
}

sub new
{
	my ($pkg, %params) = @_;
	my $queue = bless {
		dependency_graph	=> undef,
		startmore_in_progress	=> undef,
		host_overload		=> 120,
		host_is_over		=> 0,
		jobnum			=> 1000,
		jobs_per_host		=> 4,
		queue			=> {},
		status			=> {},
		ready_hosts		=> {},
		hold_all		=> 0,
		hosts			=> [ my_hostname() ],
		%params,
	}, $pkg;
	$queue->addhost($_) for @{$queue->{hosts}};
	lock_keys(%$queue);
	return $queue;
}

sub hold
{
	my ($self, $new) = @_;
	$self->{hold_all} = $new if defined $new;
	return $self->{hold_all};
}

sub add
{
	my ($queue, $job, $host) = @_;
	confess "$job not a ref" unless ref $job;
	confess "$job is not a job" unless $job->isa('Proc::JobQueue::Job');

	$job->jobnum($queue->{jobnum}++)
		unless $job->jobnum;
	my $jobnum = $job->jobnum();

	print STDERR "Adding $jobnum - ".ref($job)." to worklist\n" if $debug > 2;
	my $q;
	if ($host) {
		confess "no $host" unless $queue->{status}{$host};
		$q = $queue->{status}{$host}{queue};
	} else {
		$q = $queue->{queue};
	}

	$q->{$jobnum} = $job;

	$job->{dependency_graph} = $queue->{dependency_graph}; # TODO: do this with a method

	$job->queue($queue);
	$queue->startmore;
}

# this looks at the dependency queue.  startmore_jobs looks at the
# at the jobs queue.
sub startmore
{
	my ($job_queue) = shift;

	if ($job_queue->{startmore_in_progress}) {
		print STDERR "Re-entry to startmore prevented\n" if $debug;
		$job_queue->{startmore_in_progress}++;
		return 0;
	}
	$job_queue->{startmore_in_progress} = 2;

	my $dependency_graph = $job_queue->{dependency_graph};

	my $stuff_started = 0;

	my $jq_done;

	print STDERR "looking for more depenency graph items to queue up\n" if $debug;
	eval {
		$job_queue->checkjobs();

		while ($job_queue->{startmore_in_progress} > 1) {
			$job_queue->{startmore_in_progress} = 1;
			if ($dependency_graph) {
				while (my @runnable = $dependency_graph->independent(lock => 1)) {
					$stuff_started++;
					for my $task (@runnable) {
						print "Queuing $task->{desc}\n" if $debug;
						if ($task->can('run_dependency_task')) {
							$job_queue->{startmore_in_progress}++ if $task->run_dependency_task($dependency_graph);
						} elsif ($task->isa('Proc::JobQueue::Job')) {
							$job_queue->add($task, $task->{force_host});
						} else {
							die "don't know how to handle $task";
						}
					}
				}
			}

			$jq_done = $job_queue->startmore_jobs();

			redo if $job_queue->{startmore_in_progress} > 1;
		}
	};
	if ($@) {
		$job_queue->suicide();
	};

	$job_queue->{startmore_in_progress} = 0;

	return $jq_done unless $dependency_graph;

	if ($jq_done && $dependency_graph->alldone) {
		print STDERR "Nothing more to do\n";
		$job_queue->unloop();
		return 1;
	} elsif ($jq_done && ! $stuff_started) {
		if (keys %{$dependency_graph->{stuck}}) {
			print STDERR "All runnable jobs are done, remaining dependencies are stuck:\n";
			for my $o (values %{$dependency_graph->{stuck}}) {
				printf "\t%s\n", $dependency_graph->desc($o);
			}
			$job_queue->unloop();
			return 1;
		} else {
			print STDERR "Job queue is empty, but dependency graph doesn't think there is any work to be done!\n";
			$dependency_graph->dump_graph();
		}
	}
	return 0;
}

sub startmore_jobs
{
	my ($queue) = @_;
	return 0 if $queue->{hold_all};
	print "# Looking to start more\n" if $debug > 8;
	confess "no hosts added" unless keys %{$queue->{status}};
	my $stuff = 0;
	my $new_host_is_over = 0;
	while(1) {
		my $redo = 0;
		HOST:
		for my $host (keys %{$queue->{ready_hosts}}) {
			print STDERR "# checking $host to maybe start more jobs\n" if $debug > 3;
			my $hr = $queue->{ready_hosts}{$host};
			JOB:
			while ((! $hr->{jobs_per_host} && ! keys %{$hr->{running}}) || $hr->{jobs_per_host} > (keys %{$hr->{running}} || 0)) {
				print STDERR "# there is room for more on $host\n" if $debug > 4;
				$new_host_is_over++
					if keys(%{$hr->{queue}}) > $queue->{host_overload};
				my @q;
				push (@q, $hr->{queue});
				push (@q, $queue->{queue})
					if $hr->{jobs_per_host} && ! $queue->{host_is_over};
				for my $q (@q) {
					next unless keys %$q;
					$stuff = 1;
					for my $jobnum (reverse sort { $q->{$a}{priority} <=> $q->{$b}{priority} || $a <=> $b } keys %$q) {
						print STDERR "# looking to start $jobnum on $host\n" if $debug > 5;
						my $job = $q->{$jobnum};
						unless ($job->runnable) {
							print STDERR "# can't start $jobnum $job->{desc} on $host: not runnable\n" if $debug > 5;
							next;
						}
						delete $q->{$jobnum};
						$queue->startjob($host, $jobnum, $job);
						$queue->set_readiness($host);
						$redo = 1;
						next HOST;
					}
				}
				last;
			}
		}
		last unless $redo;
	}
	$queue->{host_is_over} = $new_host_is_over;
	return 0 if $stuff;
	return $queue->alldone();
}

sub suicide
{
	print STDERR "DIE DIE DIE DIE DIE (DT2): $@";
	# exit 1; hangs!
	POSIX::_exit(1);
}

# a hook for EventQueue
sub unloop { }


sub startjob
{
	my ($queue, $host, $jobnum, $job) = @_;
	print STDERR "# starting $jobnum $job->{desc} on $host\n" if $debug > 1;
	my $hr = $queue->{status}{$host};
	$hr->{running}{$jobnum} = $job;
	$job->host($host);
	$job->start();
}


# This routine is re-enterant: it may be called from something it calls.
sub checkjobs
{
	my ($queue) = @_;
	my $found = 0;
	for my $host (keys %{$queue->{status}}) {
		print STDERR "# checking jobs on $host\n" if $debug > 7;
		my $hr = $queue->{status}{$host} || die;
		for my $jobnum (keys %{$hr->{running}}) {
			my $job = $hr->{running}{$jobnum};
			if ($job) {
				print STDERR "# checking $jobnum $job->{desc} on $host\n" if $debug > 8;
				$found++
					if defined $job->checkjob($queue);
			} else {
				print STDERR "# job $jobnum is undef!\n" if $debug;
				delete $hr->{running}{$jobnum};
				$found++;
			}

		}
		$queue->set_readiness($host);
	}
	return $found;
}

sub jobdone
{
	my ($job_queue, $job, $do_startmore, @exit_code) = @_;
	if ($job->{dependency_graph}) {
		if ($exit_code[0]) {
			print STDERR "Things dependent on $job->{desc} will never run: @exit_code\n";
			$job->{dependency_graph}->stuck_dependency($job, "exit @exit_code");
		} else {
			$job->{dependency_graph}->remove_dependency($job);
		}
		$job->{dependency_graph} = undef;
		# unlock_keys(%$job);
		# $job->{this_is_finished} = 1;
		# lock_keys(%$job);
	}
	$job_queue->job_part_finished($job, $do_startmore, @exit_code);
}

sub job_part_finished
{
	my ($job_queue, $job, $do_startmore, @exit_code) = @_;
	$do_startmore = 1 unless defined $do_startmore;

	my $host = $job->host;
	my $jobnum = $job->jobnum;

	print STDERR "# job $jobnum $job->{desc} on $host is done\n" if $debug > 5;

	my $hr = $job_queue->{status}{$host} or confess;
	delete $hr->{running}{$jobnum} or confess;

	$job_queue->set_readiness($host);

	$job_queue->startmore() if $do_startmore;
}

sub alldone
{
	my ($queue, $skip_status) = @_;
	$queue->status() if $debug && ! $skip_status;
	return 0 if keys %{$queue->{queue}};
	for my $host (keys %{$queue->{status}}) {
		my $hr = $queue->{status}{$host};
		return 0 unless $queue->{ready_hosts}{$host};
		return 0 if keys %{$hr->{queue}};
		return 0 if keys %{$hr->{running}};
		next unless $hr->{jobs_per_host} > 0;
	}
	return 1;
}

my $last_dump = time;

sub status
{
	my ($queue) = @_;
	return if time < $last_dump + $status_frequency;
	$last_dump = time;
	print STDERR "Queue Status\n";
	printf STDERR "\titems in main queue: %d, alldone=%d\n", scalar(keys %{$queue->{queue}}), $queue->alldone(1);
	print STDERR "\tHost overload condition is true\n" if $queue->{host_is_over};
	for my $host (sort keys %{$queue->{status}}) {
		my $hr = $queue->{status}{$host};
		printf STDERR "\titems in queue for %s: %d, items running: %s, host is %sready\n",
			$host,
			scalar(keys(%{$hr->{queue}})),
			scalar(keys(%{$hr->{running}})),
			($queue->{ready_hosts}{$host} ? "" : "not ");
		for my $job (values %{$hr->{running}}) {
			print STDERR "\t\tRunning: $job->{jobnum} $job->{desc}\n";
		}
	}
	my $dg = $queue->{dependency_graph};
	printf "Dependency Graph items: %d independent (%d locked %d active), %d total, alldone=%s\n",
		scalar(keys(%{$dg->{independent}})),
		scalar(grep { $_->{dg_lock} } values %{$dg->{independent}}),
		scalar(grep { $_->{dg_active} } values %{$dg->{independent}}),
		scalar(keys(%{$dg->{addrmap}})),
		$dg->alldone
		if $dg;
}

my $canonicalizer;
sub get_canonicalizer
{
	return $canonicalizer if $canonicalizer;
	load($host_canonicalizer);
	$canonicalizer = $host_canonicalizer->new();
}

sub canonicalize
{
	my ($host) = @_;
	return get_canonicalizer()->canonicalize($host);
}

my $my_hostname;
sub my_hostname
{
	return $my_hostname if $my_hostname;
	$my_hostname = get_canonicalizer()->myname();
}

sub is_remote_host
{
	my ($host) = @_;
	return my_hostname() ne canonicalize($host);
}

sub graph
{
	my $queue = shift;
	if (@_) {
		die "a dependency graph was already set" if $queue->{dependency_graph};
		$queue->{dependency_graph} = shift;
	} elsif (! $queue->{dependency_graph}) {
		$queue->{dependency_graph} = Object::Dependency->new();
	}
	return $queue->{dependency_graph};
}


1;

__END__