Job Submission to LSF or SGE

#!/usr/bin/perl
# farm/bsub -- a MOCK of LSF's bsub, for this module's examples only.
# It accepts -q QUEUE, -J NAME, -o FILE; the rest of the words are the job's
# command. It prints the line LSF documents for a successful submission,
#   Job <ID> is submitted to queue <QUEUE>.
# and runs the job in the background on this machine. It is NOT LSF.
# If MOCK_SITE_BANNER is set, it first prints that text, the way many sites
# wrap bsub in a script that prints its own notice.
use strict;
use warnings;
use FindBin qw($Bin);
use POSIX qw(setsid);

my %QUEUES = map { $_ => 1 } qw(normal long);
my ($queue, $name, $out) = ("normal", "", "/dev/null");
while (@ARGV && $ARGV[0] =~ /^-/) {
    my $o = shift;
    if    ($o eq "-q") { $queue = shift }
    elsif ($o eq "-J") { $name  = shift }
    elsif ($o eq "-o") { $out   = shift }
    else  { print STDERR "mock bsub: unknown option $o\n"; exit 255 }
}
@ARGV or do { print STDERR "mock bsub: no command given\n"; exit 255 };
$QUEUES{$queue} or do {
    print STDERR "mock bsub: queue '$queue' does not exist; job not submitted\n";
    exit 255;
};

my $state = "$Bin/state";
mkdir $state;
open my $lk, "+>>", "$state/next_id" or die "mock bsub: $!\n";
flock $lk, 2; seek $lk, 0, 0;
my $id = <$lk> // 4100; chomp $id; $id++;
truncate $lk, 0; print {$lk} "$id\n"; close $lk;

my $cmd = join " ", @ARGV;                    # the mock keeps the command as ONE string
write_state($id, "PEND");
open my $meta, ">", "$state/$id.job" or die;
print {$meta} "name=$name\nqueue=$queue\ncmd=$cmd\n";
close $meta;

my $pid = fork // die "mock bsub: fork: $!\n";
if ($pid == 0) {                               # the "execution host"
    setsid();
    open STDIN, "<", "/dev/null";
    open STDOUT, ">", $out or exit 1;
    open STDERR, ">&", \*STDOUT;
    select(undef, undef, undef, 0.2);          # dispatch delay
    write_state($id, "RUN");
    system("/bin/sh", "-c", $cmd);             # run through a shell, as a job script would
    write_state($id, $? == 0 ? "DONE" : "EXIT", $? >> 8);
    exit 0;
}
print "$ENV{MOCK_SITE_BANNER}\n" if $ENV{MOCK_SITE_BANNER};
print "Job <$id> is submitted to queue <$queue>.\n";
exit 0;

sub write_state {
    my ($j, $s, $code) = @_;
    open my $f, ">", "$state/$j.stat.tmp" or die;
    print {$f} defined $code ? "$s $code\n" : "$s\n";
    close $f;
    rename "$state/$j.stat.tmp", "$state/$j.stat";
}

#!/usr/bin/perl
# farm/bjobs -- a MOCK of LSF's bjobs: one row per job id, or every job whose
# name matches -J NAME ('*' matches anything). Prints JOBID, STAT and JOB_NAME only (real bjobs
# prints more columns). It is NOT LSF.
use strict;
use warnings;
use FindBin qw($Bin);

my $state = "$Bin/state";
my $want_name;
if (@ARGV >= 2 && $ARGV[0] eq "-J") { shift; $want_name = shift }
my @ids = @ARGV;
if (defined $want_name) {
    @ids = ();
    for my $f (sort glob "$state/*.job") {
        my ($id) = $f =~ m{/(\d+)\.job$};
        my %m = map { /^(\w+)=(.*)$/ } do { open my $h, "<", $f or die; <$h> };
        my $re = join ".*", map { quotemeta } split /\*/, $want_name, -1;
        push @ids, $id if $m{name} =~ /^$re$/;
    }
    @ids or do { print STDERR "mock bjobs: no job matches $want_name\n"; exit 255 };
}
print "JOBID   STAT  JOB_NAME\n";
for my $id (@ids) {
    open my $s, "<", "$state/$id.stat" or do { print STDERR "mock bjobs: no job $id\n"; next };
    my ($stat) = split ' ', scalar <$s>;
    my %m = map { /^(\w+)=(.*)$/ } do { open my $h, "<", "$state/$id.job" or die; <$h> };
    printf "%-7s %-5s %s\n", $id, $stat, $m{name};
}

#!/usr/bin/perl
# farm/qsub -- a MOCK of Grid Engine's qsub, for this module's examples only.
# Accepts -q QUEUE, -N NAME, -o FILE, -b y, -terse; the rest is the command. Prints the
# line Grid Engine documents,  Your job ID ("NAME") has been submitted
# or, with -terse, only the job id. Jobs run through the mock bsub's backend.
# It is NOT Grid Engine.
use strict;
use warnings;
use FindBin qw($Bin);

