From c52365ee561bcd84c6a36d301d6959b2c68b14a7 Mon Sep 17 00:00:00 2001 From: Martin Mares Date: Sun, 30 Oct 2011 13:31:29 +0100 Subject: [PATCH 1/1] Building infrastructure... --- benq | 23 +++++++ bq | 19 ++++++ lib/BEX.pm | 177 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 219 insertions(+) create mode 100755 benq create mode 100755 bq create mode 100644 lib/BEX.pm diff --git a/benq b/benq new file mode 100755 index 0000000..0259f43 --- /dev/null +++ b/benq @@ -0,0 +1,23 @@ +#!/usr/bin/perl +# Batch EXecutor 2.0 +# (c) 2011 Martin Mares + +use strict; +use warnings; + +use lib 'lib'; +use BEX; + +my $job = BEX::Job->new; +my $fn = $job->save; +system "editor", $fn; +## FIXME: Check exit code of editor +$job = BEX::Job->new_from_file($fn); +## FIXME: Compare and exit if no changes + +my $queue = BEX::Queue->new; +for my $m (keys %BEX::Config::machines) { + $queue->enqueue($m, $job); +} + +unlink $fn; diff --git a/bq b/bq new file mode 100755 index 0000000..5a2b786 --- /dev/null +++ b/bq @@ -0,0 +1,19 @@ +#!/usr/bin/perl +# Batch EXecutor 2.0 +# (c) 2011 Martin Mares + +use strict; +use warnings; + +use lib 'lib'; +use BEX; + +my $queue = BEX::Queue->new; +for my $m (keys %BEX::Config::machines) { + print "$m:\n"; + my @q = $queue->scan($m); + for my $j (@q) { + my $job = $queue->job_metadata($j); + print "\t$j (", $job->attr('Subject'), ")\n"; + } +} diff --git a/lib/BEX.pm b/lib/BEX.pm new file mode 100644 index 0000000..cfb7fd5 --- /dev/null +++ b/lib/BEX.pm @@ -0,0 +1,177 @@ +# Batch EXecutor 2.0 +# (c) 2011 Martin Mares + +use strict; +use warnings; + +package BEX::Config; + +our %machines = ( + 'albireo' => { }, + 'localhost' => { }, +); + +package BEX::Job; + +use POSIX; + +our $job_cnt = 0; + +sub new($;$) { + my ($class, $id) = @_; + my $job = { }; + bless $job; + if (defined $id) { + $job->{'ID'} = $id; + } else { + $job_cnt++; + $job->{'ID'} = POSIX::strftime("%Y%m%d-%H%M%S-$$-$job_cnt", localtime); + } + $job->{'Subject'} = '(no subject)'; + return $job; +} + +sub new_from_file($$;$) { + my ($class, $file, $header_only) = @_; + my $job = { }; + open T, '<', $file or die "Cannot open $file: $!"; + while () { + chomp; + /^$/ and last; + /^([A-Z][A-Za-z0-9-]*):\s*(.*)/ or die "Cannot load $file: Header syntax error"; + !defined $job->{$1} or die "Cannot load $file: Header $1 re-defined"; + $job->{$1} = $2; + } + if (!$header_only) { + my @cmds = ; + $job->{'body'} = join("", @cmds); + } + close T; + $job->{'Subject'} //= '?'; + $job->{'ID'} or die "Cannot load $file: Missing ID"; + return bless $job; +} + +sub attr($$;$) { + my ($job, $attr, $val) = @_; + $job->{$attr} = $val if defined $val; + return $job->{$attr}; +} + +sub dump($) { + my ($job) = @_; + for my $k (sort keys %$job) { + print "$k: ", $job->{$k}, "\n"; + } +} + +sub save($;$) { + my ($job, $fn) = @_; + -d "tmp" or mkdir "tmp" or die "Cannot create directory tmp: $!"; + $fn //= 'tmp/t-' . $job->{'ID'}; + open T, '>', $fn or die "Cannot create $fn: $!"; + for my $k (sort grep { /^[A-Z]/ } keys %$job) { + print T "$k: ", $job->{$k}, "\n"; + } + print T "\n"; + print T $job->{'body'} if defined $job->{'body'}; + close T; + return $fn; +} + +package BEX::Queue; + +sub new($;$) { + my ($class, $name) = @_; + $name //= 'queue'; + -d $name or die "Queue directory $name does not exist\n"; + for my $d ("hosts", "jobs") { + -d "$name/$d" or mkdir "$name/$d" or die "Cannot create directory $name/$d: $!"; + } + my $queue = { + 'Name' => $name, + 'MetaCache' => {}, + }; + return bless $queue; +} + +sub host_dir($$) { + my ($queue, $machine) = @_; + return $queue->{'Name'} . '/hosts/' . $machine; +} + +sub queue_file($$) { + my ($queue, $machine, $jid) = @_; + return $queue->host_dir($machine) . '/q-' . $jid; +} + +sub status_file($$) { + my ($queue, $machine, $jid) = @_; + return $queue->host_dir($machine) . '/s-' . $jid; +} + +sub job_file($$) { + my ($queue, $jid) = @_; + return $queue->{'Name'} . '/jobs/j-' . $jid; +} + +sub enqueue($$$) { + my ($queue, $machine, $job) = @_; + my $qf = $queue->queue_file($machine, $job->{'ID'}); + if (-f $qf) { return 0; } + my $fn = $queue->job_file($job->{'ID'}); + -f $fn or $job->save($fn); + my $dir = $queue->host_dir($machine); + -d $dir or mkdir $dir or die "Cannot create directory $dir: $!"; + symlink "../../jobs/j-" . $job->{'ID'}, $qf or die "Cannot create $qf: $!"; + return 1; +} + +sub scan($$) { + my ($queue, $machine) = @_; + my @list = (); + if (opendir D, $queue->host_dir($machine)) { + while ($_ = readdir D) { + s/^q-// or next; + push @list, $_; + } + closedir D; + } + return @list; +} + +sub job_metadata($$) { + my ($queue, $jid) = @_; + my $cache = $queue->{'MetaCache'}; + if (!defined $cache->{$jid}) { + $cache->{$jid} = BEX::Job->new_from_file($queue->job_file($jid), 1); + } + return $cache->{$jid}; +} + +sub read_job_status($$$) { + my ($queue, $machine, $jid) = @_; + my %s = (); + my $sf = status_file($queue, $machine, $jid); + if (open S, '<', $sf) { + while () { + chomp; + /^(\w+):\s*(.*)/ or die "Parse error in $sf"; + $s{$1} = $2; + } + close S; + } + return \%s; +} + +sub write_job_status($$$$) { + my ($queue, $machine, $jid, $stat) = @_; + my $sf = status_file($queue, $machine, $jid); + open S, '>', $sf or die "Cannot create $sf: $!"; + for my $k (sort keys %$stat) { + print S "$k: ", $stat{$k}, "\n"; + } + close S; +} + +42; -- 2.39.2