source: trunk/oar/oar-dispatch @ 12

#!/usr/bin/perl
#
# 2011/11/03 gabriel

use strict;

use Getopt::Long();
use Pod::Usage;
use Coro;
use Coro::Signal;
use Coro::Semaphore;

my $task = 0;
my $overload = 1.1;
my $file = '';
my $verbose;
my $help;

Getopt::Long::GetOptions(
        'task=i'                => \$task,
        'overload=f'    => \$overload,
        'file=s'                => \$file,
        'verbose'       => \$verbose,
        'help'          => \$help,
        ) || pod2usage(-verbose => 0);
pod2usage(-verbose => 2) if $help;

if ($task == 0) {
        open(NODE_FILE, '<', "$ENV{OAR_NODE_FILE}") or die "can't open ENV{OAR_NODE_FILE}: $!";
        $task++ while <NODE_FILE>;
        close NODE_FILE;
}

my @job = ();
open (JOB_LIST, '<', "$file") or die "can't open $file: $!";
while (<JOB_LIST>) {
        chomp;
        next if m/^#/;
        push @job, $_ if m/^\s*oarsub/;
}
close JOB_LIST;

my $container_id=$ENV{OAR_JOB_ID};
my $insert_oar_option = "-t inner=$container_id";

# interactive job
if (not $container_id > 1) {
        $insert_oar_option = '';
        $overload = 1;
}


my $finished = new Coro::Signal;
my $job_active = new Coro::Semaphore 0;
my $job_todo = new Coro::Semaphore 0;
$job_todo->up for (@job);

my %scheduled = ();

async {
        for my $job (@job) {
                while ($job_active->count >= $task*$overload) {
                        cede;
                }
                $job =~ s/^\s*oarsub//;
                print "oarsub $insert_oar_option $job" if $verbose;
                my $job_id = `oarsub $insert_oar_option $job|grep ^OAR_JOB_ID|cut -f 2 -d '='`;
                chomp $job_id;
                if ($job_id > 1) {
                        $scheduled{$job_id}++;
                        $job_active->up;
                }
                cede;
        }
}

async {
        while () {
                for my $job_id (keys %scheduled) {
                        my $is_finish = `oarstat -s -j $job_id`;
                        chomp $is_finish;
                        if ($is_finish =~ m/Terminated/) {
                                delete $scheduled{$job_id};
                                $job_active->down;
                                $job_todo->down;
                        }
                        cede;
                }

                $finished->send if $job_todo->count == 0;
                cede;
        }
}

cede;
   
$finished->wait;


__END__

=head1 NAME

oar-dispatch - dispatch lot of small oar job

=head1 SYNOPSIS

 oar-dispatch [--core integer] [--overload real] --file filepath [--verbose]
 oar-dispatch --help

=head1 OPTIONS

 --task number of task to do in parallel.
                        default to line number of file OAR_NODE_FILE.
 
 --overload     number of OAR job to create / number of task.
                        Some job are create in advance to start whenever it's possible.
                        1.1 by default.

 --file file name which content OAR job list

 --verbose
 
 --help

Input job file name content can have

 - empty line
 - comment line begin with #
 - oarsub command without -t option
 
C<oar-dispatch> will add C<-t inner=container_id> in this command line,
just after C<oarsub>.

Example where F<$HOME/test/subjob.txt> is a list of OAR script job (and can be executable but not need here).

 oarsub -n test -l /core=1,walltime=00:05:00 $HOME/test/subjob1.oar
 oarsub -n test -l /core=1,walltime=00:05:00 $HOME/test/subjob2.oar
 oarsub -n test -l /core=1,walltime=00:05:00 $HOME/test/subjob3.oar
 oarsub -n test -l /core=1,walltime=00:05:00 $HOME/test/subjob4.oar
 ...
 oarsub -n test -l /core=1,walltime=00:05:00 $HOME/test/subjob38.oar
 oarsub -n test -l /core=1,walltime=00:05:00 $HOME/test/subjob39.oar
 oarsub -n test -l /core=1,walltime=00:05:00 $HOME/test/subjob40.oar

These jobs could be launch with

 oarsub -t container -n test-container -l /core=6,walltime=00:35:00 "oar-dispatch -f ./subjob.list.txt"

Total C<walltime> is defined by the formula:

 total_walltime = subjob_walltime * total_subjob / core + global_delay

In practise, C<oar-dispatch> take few second and each subjob run in less than it's walltime so

 total_walltime < subjob_walltime * total_subjob / core

If launch in interactif, C<overload> parameter is equal to 1,
C<task> must be define
and no inner container is add to the C<oarsub> command line.


=head1 AUTHORS

Gabriel Moreau (C) 2011