my ($queue, $name, $out, $terse) = ("normal", "", "/dev/null", 0);
while (@ARGV && $ARGV[0] =~ /^-/) {
    my $o = shift;
    if    ($o eq "-q")     { $queue = shift }
    elsif ($o eq "-N")     { $name = shift }
    elsif ($o eq "-o")     { $out  = shift }
    elsif ($o eq "-b")     { shift }               # -b y: the command is a binary
    elsif ($o eq "-terse") { $terse = 1 }
    else  { print STDERR "mock qsub: unknown option $o\n"; exit 1 }
}
my %QUEUES = map { $_ => 1 } qw(normal long);
$QUEUES{$queue} or do {
    print STDERR "mock qsub: queue '$queue' does not exist; job not submitted\n";
    exit 1;
};
open my $p, "-|", "$Bin/bsub", "-q", $queue, "-J", $name, "-o", $out, @ARGV or exit 1;
my $said = do { local $/; <$p> };
close $p;
my ($id) = $said =~ /^Job <(\d+)>/ or exit 1;
print $terse ? "$id\n" : "Your job $id (\"$name\") has been submitted\n";

#!/usr/bin/perl
# naive_submit.pl -- farm submission as it is usually first written.
use strict;
use warnings;

open STDERR, ">&", \*STDOUT or die;            # show warnings where they happen
$| = 1;
system("bin/build") == 0 or die "build failed\n";

my @runs = (["smoke_s1",  "normal", "+test=smoke +seed=1"],
            ["rand_s2",   "lng",    "+test=rand +seed=2"],      # a typo'd queue
            ["corner_s1", "normal", "+test=corner +seed=1"],
            ["rand_s3",   "lnog",   "+test=rand +seed=3"]);     # another typo
my %job;
for my $r (@runs) {
    my ($run, $queue, $args) = @$r;
    mkdir "runs"; mkdir "runs/$run";
    my $out = `farm/bsub -q $queue -J $run -o runs/$run/sim.log "bin/simv $args"`;
    my ($jid) = $out =~ /Job <(\d+)>/;
    $job{$jid} = "runs/$run";
}

print "\n%job holds ", scalar(keys %job), " entries for ", scalar(@runs), " runs:\n";
printf "  %-6s => %s\n", ($_ eq "" ? '""' : $_), $job{$_} for sort keys %job;

sleep 1 while `farm/bjobs @{[ grep { length } keys %job ]} 2>/dev/null` =~ /\b(?:PEND|RUN)\b/;

mock bsub: queue 'lng' does not exist; job not submitted
Use of uninitialized value $jid in hash element at naive_submit.pl line 20.
mock bsub: queue 'lnog' does not exist; job not submitted
Use of uninitialized value $jid in hash element at naive_submit.pl line 20.

%job holds 3 entries for 4 runs:
  ""     => runs/rand_s3
  4101   => runs/smoke_s1
  4102   => runs/corner_s1

#!/usr/bin/perl
# banner_parse.pl -- two ways to find the job id when the site's bsub wrapper
# prints a notice first.
use strict;
use warnings;

$ENV{MOCK_SITE_BANNER} = "site policy: jobs longer than 4h must use queue long";
my $out = `farm/bsub -J demo -o /dev/null bin/simv +test=smoke`;
print "bsub printed:\n", map { "  $_\n" } split /\n/, $out;

my ($first_number) = $out =~ /(\d+)/;
my ($anchored)     = $out =~ /^Job <(\d+)> is submitted to queue <\S+>\.$/m;
print "first number in the output:  $first_number\n";
print "the documented line's id:    $anchored\n";
sleep 1 while `farm/bjobs $anchored` =~ /\b(?:PEND|RUN)\b/;

bsub printed:
  site policy: jobs longer than 4h must use queue long
  Job <4101> is submitted to queue <normal>.
first number in the output:  4
the documented line's id:    4101

#!/bin/sh
# what the farm runs: the command, its log, its status, then 'done'
cd "$(dirname "$0")" || exit 2
./cmd.sh > sim.log 2>&1
echo "exit=$?" > status
: > done

#!/usr/bin/perl
# crash_demo.pl -- a runner dies (signal 9) after submitting three jobs.
# Does its job map survive? Can the jobs be found again?
use strict;
use warnings;
use IO::Handle;

system("bin/build") == 0 or die "build failed\n";

