File Coverage

blib/lib/Helios/TS.pm

Criterion	Covered	Total	%
statement	27	113	23.8
branch	0	44	0.0
condition	0	25	0.0
subroutine	9	16	56.2
pod	3	6	50.0
total	39	204	19.1

package Helios::TS;

use 5.008;
use strict;
use warnings;
use base qw(TheSchwartz);
use fields qw(active_worker_class);        # new fields for this subclass
use Carp qw( croak );
use List::Util qw( shuffle );
use Helios::TS::Job;

use constant OK_ERRORS => { map { $_ => 1 } Data::ObjectDriver::Errors->UNIQUE_CONSTRAINT, };

our $VERSION = '2.80';

# FILE CHANGE HISTORY
# (This code is modified from the original TheSchwartz.pm where noted.)
# [LH] [2012-07-11]: driver_for(): Changed driver creation to use Helios driver
# to cache database connections.
# [LH] [2013-09-21]: find_job_for_workers(): Added code to enable job
# prioritization.
# [LH] [2013-10-04]: Implemented "virtual jobtypes" - funcmap entries without
# actual TheSchwartz::Worker subclasses to back them up.  Switched to using
# Helios::TS::Job instead of base TheSchwartz::Job because of this.
# [LH] [2013-10-04]: work_once(): Commented out call to
# temporarily_remove_ability() because we do not think the issue it solves is
# a concern for Helios::TS (Oracle's indexes do not exhibit the issue t_r_a()
# is supposed to solve, and we're not sure MySQL indexes do anymore either).
# [LH] [2013-10-04]: Fix for Helios bug [RT79690], which appears to be a DBD
# problem where a LOB becomes unbound in a query.
# [LH] [2013-11-24]: Removed old code already commented out.

our $T_AFTER_GRAB_SELECT_BEFORE_UPDATE;
our $T_LOST_RACE;
our $FIND_JOB_BATCH_SIZE = 50;

# BEGIN CODE COPYRIGHT (C) 2013 LOGICAL HELION, LLC.
# [LH] [2013-10-04]: Virtual jobtypes: Helios::TS->{active_worker_class}
# attribute and accessors for it.
sub new {
	my $class = shift;
	my %params = @_;
	my $self = fields::new($class);
	$self->SUPER::new(@_);                # init base fields
	if ( defined($params{active_worker_class})) {
		$self->{active_worker_class} = $params{active_worker_class};
	}
    return $self;
}

sub active_worker_class {
	my Helios::TS $hts = shift;
	return $hts->{active_worker_class};
}
sub set_active_worker_class {
	my Helios::TS $hts = shift;
	$hts->{active_worker_class} = shift;
}
# END CODE COPYRIGHT (C) 2013 LOGICAL HELION, LLC.


sub driver_for {
    my Helios::TS $client = shift;
    my($hashdsn) = @_;
    my $driver;
    my $t = time;
    my $cache_duration = $client->{driver_cache_expiration};
    if ($cache_duration && $client->{cached_drivers}{$hashdsn}{create_ts} && $client->{cached_drivers}{$hashdsn}{create_ts} + $cache_duration > $t) {
        $driver = $client->{cached_drivers}{$hashdsn}{driver};
    } else {
        my $db = $client->{databases}{$hashdsn}
            or croak "Ouch, I don't know about a database whose hash is $hashdsn";
        if ($db->{driver}) {
            $driver = $db->{driver};
        } else {
			# [LH] 2012-07-11: Changed driver creation to use Helios driver to
			# cache database connections.
            $driver = Helios::ObjectDriver::DBI->new(
                        dsn      => $db->{dsn},
                        username => $db->{user},
                        password => $db->{pass},
                      );
        }
        $driver->prefix($db->{prefix}) if exists $db->{prefix};

        if ($cache_duration) {
            $client->{cached_drivers}{$hashdsn}{driver} = $driver;
            $client->{cached_drivers}{$hashdsn}{create_ts} = $t;
        }
    }
    return $driver;
}


sub find_job_for_workers {
    my Helios::TS $client = shift;
    my($worker_classes) = @_;
    $worker_classes ||= $client->{current_abilities};
    for my $hashdsn ($client->shuffled_databases) {
        ## If the database is dead, skip it.
        next if $client->is_database_dead($hashdsn);

        my $driver = $client->driver_for($hashdsn);
        my $unixtime = $driver->dbd->sql_for_unixtime;

        my @jobs;
        eval {
            ## Search for jobs in this database where:
            ## 1. funcname is in the list of abilities this $client supports;
            ## 2. the job is scheduled to be run (run_after is in the past);
            ## 3. no one else is working on the job (grabbed_until is in
            ##    in the past).
            my @ids = map { $client->funcname_to_id($driver, $hashdsn, $_) }
                      @$worker_classes;

# BEGIN CODE Copyright (C) 2012-3 by Logical Helion, LLC.
			# [LH] [2013-09-21]: Added code to enable job prioritization.
			my $direction = 'descend';
			if ( $client->prioritize eq 'low' ) {
				$direction = 'ascend';
			}
# END CODE Copyright (C) 2012-3 by Logical Helion, LLC.

# [LH] [2013-10-04]: Implemented "virtual jobtypes" - funcmap entries without
# actual TheSchwartz::Worker subclasses to back them up.  Switched to using
# Helios::TS::Job instead of base TheSchwartz::Job because of this.
            @jobs = $driver->search('Helios::TS::Job' => {
                    funcid        => \@ids,
                    run_after     => \ "<= $unixtime",
                    grabbed_until => \ "<= $unixtime",
                }, { limit => $FIND_JOB_BATCH_SIZE,
                    ( $client->prioritize ? ( sort => 'priority',
                    direction => $direction ) : () )
                }
            );
        };
        if ($@) {
            unless (OK_ERRORS->{ $driver->last_error || 0 }) {
                $client->mark_database_as_dead($hashdsn);
            }
        }

        # for test harness race condition testing
        $T_AFTER_GRAB_SELECT_BEFORE_UPDATE->() if $T_AFTER_GRAB_SELECT_BEFORE_UPDATE;

        my $job = $client->_grab_a_job($hashdsn, @jobs);
        return $job if $job;
    }
}


