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