lib/bin/bex-prun

   1 #!/usr/bin/perl
   2 # Batch EXecutor 3.0 -- Parallel Execution Using Screen
   3 # (c) 2011-2012 Martin Mares <mj@ucw.cz>
   4
   5 use strict;
   6 use warnings;
   7 use feature 'switch';
   8 use Getopt::Long;
   9 use POSIX;
  10 use BEX;
  11
  12 my $queue_name;
  13 my $text_mode;
  14 my $debug_children;
  15
  16 sub usage() {
  17         print <<AMEN ;
  18 Usage: bex prun [<options>] [[!]<machine-or-class> ...]
  19
  20 Options:
  21     --debug-children    Log stdout and stderr to ./debug.log
  22 -p, --parallel=<n>      Set limit on the number of jobs run in parallel
  23 -q, --queue=<name>      Run jobs in the given queue
  24     --text              Use plain-text user interface instead of curses
  25 AMEN
  26         exit 0;
  27 }
  28
  29 GetOptions(
  30         "q|queue=s" => \$queue_name,
  31         "text!" => \$text_mode,
  32         "debug-children!" => \$debug_children,
  33         "p|parallel=i" => \$BEX::Config::max_parallel_jobs,
  34         "help" => \&usage,
  35 ) or die "Try `bex prun --help' for more information.\n";
  36
  37 system 'tmux', 'has-session';
  38 !$? or die "You need to start tmux first.\n";
  39
  40 my $queue = BEX::Queue->new($queue_name);
  41 my $fifo_name = $queue->{'Path'} . '/status-fifo';
  42 unlink $fifo_name;
  43 mkfifo $fifo_name, 0700 or die "Cannot create $fifo_name: $!";
  44 open FIFO, '+<', $fifo_name or die "Cannot open $fifo_name: $!";
  45
  46 my $ui = ($text_mode ? BEX::bprun::text->new : BEX::bprun::curses->new);
  47
  48 my @machines = ();
  49 for my $mach (BEX::Config::parse_machine_list(@ARGV ? @ARGV : '*')) {
  50         my @jobs = $queue->scan($mach);
  51         @jobs or next;
  52         push @machines, $mach;
  53         for (@jobs) { $ui->update($mach, $_, 'READY'); }
  54 }
  55
  56 my %running = ();
  57 my $max = $BEX::Config::max_parallel_jobs;
  58
  59 while (keys %running || @machines) {
  60         if (@machines && keys %running < $max) {
  61                 my $mach = shift @machines;
  62                 $ui->update($mach, undef, 'START');
  63                 my @tm = ('tmux', 'new-window', '-n', $mach, '-d');
  64                 my $P5LIB = $ENV{"PERL5LIB"} // "";
  65                 my @cmd = (
  66                         "BEX_HOME='$BEX::Config::home'",
  67                         "BEX_LIB='$BEX::Config::lib'",
  68                         "PERL5LIB='$P5LIB'",
  69                         "$BEX::Config::lib/bin/bex-run",
  70                         "--status-fifo=$fifo_name",
  71                         "--queue=" . $queue->{'Name'},
  72                         $mach,
  73                         );
  74                 push @cmd, ">>debug.log", "2>&1" if $debug_children;
  75                 push @tm, join(" ", @cmd);
  76                 system @tm;
  77                 !$? or $ui->update($mach, undef, 'INTERR');
  78                 $running{$mach} = 'START';
  79                 next;
  80         }
  81         $_ = <FIFO>;
  82         chomp;
  83         my ($mach, $jid, $stat) = /^! (\S+) (\S+) (\S+)$/;
  84         if (!defined $stat) {
  85                 $ui->err("Received invalid status message <$_>");
  86                 next;
  87         }
  88         if (!defined $running{$mach}) {
  89                 $ui->err("Received status message <$_> for a machine which does not run");
  90                 next;
  91         }
  92         $running{$mach} = $stat;
  93         $ui->update($mach, ($jid eq '-' ? undef : $jid), $stat);
  94         if ($stat eq 'DONE') {
  95                 delete $running{$mach};
  96         }
  97 }
  98
  99 close FIFO;
 100 unlink $fifo_name;
 101 $ui->done;
 102
 103 package BEX::bprun::text;
 104
 105 sub new($) {
 106         return bless {};
 107 }
 108
 109 sub done($) {
 110 }
 111
 112 sub update($$$$) {
 113         my ($ui, $mach, $jid, $stat) = @_;
 114         print +($mach // '-'), (defined($jid) ? ":$jid" : ""), " $stat\n";
 115 }
 116
 117 sub err($$) {
 118         my ($ui, $msg) = @_;
 119         print STDERR "ERROR: $msg\n";
 120 }
 121
 122 package BEX::bprun::curses;
 123
 124 use Curses;
 125
 126 my $C;
 127
 128 my $nrows;
 129 my @by_row;
 130 my %by_host;
 131
 132 my %host_state;
 133 my %host_cnt;
 134
 135 my %job_state;
 136 my %job_cnt;
 137
 138 my %host_last_fail_job;
 139 my %host_last_fail_stat;
 140
 141 my @states;
 142 my %state_to_pri;
 143
 144 BEGIN {
 145         @by_row = ();
 146         %by_host = ();
 147         @states = qw(unknown ready running done failed);
 148         %state_to_pri = (
 149                 'unknown' => 0,
 150                 'ready' => 1,
 151                 'done' => 2,
 152                 'failed' => 3,
 153                 'running' => 4,
 154         );
 155 }
 156
 157 sub new($) {
 158         $C = new Curses;
 159         start_color;
 160         has_colors && COLORS >= 8 && COLOR_PAIRS >= 8 or die "Your terminal is too dumb for me\n";
 161         cbreak; noecho;
 162         $C->intrflush(0);
 163         $C->keypad(1);
 164         $C->meta(1);
 165         $C->clear;
 166         init_pair(1, COLOR_YELLOW, COLOR_BLUE);
 167         init_pair(2, COLOR_YELLOW, COLOR_RED);
 168         init_pair(3, COLOR_YELLOW, COLOR_BLACK);
 169         init_pair(4, COLOR_RED, COLOR_BLACK);
 170         init_pair(5, COLOR_BLUE, COLOR_BLACK);
 171
 172         $nrows = $C->getmaxy - 2;
 173         if ($BEX::Config::max_parallel_jobs > $nrows) {
 174                 $BEX::Config::max_parallel_jobs = $nrows;
 175         }
 176
 177         %host_state = %host_cnt = ();
 178         %job_state = %job_cnt = ();
 179         for my $s (@states) {
 180                 $host_cnt{$s} = 0;
 181                 $job_cnt{'*'}{$s} = 0;
 182         }
 183
 184         my $ui = bless {};
 185         $ui->refresh_status;
 186         return $ui;
 187 }
 188
 189 sub done($)
 190 {
 191         $C->bkgdset(COLOR_PAIR(1) | A_BOLD);
 192         $C->addstr($C->getmaxy-1, 0, "Press any key to quit...");
 193         $C->clrtoeol;
 194         $C->getch;
 195         endwin;
 196 }
 197
 198 sub err($$) {
 199         my ($ui, $msg) = @_;
 200         $C->bkgdset(COLOR_PAIR(2) | A_BOLD);
 201         $C->addnstr($C->getmaxy-1, 0, "ERROR: $msg", $C->getmaxx);
 202         $C->clrtoeol;
 203         $C->refresh;
 204 }
 205
 206 sub set_host_status($$$) {
 207         my ($ui, $mach, $stat) = @_;
 208         my $prev_stat = $host_state{$mach};
 209         if (defined $prev_stat) {
 210                 $host_cnt{$prev_stat}--;
 211         } else {
 212                 for my $s (@states) { $job_cnt{$mach}{$s} = 0; }
 213         }
 214         $host_state{$mach} = $stat;
 215         $host_cnt{$stat}++;
 216 }
 217
 218 sub set_job_status($$$$) {
 219         my ($ui, $mach, $jid, $stat) = @_;
 220         my $prev_stat = $job_state{$mach}{$jid} // 'unknown';
 221         $job_cnt{$mach}{$prev_stat}--;
 222         $job_cnt{'*'}{$prev_stat}--;
 223         $job_state{$mach}{$jid} = $stat;
 224         $job_cnt{$mach}{$stat}++;
 225         $job_cnt{'*'}{$stat}++;
 226 }
 227
 228 sub refresh_status($) {
 229         $C->bkgdset(COLOR_PAIR(1) | A_BOLD);
 230         $C->addnstr(0, 0,
 231                 sprintf("BEX  Hosts: %dR %dD %dE %dW  Jobs: %dR %dD %dE %dW",
 232                         $host_cnt{'running'},
 233                         $host_cnt{'done'},
 234                         $host_cnt{'failed'},
 235                         $host_cnt{'ready'},
 236                         $job_cnt{'*'}{'running'},
 237                         $job_cnt{'*'}{'done'},
 238                         $job_cnt{'*'}{'failed'},
 239                         $job_cnt{'*'}{'ready'},
 240                 ), $C->getmaxx);
 241         $C->clrtoeol;
 242         $C->refresh;
 243 }
 244
 245 sub get_slot($) {
 246         my ($mach) = @_;
 247         my $s = $by_host{$mach};
 248         if (!defined $s) {
 249                 $s = $by_host{$mach} = { 'Host' => $mach };
 250         }
 251         return $s;
 252 }
 253
 254 my $place_counter;
 255
 256 sub place_slot($) {
 257         my ($s) = @_;
 258         $s->{'LastUpdate'} = $place_counter++;
 259         return $s if defined $s->{'Row'};
 260
 261         my $pri = $state_to_pri{$host_state{$s->{'Host'}}};
 262         my ($best, $besti);
 263         my $bestpri = -1;
 264         for my $i (0..$nrows-1) {
 265                 my $r = $by_row[$i];
 266                 if (!defined $r) {
 267                         $besti = $i;
 268                         $best = undef;
 269                         last;
 270                 }
 271                 my $rpri = $state_to_pri{$host_state{$r->{'Host'}}};
 272                 next if $rpri > $pri;
 273
 274                 if ($rpri < $bestpri ||
 275                     $rpri == $bestpri && $r->{'LastUpdate'} < $best->{'LastUpdate'}) {
 276                         # Trick: $best must be defined, as otherwise $bestpri == -1
 277                         $best = $r;
 278                         $besti = $i;
 279                         $bestpri = $rpri;
 280                 }
 281         }
 282
 283         if (defined $besti) {
 284                 if ($best) {
 285                         delete $best->{'Row'};
 286                 }
 287                 $s->{'Row'} = $besti;
 288                 $by_row[$besti] = $s;
 289         }
 290 }
 291
 292 sub redraw_slot($) {
 293         my ($s) = @_;
 294         my $r = $s->{'Row'} // return;
 295         $r++;
 296         my $mach = $s->{'Host'};
 297         my $stat = $s->{'Status'} // "?";
 298         my $jid = $s->{'Job'} // "";
 299         my $jname = ($jid eq "" ? "" : $queue->job_name($jid));
 300         my $jcnt = $job_cnt{$mach};
 301         if ($jcnt->{'running'}) {
 302                 if ($jcnt->{'failed'}) {
 303                         $C->bkgdset(COLOR_PAIR(4) | A_BOLD);
 304                 } else {
 305                         $C->bkgdset(COLOR_PAIR(3) | A_BOLD);
 306                 }
 307         } else {
 308                 if ($jcnt->{'failed'}) {
 309                         $C->bkgdset(COLOR_PAIR(4));
 310                 } else {
 311                         $C->bkgdset(0);
 312                 }
 313         }
 314         $C->addstr($r, 0, sprintf("%-20.20s", $mach));
 315         if ($jcnt->{'failed'}) {
 316                 $C->bkgdset(COLOR_PAIR(4));
 317                 $C->addstr(sprintf("%3dE ", $jcnt->{'failed'}));
 318         } else {
 319                 $C->bkgdset(0);
 320                 $C->addstr("     ");
 321         }
 322         $C->bkgdset(0);
 323         $C->addstr(sprintf("%3dD %3dW", $jcnt->{'done'}, $jcnt->{'ready'}));
 324         if ($stat eq 'DONE') {
 325                 if (defined $host_last_fail_stat{$mach}) {
 326                         $C->bkgdset(($host_last_fail_stat{$mach} eq 'NOPING') ? COLOR_PAIR(5) : COLOR_PAIR(4));
 327                         $C->addstr(sprintf("  %-8s %s", $host_last_fail_stat{$mach}, $queue->job_name($host_last_fail_job{$mach})));
 328                 }
 329         } else {
 330                 my $text = sprintf("  %-8s %s", $stat, $jname);
 331                 $C->addstr($text);
 332         }
 333         $C->clrtoeol;
 334         $C->refresh;
 335 }
 336
 337 sub update($$$$) {
 338         my ($ui, $mach, $jid, $stat) = @_;
 339         given ($stat) {
 340                 when ('READY') {
 341                         # Pseudo-state generated internally
 342                         $ui->set_host_status($mach, 'ready');
 343                         $ui->set_job_status($mach, $jid, 'ready');
 344                 }
 345                 when ('OK') {
 346                         $ui->set_job_status($mach, $jid, 'done');
 347                 }
 348                 when (['FAILED', 'INTERR', 'NOPING', 'PREPFAIL', 'NOXFER']) {
 349                         $ui->set_job_status($mach, $jid, 'failed');
 350                         $host_last_fail_job{$mach} = $jid;
 351                         $host_last_fail_stat{$mach} = $stat;
 352                 }
 353                 when ('DONE') {
 354                         if ($job_cnt{$mach}{'failed'}) {
 355                                 $ui->set_host_status($mach, 'failed');
 356                         } else {
 357                                 $ui->set_host_status($mach, 'done');
 358                         }
 359                 }
 360                 when ('INIT') {
 361                         $ui->set_host_status($mach, 'running');
 362                         $ui->set_job_status($mach, $jid, 'running') if defined $jid;
 363                 }
 364                 when ('LOCKED') {
 365                         if (defined $jid) {
 366                                 $ui->set_job_status($mach, $jid, 'failed');
 367                         } else {
 368                                 for my $j (keys %{$job_state{$mach}}) {
 369                                         $ui->set_job_status($mach, $jid, 'failed');
 370                                 }
 371                                 $ui->set_host_status($mach, 'failed');
 372                                 $host_last_fail_job{$mach} = $jid;
 373                                 $host_last_fail_stat{$mach} = $stat;
 374                         }
 375                 }
 376                 when (['START', 'PING', 'SEND', 'RUN']) {
 377                 }
 378                 default {
 379                         $ui->err("Received unknown job status $stat");
 380                 }
 381         }
 382         my $s = get_slot($mach);
 383         $s->{'Job'} = $jid;
 384         $s->{'Status'} = $stat;
 385         place_slot($s);
 386         redraw_slot($s);
 387         $ui->refresh_status;
 388 }