view release on metacpan or  search on metacpan

src/grid_request_worker  view on Meta::CPAN

#!/usr/bin/perl

use strict;
use Data::Dumper;
use Grid::Request::HTC;
use Grid::Request::JobFormulator;
use Log::Log4perl qw(:easy);
use Config::IniFiles;
use Sys::Hostname;
Log::Log4perl->easy_init($DEBUG);

my $logger = Log::Log4perl->get_logger();
# Deterine what host we are running on.
my $host = hostname;
$logger->info("Running on $host");

# Load which grid we are using from the configuration file.
my $config_file = Grid::Request::HTC->config();
# The [section] heading in the configuration file.
my $section = $Grid::Request::HTC::config_section;
my $config = Config::IniFiles->new(-file => $config_file);
my $grid_type = $config->val($section, "drm");

$logger->debug("Grid type: $grid_type");
my $MW_PARAM_DELIMITER = ":";

# Get the task ID that we are. There are going to be several instances of this
# script running. We need to know which one we are. This is actually different
# for every execution environment (grid), so we have a subroutine to hide that.
my $task_id = get_task_id();

check_arguments();
# the first argument is the executable to invoke
my $executable = shift(@ARGV); 
$logger->info("The executable is $executable.");
# The second argument is the blocksize, for the window of tasks to do.
my $block_size = shift(@ARGV);
$logger->info("The block size is $executable.");
# The remaining arguments are the MW parameters
my @mw_params = @ARGV;
$logger->debug("Number of MW parameters: " . scalar(@mw_params));

# Now that we have the MW parameters, let's just check that they
# make sense.
validate_mw_params(\@mw_params);
# If we are here, then the MW parameters look fine.
$logger->debug("MW parameters validated correctly.");

if (! defined $task_id || $task_id <= 0) {
    $logger->logdie("Unable to determine this worker's ID.");
}
$logger->info("Task id for this worker: $task_id");

# an array of arrays for all the parameters to invoke
my $jb = Grid::Request::JobFormulator->new();
my @invocations = $jb->formulate($block_size, $executable, @mw_params);

execute(\@invocations);

#############################################################################

sub execute {
    $logger->debug("In execute.");
    my ($arg_group_ref) = @_;
    my ($success, $failed) = (0,0);
    my $count = 0;

    my $group_size = scalar @$arg_group_ref;
    $logger->debug("The number of arguments each command invocation of will have: $group_size");

    if ( $group_size > 0 ) {
        # Length of the various argument arrays should all be the same, so we'll just use
        # the length of the first one.
        my $arg_length = scalar(@{ $arg_group_ref->[0] });
        
        # This loop is to iterate across the argument arrays
        for (my $arg_index = 0; $arg_index < $arg_length; $arg_index++) {
            $count++;

            my @exec = ();

            for (my $group_index = 0; $group_index < $group_size; $group_index++) {
                my $arg = $arg_group_ref->[$group_index]->[$arg_index];

                # TODO: This should be done in the JobFormulator formulate() method
                # Replace $(Index) with the task id.
                $arg =~ s/\$\(Index\)/$task_id/g;
                
                push(@exec, $arg);
            }

            $logger->info("Invoking: ", sub { join(" ", @exec) } );

            if ($logger->is_debug()) {
                $logger->debug("Arg list for invokation of $executable:\n", sub{ Dumper(\@exec) } );
                $logger->debug(sub { sprintf("Invoking %s. This is time #%d", $executable, $count) } );
                $logger->debug("==============================================================");
            }

            system(@exec);

            my $exit_value = $? >> 8;

            if ($logger->is_debug()) {
                $logger->debug("==============================================================");
                $logger->info(sub { sprintf("Completed run of %s. Exit value: %d", $executable, $exit_value) });
            }

            if ($exit_value == 0) {
                $success++;
            } else {
                $failed++;
            }
        }
    } else {
        $logger->warn("No arguments! Just invoking the configured executable with no args.");
    }

    $logger->info(sub { sprintf("Successful: %d. Failed: %d. Total: %d", $success, $failed, $count) });
    my $exit = ($failed == 0) ? 0 : 1;
    $logger->info("$0 has completed execution. Exiting with exit value: $exit");
    exit $exit;
}

sub check_arguments {
    if (scalar @ARGV < 3) {
        $logger->fatal("An invalid number of parameters was specified.");
        print_usage();
    }
    # Check for a valid block size
    my $block_size = $ARGV[1];
    if (int($block_size) != $block_size || $block_size <= 0) {
        $logger->logdie("A non-integer or non-positive block size was specified: $block_size.");
    }
}

sub get_task_id {
    my $task_id;
    $grid_type = lc($grid_type);

    # Delete the environment variables when done with them in case the
    # executable somehow wants to submit jobs and pass the environment.
    if ($grid_type eq "sge") {
        $task_id = $ENV{SGE_TASK_ID};
        delete $ENV{SGE_TASK_ID};