for my $variant ("buffered", "autoflush") {
    my $regr = "r_$variant";
    my $pid = fork // die "fork: $!\n";
    if ($pid == 0) {                                   # the runner
        open my $map, ">", "$regr.jobs.tsv" or die;
        $map->autoflush(1) if $variant eq "autoflush";
        for my $n (1 .. 3) {
            my $out = `farm/bsub -J $regr.smoke_s$n -o /dev/null bin/simv +test=smoke +seed=$n`;
            my ($jid) = $out =~ /^Job <(\d+)> is submitted/m or die;
            print {$map} "$jid\tsmoke_s$n\n";
        }
        kill "KILL", $$;                               # the runner dies here
    }
    waitpid($pid, 0);
    my $sig = $? & 127;                                # copy it before the next command
    open my $m, "<", "$regr.jobs.tsv" or die;
    my @lines = <$m>;
    my @found = grep { /^\d/ } `farm/bjobs -J '$regr.*'`;
    printf "%-9s runner died by signal %d; its map has %d line(s); the farm has %d job(s) named %s.*\n",
           $variant, $sig, scalar @lines, scalar @found, $regr;
}

sleep 1 while `farm/bjobs -J 'r_*'` =~ /\b(?:PEND|RUN)\b/;   # let the jobs finish
print "\nwhat r_buffered.jobs.tsv kept:\n";
open my $kept, "<", "r_buffered.jobs.tsv" or die;
print "  $_" while <$kept>;
print "\nrecovering the full map from the job names:\n";
for (`farm/bjobs -J 'r_buffered.*'`) {
    next unless /^(\d+)\s+(\S+)\s+r_buffered\.(\S+)$/;
    print "  job $1 -> run $3   ($2)\n";
}

buffered  runner died by signal 9; its map has 2 line(s); the farm has 3 job(s) named r_buffered.*
autoflush runner died by signal 9; its map has 3 line(s); the farm has 3 job(s) named r_autoflush.*

what r_buffered.jobs.tsv kept:
  4101	smoke_s1
  4102	smoke_s2

recovering the full map from the job names:
  job 4101 -> run smoke_s1   (DONE)
  job 4102 -> run smoke_s2   (DONE)
  job 4103 -> run smoke_s3   (DONE)

#!/usr/bin/perl
# submit.pl -- submit planned runs to a farm and keep the job-id -> run map.
# Usage: perl submit.pl [lsf|sge]
use strict;
use warnings;
use IO::Handle;
use File::Path qw(make_path remove_tree);

my $REGR = "r20261013";                    # this regression's id, used in job names

# Everything scheduler-specific lives in this table.
my %SCHED = (
    lsf => {
        argv => sub { my ($r, $name, $q) = @_;
                      ("farm/bsub", "-q", $q, "-J", $name, "-o", "$r->{dir}/farm.out",
                       "$r->{dir}/job.sh") },
        id   => qr/^Job <(\d+)> is submitted to queue <\S+>\.$/,
    },
    sge => {
        argv => sub { my ($r, $name, $q) = @_;
                      ("farm/qsub", "-terse", "-q", $q, "-N", $name, "-o", "$r->{dir}/farm.out",
                       "-b", "y", "$r->{dir}/job.sh") },
        id   => qr/^(\d+)$/,
    },
);
my $sched = shift // "lsf";
my $S = $SCHED{$sched} or die "unknown scheduler '$sched'\n";

my @plan = (
    { id => "smoke_s1",  args => ["+test=smoke",  "+seed=1"] },
    { id => "corner_s1", args => ["+test=corner", "+seed=1"] },
    { id => "rand_s2",   args => ["+test=rand",   "+seed=2"] },
    { id => "rand_s3",   args => ["+test=rand",   "+seed=3"], queue => "lng" },  # a typo
    { id => "rand_s4",   args => ["+test=rand",   "+seed=4"] },
);

# Run a command WITHOUT a shell; return its status and everything it printed,
# standard error included.
sub capture {
    my @cmd = @_;
    my $pid = open(my $p, "-|") // die "fork: $!\n";
    if ($pid == 0) {
        open STDERR, ">&", \*STDOUT or exit 126;
        { no warnings "exec"; exec { $cmd[0] } @cmd }
        print "cannot run $cmd[0]: $!\n";
        exit 127;
    }
    my @out = <$p>;
    close $p;                              # sets $? to the command's status
    return ($?, @out);
}

sub write_script {
    my ($path, @lines) = @_;
    open my $f, ">", $path or die "$path: $!\n";
    print {$f} map { "$_\n" } @lines;
    close $f or die "$path: $!\n";
    chmod 0755, $path;
}

system("bin/build") == 0 or die "build failed\n";
remove_tree("runs");
make_path("runs");
open my $map, ">", "runs/jobs.tsv" or die "runs/jobs.tsv: $!\n";
$map->autoflush(1);                        # each line reaches the disk as written
print {$map} "job_id\trun_id\tjob_name\n";

