1 # Batch EXecutor 2.0 -- Queues
2 # (c) 2011 Martin Mares <mj@ucw.cz>
16 my ($class, $name) = @_;
18 -d $name or die "Queue directory $name does not exist\n";
19 for my $d ("hosts", "jobs") {
20 -d "$name/$d" or mkdir "$name/$d" or die "Cannot create directory $name/$d: $!";
30 my ($queue, $machine, $jid) = @_;
31 return $queue->host_dir($machine) . '/' . $jid . '.log';
34 # Most actions have to be logged by the caller
36 my ($queue, $mach, $jid, $stat, $msg) = @_;
37 my $t = POSIX::strftime("%Y-%m-%d %H:%M:%S", localtime);
38 my $m = join(" ", $t, $mach, $jid, $stat);
39 $m .= " $msg" if defined $msg;
41 my $fh = $queue->{'LogFH'} //= new IO::File $queue->{'Name'} . '/log', '>>' or die "Cannot open log: $!";
44 # Append to the per-job log file
45 if (open L, '>>', $queue->log_file($mach, $jid)) {
52 my ($queue, $machine) = @_;
53 return $queue->{'Name'} . '/hosts/' . $machine;
57 my ($queue, $machine, $jid) = @_;
58 return $queue->host_dir($machine) . '/' . $jid . '.job';
62 my ($queue, $machine, $jid) = @_;
63 return $queue->host_dir($machine) . '/' . $jid . '.stat';
67 my ($queue, $machine, $jid) = @_;
68 return $queue->host_dir($machine) . '/' . $jid . '.tmp';
72 my ($queue, $jid) = @_;
73 return $queue->{'Name'} . '/jobs/' . $jid. '.job';
77 my ($queue, $machine, $job) = @_;
78 my $qf = $queue->queue_file($machine, $job->id);
82 my $fn = $queue->job_file($job->id);
83 -f $fn or $job->save($fn);
84 my $dir = $queue->host_dir($machine);
85 -d $dir or mkdir $dir or die "Cannot create directory $dir: $!";
86 symlink '../../jobs/' . $job->id . '.job', $qf or die "Cannot create $qf: $!";
91 my ($queue, $machine) = @_;
93 if (opendir D, $queue->host_dir($machine)) {
94 while ($_ = readdir D) {
105 my ($queue, $machine, $jid, $force_remove) = @_;
106 if ($BEX::Config::keep_history && !$force_remove) {
107 my $s = $queue->{'Name'} . '/hosts/' . $machine;
108 my $d = $queue->{'Name'} . '/history/' . $machine;
109 File::Path::mkpath($d);
110 for my $suff ('job', 'stat', 'log') {
111 my $src = "$s/$jid.$suff";
112 my $dst = "$d/$jid.$suff";
114 rename $src, $dst or die "Cannot rename $src to $dst: $!";
116 # Might be present from the previous incarnation of the same job
121 unlink $queue->queue_file($machine, $jid);
122 unlink $queue->status_file($machine, $jid);
123 unlink $queue->log_file($machine, $jid);
125 unlink $queue->temp_file($machine, $jid);
128 sub job_metadata($$) {
129 my ($queue, $jid) = @_;
130 my $cache = $queue->{'MetaCache'};
131 if (!defined $cache->{$jid}) {
132 $cache->{$jid} = BEX::Job->new_from_file($queue->job_file($jid), 1);
134 return $cache->{$jid};
138 my ($queue, $jid) = @_;
139 return $queue->job_metadata($jid)->name;
142 sub read_job_status($$$) {
143 my ($queue, $machine, $jid) = @_;
145 my $sf = $queue->status_file($machine, $jid);
146 if (open S, '<', $sf) {
149 /^(\w+):\s*(.*)/ or die "Parse error in $sf";
157 sub write_job_status($$$$) {
158 my ($queue, $machine, $jid, $stat) = @_;
159 my $sf = $queue->status_file($machine, $jid);
160 open S, '>', "$sf.$$" or die "Cannot create $sf.$$: $!";
161 for my $k (sort keys %$stat) {
162 print S "$k: ", $stat->{$k}, "\n" if defined $stat->{$k};
165 rename "$sf.$$", $sf or die "Cannot rename $sf.$$ to $sf: $!";
168 sub update_job_status($$$$;$) {
169 my ($queue, $machine, $jid, $stat, $msg) = @_;
175 $queue->write_job_status($machine, $jid, $s);
176 $queue->log($machine, $jid, $stat, $msg);
180 my ($queue, $machine, $jid) = @_;
181 my $lock = $queue->{'Name'};
183 $lock .= "/hosts/$machine/$jid.lock";
184 } elsif (defined $machine) {
185 $lock .= "/hosts/$machine/lock";
191 # Whenever we want to run a job on a machine, we must obtain a lock;
192 # at most one lock can be held at a time by a single BEX::Queue object.
193 # See the description of locking schemes in BEX::Config.
195 my ($queue, $machine, $jid) = @_;
197 given ($BEX::Config::locking_scheme) {
199 $lock = lock_name($queue, undef, undef);
202 defined($machine) or return 1;
203 $lock = lock_name($queue, $machine, undef);
206 defined($machine) && defined($jid) or return 1;
207 $lock = lock_name($queue, $machine, $jid);
209 when ('none') { return 1; }
210 default { die "Invalid BEX::Config::locking_scheme"; }
212 if (defined($queue->{'LockName'})) {
213 return 1 if ($queue->{'LockName'} eq $lock);
216 open $queue->{'LockHandle'}, '>>', $lock or die "Cannot create $lock: $!";
217 if (!flock($queue->{'LockHandle'}, LOCK_EX | LOCK_NB)) {
218 close $queue->{'LockHandle'};
219 delete $queue->{'LockHandle'};
222 $queue->{'LockName'} = $lock;
228 defined $queue->{'LockName'} or return;
229 unlink $queue->{'LockName'};
230 flock $queue->{'LockHandle'}, LOCK_UN;
231 close $queue->{'LockHandle'};
232 delete $queue->{'LockHandle'};
233 delete $queue->{'LockName'};
236 # Unsafe (does not check fcntl, only existence of a lock file), but should be enough for bq
238 my ($queue, $machine, $jid) = @_;
239 given ($BEX::Config::locking_scheme) {
241 when ('host') { return unless defined $machine; }
242 when ('jid') { return unless defined $jid; }
243 when ('none') { return; }
245 my $lock = lock_name($queue, $machine, $jid);