sub _grab_a_job {
    my Helios::TS $client = shift;
    my $hashdsn = shift;
    my $driver = $client->driver_for($hashdsn);

    ## Got some jobs! Randomize them to avoid contention between workers.
    my @jobs = shuffle(@_);

  JOB:
    while (my $job = shift @jobs) {
# BEGIN CODE COPYRIGHT (C) 2013 LOGICAL HELION, LLC.
		# [LH] [2013-10-04] [RT79690] Check the job to see that it has an arg()
		# value. If it doesn't, throw it away and get a new one.  This won't
		# prevent the LOB from unbinding, but it will work around it in a
		# relatively transparent way.
		unless ( ref($job->arg()) ) {
			next;
		}
# END CODE COPYRIGHT (C) 2013 LOGICAL HELION, LLC.
        ## Convert the funcid to a funcname, based on this database's map.
        $job->funcname( $client->funcid_to_name($driver, $hashdsn, $job->funcid) );

        ## Update the job's grabbed_until column so that
        ## no one else takes it.
#        my $worker_class = $job->funcname;
# BEGIN CODE COPYRIGHT (C) 2013 LOGICAL HELION, LLC.
		# [LH] [2013-10-04] The worker class is the "Active Worker Class" if
		# it's set.  Otherwise, assume it's just the job's jobtype (funcname).
		my $worker_class = $client->{active_worker_class} || $job->funcname;
# END CODE COPYRIGHT (C) 2013 LOGICAL HELION, LLC.
        my $old_grabbed_until = $job->grabbed_until;

        my $server_time = $client->get_server_time($driver)
            or die "expected a server time";

        $job->grabbed_until($server_time + ($worker_class->grab_for || 1));

        ## Update the job in the database, and end the transaction.
        if ($driver->update($job, { grabbed_until => $old_grabbed_until }) < 1) {
            ## We lost the race to get this particular job--another worker must
            ## have got it and already updated it. Move on to the next job.
            $T_LOST_RACE->() if $T_LOST_RACE;
            next JOB;
        }

        ## Now prepare the job, and return it.
        my $handle = TheSchwartz::JobHandle->new({
            dsn_hashed => $hashdsn,
            jobid      => $job->jobid,
        });
        $handle->client($client);
        $job->handle($handle);
        return $job;
    }

    return undef;
}


sub work_once {
	# [LH] [2013-10-04] Using Helios::TS not TheSchwartz.
    my Helios::TS $client = shift;
    my $job = shift;  # optional specific job to work on

    ## Look for a job with our current set of abilities. Note that the
    ## list of current abilities may not be equal to the full set of
    ## abilities, to allow for even distribution between jobs.
    $job ||= $client->find_job_for_workers;

    ## If we didn't find anything, restore our full abilities, and try
    ## again.
    if (!$job &&
        @{ $client->{current_abilities} } < @{ $client->{all_abilities} }) {
        $client->restore_full_abilities;
        $job = $client->find_job_for_workers;
    }

	# [LH] [2013-10-04]: Virtual Jobtypes: Use the active_worker_class
	# instead of the job's funcname if active_worker_class is set.
	my $class = $client->{active_worker_class} || ($job ? $job->funcname : undef);

    if ($job) {
        my $priority = $job->priority ? ", priority " . $job->priority : "";
# BEGIN CODE COPYRIGHT (C) 2013 LOGICAL HELION, LLC.
		if ($client->{active_worker_class}) {
			$job->{active_worker_class} = $client->{active_worker_class};
		}
# END CODE COPYRIGHT (C) 2013 LOGICAL HELION, LLC.
        $job->debug("TheSchwartz::work_once got job of class '$class'$priority");
    } else {
        $client->debug("TheSchwartz::work_once found no jobs");
    }

    ## If we still don't have anything, return.
    return unless $job;

    ## Now that we found a job for this particular funcname, remove it
    ## from our list of current abilities. So the next time we look for a
    ## we'll find a job for a different funcname. This prevents starvation of
    ## high funcid values because of the way MySQL's indexes work.
	# [LH] [2013-10-04]: work_once(): Commented out call to
	# temporarily_remove_ability() because we do not think the issue it solves is
	# a concern for Helios::TS (Oracle's indexes do not exhibit the issue t_r_a()
	# is supposed to solve, and we're not sure MySQL indexes do anymore either).
#    $client->temporarily_remove_ability($class);

    $class->work_safely($job);

    ## We got a job, so return 1 so work_until_done (which calls this method)
    ## knows to keep looking for jobs.
    return 1;
}


1;
__END__