lib/BEX.pm

   1 # Batch EXecutor 2.0
   2 # (c) 2011 Martin Mares <mj@ucw.cz>
   3
   4 use strict;
   5 use warnings;
   6
   7 package BEX::Config;
   8
   9 our %machines = (
  10         'albireo' => { },
  11         'localhost' => { },
  12 );
  13
  14 package BEX::Job;
  15
  16 use POSIX;
  17
  18 our $job_cnt = 0;
  19
  20 sub new($;$) {
  21         my ($class, $id) = @_;
  22         my $job = { };
  23         bless $job;
  24         if (defined $id) {
  25                 $job->{'ID'} = $id;
  26         } else {
  27                 $job_cnt++;
  28                 $job->{'ID'} = POSIX::strftime("%Y%m%d-%H%M%S-$$-$job_cnt", localtime);
  29         }
  30         $job->{'Subject'} = '(no subject)';
  31         return $job;
  32 }
  33
  34 sub new_from_file($$;$) {
  35         my ($class, $file, $header_only) = @_;
  36         my $job = { };
  37         open T, '<', $file or die "Cannot open $file: $!";
  38         while (<T>) {
  39                 chomp;
  40                 /^$/ and last;
  41                 /^([A-Z][A-Za-z0-9-]*):\s*(.*)/ or die "Cannot load $file: Header syntax error";
  42                 !defined $job->{$1} or die "Cannot load $file: Header $1 re-defined";
  43                 $job->{$1} = $2;
  44         }
  45         if (!$header_only) {
  46                 my @cmds = <T>;
  47                 $job->{'body'} = join("", @cmds);
  48         }
  49         close T;
  50         $job->{'Subject'} //= '?';
  51         $job->{'ID'} or die "Cannot load $file: Missing ID";
  52         return bless $job;
  53 }
  54
  55 sub attr($$;$) {
  56         my ($job, $attr, $val) = @_;
  57         $job->{$attr} = $val if defined $val;
  58         return $job->{$attr};
  59 }
  60
  61 sub dump($) {
  62         my ($job) = @_;
  63         for my $k (sort keys %$job) {
  64                 print "$k: ", $job->{$k}, "\n";
  65         }
  66 }
  67
  68 sub save($;$) {
  69         my ($job, $fn) = @_;
  70         -d "tmp" or mkdir "tmp" or die "Cannot create directory tmp: $!";
  71         $fn //= 'tmp/t-' . $job->{'ID'};
  72         open T, '>', $fn or die "Cannot create $fn: $!";
  73         for my $k (sort grep { /^[A-Z]/ } keys %$job) {
  74                 print T "$k: ", $job->{$k}, "\n";
  75         }
  76         print T "\n";
  77         print T $job->{'body'} if defined $job->{'body'};
  78         close T;
  79         return $fn;
  80 }
  81
  82 package BEX::Queue;
  83
  84 sub new($;$) {
  85         my ($class, $name) = @_;
  86         $name //= 'queue';
  87         -d $name or die "Queue directory $name does not exist\n";
  88         for my $d ("hosts", "jobs") {
  89                 -d "$name/$d" or mkdir "$name/$d" or die "Cannot create directory $name/$d: $!";
  90         }
  91         my $queue = {
  92                 'Name' => $name,
  93                 'MetaCache' => {},
  94         };
  95         return bless $queue;
  96 }
  97
  98 sub host_dir($$) {
  99         my ($queue, $machine) = @_;
 100         return $queue->{'Name'} . '/hosts/' . $machine;
 101 }
 102
 103 sub queue_file($$) {
 104         my ($queue, $machine, $jid) = @_;
 105         return $queue->host_dir($machine) . '/q-' . $jid;
 106 }
 107
 108 sub status_file($$) {
 109         my ($queue, $machine, $jid) = @_;
 110         return $queue->host_dir($machine) . '/s-' . $jid;
 111 }
 112
 113 sub job_file($$) {
 114         my ($queue, $jid) = @_;
 115         return $queue->{'Name'} . '/jobs/j-' . $jid;
 116 }
 117
 118 sub enqueue($$$) {
 119         my ($queue, $machine, $job) = @_;
 120         my $qf = $queue->queue_file($machine, $job->{'ID'});
 121         if (-f $qf) { return 0; }
 122         my $fn = $queue->job_file($job->{'ID'});
 123         -f $fn or $job->save($fn);
 124         my $dir = $queue->host_dir($machine);
 125         -d $dir or mkdir $dir or die "Cannot create directory $dir: $!";
 126         symlink "../../jobs/j-" . $job->{'ID'}, $qf or die "Cannot create $qf: $!";
 127         return 1;
 128 }
 129
 130 sub scan($$) {
 131         my ($queue, $machine) = @_;
 132         my @list = ();
 133         if (opendir D, $queue->host_dir($machine)) {
 134                 while ($_ = readdir D) {
 135                         s/^q-// or next;
 136                         push @list, $_;
 137                 }
 138                 closedir D;
 139         }
 140         return @list;
 141 }
 142
 143 sub job_metadata($$) {
 144         my ($queue, $jid) = @_;
 145         my $cache = $queue->{'MetaCache'};
 146         if (!defined $cache->{$jid}) {
 147                 $cache->{$jid} = BEX::Job->new_from_file($queue->job_file($jid), 1);
 148         }
 149         return $cache->{$jid};
 150 }
 151
 152 sub read_job_status($$$) {
 153         my ($queue, $machine, $jid) = @_;
 154         my %s = ();
 155         my $sf = status_file($queue, $machine, $jid);
 156         if (open S, '<', $sf) {
 157                 while (<S>) {
 158                         chomp;
 159                         /^(\w+):\s*(.*)/ or die "Parse error in $sf";
 160                         $s{$1} = $2;
 161                 }
 162                 close S;
 163         }
 164         return \%s;
 165 }
 166
 167 sub write_job_status($$$$) {
 168         my ($queue, $machine, $jid, $stat) = @_;
 169         my $sf = status_file($queue, $machine, $jid);
 170         open S, '>', $sf or die "Cannot create $sf: $!";
 171         for my $k (sort keys %$stat) {
 172                 print S "$k: ", $stat{$k}, "\n";
 173         }
 174         close S;
 175 }
 176
 177 42;