my $failed = 0;
for my $r (@plan) {
    $r->{dir} = "runs/$r->{id}";
    make_path($r->{dir});
    write_script("$r->{dir}/cmd.sh", "#!/bin/sh",
        "cd \"\$(dirname \"\$0\")/../..\" || exit 2",
        "exec bin/simv @{ $r->{args} }");
    write_script("$r->{dir}/job.sh", "#!/bin/sh",
        "# what the farm runs: the command, its log, its status, then 'done'",
        "cd \"\$(dirname \"\$0\")\" || exit 2",
        "./cmd.sh > sim.log 2>&1",
        "echo \"exit=\$?\" > status",
        ": > done");

    my $name = "$REGR.$r->{id}";
    my ($st, @said) = capture($S->{argv}->($r, $name, $r->{queue} // "normal"));
    my ($jid) = map { /$S->{id}/ ? $1 : () } @said;

    if ($st != 0 || !defined $jid) {
        my $why = join " | ", map { s/\s+\z//r } @said;
        printf "%-10s NOT SUBMITTED (status %d): %s\n", $r->{id}, $st >> 8, $why;
        $failed++;
        next;
    }
    print {$map} "$jid\t$r->{id}\t$name\n";
    printf "%-10s job %s\n", $r->{id}, $jid;
}
close $map or die "runs/jobs.tsv: $!\n";
printf "%d submitted, %d not submitted; map in runs/jobs.tsv\n",
       @plan - $failed, $failed;
exit($failed ? 1 : 0);

#!/usr/bin/perl
# farm_status.pl -- join the job map with the farm's view of each job.
use strict;
use warnings;

open my $m, "<", "runs/jobs.tsv" or die "runs/jobs.tsv: $!\n";
my $header = <$m>;
my %run_of = map { chomp; my ($j, $r) = split /\t/; ($j => $r) } <$m>;

my @ids = sort { $a <=> $b } keys %run_of;
sleep 1 while `farm/bjobs @ids` =~ /\b(?:PEND|RUN)\b/;    # (Chapter 25.6 bounds this)

my @rows = `farm/bjobs @ids`;
my @col  = split ' ', shift @rows;                        # find columns BY NAME
my %at   = map { $col[$_] => $_ } 0 .. $#col;
printf "%-10s %-6s %-5s %s\n", "RUN", "JOB", "STAT", "IN THE RUN DIRECTORY";
for (@rows) {
    my @f   = split ' ';
    my $jid = $f[ $at{JOBID} ];
    my $run = $run_of{$jid} // "?";
    my $dir = "runs/$run";
    my $st  = -e "$dir/status" ? do { open my $s, "<", "$dir/status"; scalar <$s> } : "no status\n";
    printf "%-10s %-6s %-5s %s%s", $run, $jid, $f[ $at{STAT} ],
           (-e "$dir/done" ? "done, " : "NOT done, "), $st;
}

#!/usr/bin/perl
# submit_demo.pl -- submit the plan to the (mock) LSF, then ask where each run is.
use strict;
use warnings;

$| = 1;
system($^X, "submit.pl", "lsf");
print "(submit.pl exit ", $? >> 8, ")\n\n";
system($^X, "farm_status.pl");

print "\n--- the same plan, submitted to the (mock) Grid Engine ---\n";
system($^X, "submit.pl", "sge");
print "(submit.pl exit ", $? >> 8, ")\n\n";
system($^X, "farm_status.pl");

smoke_s1   job 4101
corner_s1  job 4102
rand_s2    job 4103
rand_s3    NOT SUBMITTED (status 255): mock bsub: queue 'lng' does not exist; job not submitted
rand_s4    job 4104
4 submitted, 1 not submitted; map in runs/jobs.tsv
(submit.pl exit 1)

RUN        JOB    STAT  IN THE RUN DIRECTORY
smoke_s1   4101   DONE  done, exit=0
corner_s1  4102   DONE  done, exit=0
rand_s2    4103   DONE  done, exit=0
rand_s4    4104   DONE  done, exit=0

--- the same plan, submitted to the (mock) Grid Engine ---
smoke_s1   job 4105
corner_s1  job 4106
rand_s2    job 4107
rand_s3    NOT SUBMITTED (status 1): mock qsub: queue 'lng' does not exist; job not submitted
rand_s4    job 4108
4 submitted, 1 not submitted; map in runs/jobs.tsv
(submit.pl exit 1)

RUN        JOB    STAT  IN THE RUN DIRECTORY
smoke_s1   4105   DONE  done, exit=0
corner_s1  4106   DONE  done, exit=0
rand_s2    4107   DONE  done, exit=0
rand_s4    4108   DONE  done, exit=0