sdk/cli/bin/crunch-job

   1 #!/usr/bin/perl
   2 # -*- mode: perl; perl-indent-level: 2; -*-
   3
   4 =head1 NAME
   5
   6 crunch-job: Execute job steps, save snapshots as requested, collate output.
   7
   8 =head1 SYNOPSIS
   9
  10 Obtain job details from Arvados, run tasks on compute nodes (typically
  11 invoked by scheduler on controller):
  12
  13  crunch-job --job x-y-z
  14
  15 Obtain job details from command line, run tasks on local machine
  16 (typically invoked by application or developer on VM):
  17
  18  crunch-job --job '{"script_version":"/path/to/tree","script":"scriptname",...}'
  19
  20 =head1 OPTIONS
  21
  22 =over
  23
  24 =item --force-unlock
  25
  26 If the job is already locked, steal the lock and run it anyway.
  27
  28 =item --git-dir
  29
  30 Path to .git directory where the specified commit is found.
  31
  32 =item --job-api-token
  33
  34 Arvados API authorization token to use during the course of the job.
  35
  36 =back
  37
  38 =head1 RUNNING JOBS LOCALLY
  39
  40 crunch-job's log messages appear on stderr along with the job tasks'
  41 stderr streams. The log is saved in Keep at each checkpoint and when
  42 the job finishes.
  43
  44 If the job succeeds, the job's output locator is printed on stdout.
  45
  46 While the job is running, the following signals are accepted:
  47
  48 =over
  49
  50 =item control-C, SIGINT, SIGQUIT
  51
  52 Save a checkpoint, terminate any job tasks that are running, and stop.
  53
  54 =item SIGALRM
  55
  56 Save a checkpoint and continue.
  57
  58 =item SIGHUP
  59
  60 Refresh node allocation (i.e., check whether any nodes have been added
  61 or unallocated) and attributes of the Job record that should affect
  62 behavior (e.g., cancel job if cancelled_at becomes non-nil).
  63
  64 =back
  65
  66 =cut
  67
  68
  69 use strict;
  70 use POSIX ':sys_wait_h';
  71 use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
  72 use Arvados;
  73 use Getopt::Long;
  74 use IPC::Open2;
  75 use IO::Select;
  76
  77 $ENV{"TMPDIR"} ||= "/tmp";
  78 unless (defined $ENV{"CRUNCH_TMP"}) {
  79   $ENV{"CRUNCH_TMP"} = $ENV{"TMPDIR"} . "/crunch-job";
  80   if ($ENV{"USER"} ne "crunch" && $< != 0) {
  81     # use a tmp dir unique for my uid
  82     $ENV{"CRUNCH_TMP"} .= "-$<";
  83   }
  84 }
  85 $ENV{"JOB_WORK"} = $ENV{"CRUNCH_TMP"} . "/work";
  86 $ENV{"CRUNCH_INSTALL"} = "$ENV{CRUNCH_TMP}/opt";
  87 $ENV{"CRUNCH_WORK"} = $ENV{"JOB_WORK"}; # deprecated
  88 mkdir ($ENV{"JOB_WORK"});
  89
  90 my $force_unlock;
  91 my $git_dir;
  92 my $jobspec;
  93 my $job_api_token;
  94 my $resume_stash;
  95 GetOptions('force-unlock' => \$force_unlock,
  96            'git-dir=s' => \$git_dir,
  97            'job=s' => \$jobspec,
  98            'job-api-token=s' => \$job_api_token,
  99            'resume-stash=s' => \$resume_stash,
 100     );
 101
 102 if (defined $job_api_token) {
 103   $ENV{ARVADOS_API_TOKEN} = $job_api_token;
 104 }
 105
 106 my $have_slurm = exists $ENV{SLURM_JOBID} && exists $ENV{SLURM_NODELIST};
 107 my $job_has_uuid = $jobspec =~ /^[-a-z\d]+$/;
 108 my $local_job = !$job_has_uuid;
 109
 110
 111 $SIG{'USR1'} = sub
 112 {
 113   $main::ENV{CRUNCH_DEBUG} = 1;
 114 };
 115 $SIG{'USR2'} = sub
 116 {
 117   $main::ENV{CRUNCH_DEBUG} = 0;
 118 };
 119
 120
 121
 122 my $arv = Arvados->new('apiVersion' => 'v1');
 123 my $metastream;
 124
 125 my $User = $arv->{'users'}->{'current'}->execute;
 126
 127 my $Job = {};
 128 my $job_id;
 129 my $dbh;
 130 my $sth;
 131 if ($job_has_uuid)
 132 {
 133   $Job = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec);
 134   if (!$force_unlock) {
 135     if ($Job->{'is_locked_by_uuid'}) {
 136       croak("Job is locked: " . $Job->{'is_locked_by_uuid'});
 137     }
 138     if ($Job->{'success'} ne undef) {
 139       croak("Job 'success' flag (" . $Job->{'success'} . ") is not null");
 140     }
 141     if ($Job->{'running'}) {
 142       croak("Job 'running' flag is already set");
 143     }
 144     if ($Job->{'started_at'}) {
 145       croak("Job 'started_at' time is already set (" . $Job->{'started_at'} . ")");
 146     }
 147   }
 148 }
 149 else
 150 {
 151   $Job = JSON::decode_json($jobspec);
 152
 153   if (!$resume_stash)
 154   {
 155     map { croak ("No $_ specified") unless $Job->{$_} }
 156     qw(script script_version script_parameters);
 157   }
 158
 159   $Job->{'is_locked_by_uuid'} = $User->{'uuid'};
 160   $Job->{'started_at'} = gmtime;
 161
 162   $Job = $arv->{'jobs'}->{'create'}->execute('job' => $Job);
 163
 164   $job_has_uuid = 1;
 165 }
 166 $job_id = $Job->{'uuid'};
 167
 168 # $metastream = Warehouse::Stream->new(whc => new Warehouse);
 169 # $metastream->clear;
 170 # $metastream->name('.');
 171 # $metastream->write_start($job_id . '.log.txt');
 172
 173
 174 $Job->{'runtime_constraints'} ||= {};
 175 $Job->{'runtime_constraints'}->{'max_tasks_per_node'} ||= 0;
 176 my $max_ncpus = $Job->{'runtime_constraints'}->{'max_tasks_per_node'};
 177
 178
 179 Log (undef, "check slurm allocation");
 180 my @slot;
 181 my @node;
 182 # Should use $ENV{SLURM_TASKS_PER_NODE} instead of sinfo? (eg. "4(x3),2,4(x2)")
 183 my @sinfo;
 184 if (!$have_slurm)
 185 {
 186   my $localcpus = 0 + `grep -cw ^processor /proc/cpuinfo` || 1;
 187   push @sinfo, "$localcpus localhost";
 188 }
 189 if (exists $ENV{SLURM_NODELIST})
 190 {
 191   push @sinfo, `sinfo -h --format='%c %N' --nodes='$ENV{SLURM_NODELIST}'`;
 192 }
 193 foreach (@sinfo)
 194 {
 195   my ($ncpus, $slurm_nodelist) = split;
 196   $ncpus = $max_ncpus if $max_ncpus && $ncpus > $max_ncpus;
 197
 198   my @nodelist;
 199   while ($slurm_nodelist =~ s/^([^\[,]+?(\[.*?\])?)(,|$)//)
 200   {
 201     my $nodelist = $1;
 202     if ($nodelist =~ /\[((\d+)(-(\d+))?(,(\d+)(-(\d+))?)*)\]/)
 203     {
 204       my $ranges = $1;
 205       foreach (split (",", $ranges))
 206       {
 207         my ($a, $b);
 208         if (/(\d+)-(\d+)/)
 209         {
 210           $a = $1;
 211           $b = $2;
 212         }
 213         else
 214         {
 215           $a = $_;
 216           $b = $_;
 217         }
 218         push @nodelist, map {
 219           my $n = $nodelist;
 220           $n =~ s/\[[-,\d]+\]/$_/;
 221           $n;
 222         } ($a..$b);
 223       }
 224     }
 225     else
 226     {
 227       push @nodelist, $nodelist;
 228     }
 229   }
 230   foreach my $nodename (@nodelist)
 231   {
 232     Log (undef, "node $nodename - $ncpus slots");
 233     my $node = { name => $nodename,
 234                  ncpus => $ncpus,
 235                  losing_streak => 0,
 236                  hold_until => 0 };
 237     foreach my $cpu (1..$ncpus)
 238     {
 239       push @slot, { node => $node,
 240                     cpu => $cpu };
 241     }
 242   }
 243   push @node, @nodelist;
 244 }
 245
 246
 247
 248 # Ensure that we get one jobstep running on each allocated node before
 249 # we start overloading nodes with concurrent steps
 250
 251 @slot = sort { $a->{cpu} <=> $b->{cpu} } @slot;
 252
 253
 254
 255 my $jobmanager_id;
 256 if ($job_has_uuid)
 257 {
 258   # Claim this job, and make sure nobody else does
 259   unless ($Job->update_attributes('is_locked_by_uuid' => $User->{'uuid'}) &&
 260           $Job->{'is_locked_by_uuid'} == $User->{'uuid'}) {
 261     croak("Error while updating / locking job");
 262   }
 263   $Job->update_attributes('started_at' => scalar gmtime,
 264                           'running' => 1,
 265                           'success' => undef,
 266                           'tasks_summary' => { 'failed' => 0,
 267                                                'todo' => 1,
 268                                                'running' => 0,
 269                                                'done' => 0 });
 270 }
 271
 272
 273 Log (undef, "start");
 274 $SIG{'INT'} = sub { $main::please_freeze = 1; };
 275 $SIG{'QUIT'} = sub { $main::please_freeze = 1; };
 276 $SIG{'TERM'} = \&croak;
 277 $SIG{'TSTP'} = sub { $main::please_freeze = 1; };
 278 $SIG{'ALRM'} = sub { $main::please_info = 1; };
 279 $SIG{'CONT'} = sub { $main::please_continue = 1; };
 280 $SIG{'HUP'} = sub { $main::please_refresh = 1; };
 281
 282 $main::please_freeze = 0;
 283 $main::please_info = 0;
 284 $main::please_continue = 0;
 285 $main::please_refresh = 0;
 286 my $jobsteps_must_output_keys = 0;      # becomes 1 when any task outputs a key
 287
 288 grep { $ENV{$1} = $2 if /^(NOCACHE.*?)=(.*)/ } split ("\n", $$Job{knobs});
 289 $ENV{"CRUNCH_JOB_UUID"} = $job_id;
 290 $ENV{"JOB_UUID"} = $job_id;
 291
 292
 293 my @jobstep;
 294 my @jobstep_todo = ();
 295 my @jobstep_done = ();
 296 my @jobstep_tomerge = ();
 297 my $jobstep_tomerge_level = 0;
 298 my $squeue_checked;
 299 my $squeue_kill_checked;
 300 my $output_in_keep = 0;
 301 my $latest_refresh = scalar time;
 302
 303
 304
 305 if (defined $Job->{thawedfromkey})
 306 {
 307   thaw ($Job->{thawedfromkey});
 308 }
 309 else
 310 {
 311   my $first_task = $arv->{'job_tasks'}->{'create'}->execute('job_task' => {
 312     'job_uuid' => $Job->{'uuid'},
 313     'sequence' => 0,
 314     'qsequence' => 0,
 315     'parameters' => {},
 316                                                           });
 317   push @jobstep, { 'level' => 0,
 318                    'failures' => 0,
 319                    'arvados_task' => $first_task,
 320                  };
 321   push @jobstep_todo, 0;
 322 }
 323
 324
 325 my $build_script;
 326
 327
 328 $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{script_version};
 329
 330 my $skip_install = ($local_job && $Job->{script_version} =~ m{^/});
 331 if ($skip_install)
 332 {
 333   $ENV{"CRUNCH_SRC"} = $Job->{script_version};
 334   for my $src_path ("$ENV{CRUNCH_SRC}/arvados/sdk/python") {
 335     if (-d $src_path) {
 336       system("virtualenv", "$ENV{CRUNCH_TMP}/opt") == 0
 337           or croak ("virtualenv $ENV{CRUNCH_TMP}/opt failed: exit ".($?>>8));
 338       system ("cd $src_path && ./build.sh && \$CRUNCH_TMP/opt/bin/python setup.py install")
 339           == 0
 340           or croak ("setup.py in $src_path failed: exit ".($?>>8));
 341     }
 342   }
 343 }
 344 else
 345 {
 346   do {
 347     local $/ = undef;
 348     $build_script = <DATA>;
 349   };
 350   Log (undef, "Install revision ".$Job->{script_version});
 351   my $nodelist = join(",", @node);
 352
 353   # Clean out crunch_tmp/work, crunch_tmp/opt, crunch_tmp/src*
 354
 355   my $cleanpid = fork();
 356   if ($cleanpid == 0)
 357   {
 358     srun (["srun", "--nodelist=$nodelist", "-D", $ENV{'TMPDIR'}],
 359           ['bash', '-c', 'if mount | grep -q $JOB_WORK/; then sudo /bin/umount $JOB_WORK/* 2>/dev/null; fi; sleep 1; rm -rf $JOB_WORK $CRUNCH_TMP/opt $CRUNCH_TMP/src*']);
 360     exit (1);
 361   }
 362   while (1)
 363   {
 364     last if $cleanpid == waitpid (-1, WNOHANG);
 365     freeze_if_want_freeze ($cleanpid);
 366     select (undef, undef, undef, 0.1);
 367   }
 368   Log (undef, "Clean-work-dir exited $?");
 369
 370   # Install requested code version
 371
 372   my @execargs;
 373   my @srunargs = ("srun",
 374                   "--nodelist=$nodelist",
 375                   "-D", $ENV{'TMPDIR'}, "--job-name=$job_id");
 376
 377   $ENV{"CRUNCH_SRC_COMMIT"} = $Job->{script_version};
 378   $ENV{"CRUNCH_SRC"} = "$ENV{CRUNCH_TMP}/src";
 379
 380   my $commit;
 381   my $git_archive;
 382   my $treeish = $Job->{'script_version'};
 383   my $repo = $git_dir || $ENV{'CRUNCH_DEFAULT_GIT_DIR'};
 384   # Todo: let script_version specify repository instead of expecting
 385   # parent process to figure it out.
 386   $ENV{"CRUNCH_SRC_URL"} = $repo;
 387
 388   # Create/update our clone of the remote git repo
 389
 390   if (!-d $ENV{"CRUNCH_SRC"}) {
 391     system(qw(git clone), $repo, $ENV{"CRUNCH_SRC"}) == 0
 392         or croak ("git clone $repo failed: exit ".($?>>8));
 393     system("cd $ENV{CRUNCH_SRC} && git config clean.requireForce false");
 394   }
 395   `cd $ENV{CRUNCH_SRC} && git remote set-url origin \"\$CRUNCH_SRC_URL\" && git fetch -q origin`;
 396
 397   # If this looks like a subversion r#, look for it in git-svn commit messages
 398
 399   if ($treeish =~ m{^\d{1,4}$}) {
 400     my $gitlog = `cd $ENV{CRUNCH_SRC} && git log --pretty="format:%H" --grep="git-svn-id:.*\@$treeish " origin/master`;
 401     chomp $gitlog;
 402     if ($gitlog =~ /^[a-f0-9]{40}$/) {
 403       $commit = $gitlog;
 404       Log (undef, "Using commit $commit for script_version $treeish");
 405     }
 406   }
 407
 408   # If that didn't work, try asking git to look it up as a tree-ish.
 409
 410   if (!defined $commit) {
 411
 412     my $cooked_treeish = $treeish;
 413     if ($treeish !~ m{^[0-9a-f]{5,}$}) {
 414       # Looks like a git branch name -- make sure git knows it's
 415       # relative to the remote repo
 416       $cooked_treeish = "origin/$treeish";
 417     }
 418
 419     my $found = `cd $ENV{CRUNCH_SRC} && git rev-list -1 $cooked_treeish`;
 420     chomp $found;
 421     if ($found =~ /^[0-9a-f]{40}$/s) {
 422       $commit = $found;
 423       if ($commit ne $treeish) {
 424         # Make sure we record the real commit id in the database,
 425         # frozentokey, logs, etc. -- instead of an abbreviation or a
 426         # branch name which can become ambiguous or point to a
 427         # different commit in the future.
 428         $ENV{"CRUNCH_SRC_COMMIT"} = $commit;
 429         Log (undef, "Using commit $commit for tree-ish $treeish");
 430         if ($commit ne $treeish) {
 431           $Job->{'script_version'} = $commit;
 432           !$job_has_uuid or
 433               $Job->update_attributes('script_version' => $commit) or
 434               croak("Error while updating job");
 435         }
 436       }
 437     }
 438   }
 439
 440   if (defined $commit) {
 441     $ENV{"CRUNCH_SRC_COMMIT"} = $commit;
 442     @execargs = ("sh", "-c",
 443                  "mkdir -p $ENV{CRUNCH_INSTALL} && cd $ENV{CRUNCH_TMP} && perl -");
 444     $git_archive = `cd $ENV{CRUNCH_SRC} && git archive $commit`;
 445   }
 446   else {
 447     croak ("could not figure out commit id for $treeish");
 448   }
 449
 450   my $installpid = fork();
 451   if ($installpid == 0)
 452   {
 453     srun (\@srunargs, \@execargs, {}, $build_script . $git_archive);
 454     exit (1);
 455   }
 456   while (1)
 457   {
 458     last if $installpid == waitpid (-1, WNOHANG);
 459     freeze_if_want_freeze ($installpid);
 460     select (undef, undef, undef, 0.1);
 461   }
 462   Log (undef, "Install exited $?");
 463 }
 464
 465
 466
 467 foreach (qw (script script_version script_parameters runtime_constraints))
 468 {
 469   Log (undef,
 470        "$_ " .
 471        (ref($Job->{$_}) ? JSON::encode_json($Job->{$_}) : $Job->{$_}));
 472 }
 473 foreach (split (/\n/, $Job->{knobs}))
 474 {
 475   Log (undef, "knob " . $_);
 476 }
 477
 478
 479
 480 $main::success = undef;
 481
 482
 483
 484 ONELEVEL:
 485
 486 my $thisround_succeeded = 0;
 487 my $thisround_failed = 0;
 488 my $thisround_failed_multiple = 0;
 489
 490 @jobstep_todo = sort { $jobstep[$a]->{level} <=> $jobstep[$b]->{level}
 491                        or $a <=> $b } @jobstep_todo;
 492 my $level = $jobstep[$jobstep_todo[0]]->{level};
 493 Log (undef, "start level $level");
 494
 495
 496
 497 my %proc;
 498 my @freeslot = (0..$#slot);
 499 my @holdslot;
 500 my %reader;
 501 my $progress_is_dirty = 1;
 502 my $progress_stats_updated = 0;
 503
 504 update_progress_stats();
 505
 506
 507
 508 THISROUND:
 509 for (my $todo_ptr = 0; $todo_ptr <= $#jobstep_todo; $todo_ptr ++)
 510 {
 511   my $id = $jobstep_todo[$todo_ptr];
 512   my $Jobstep = $jobstep[$id];
 513   if ($Jobstep->{level} != $level)
 514   {
 515     next;
 516   }
 517
 518   pipe $reader{$id}, "writer" or croak ($!);
 519   my $flags = fcntl ($reader{$id}, F_GETFL, 0) or croak ($!);
 520   fcntl ($reader{$id}, F_SETFL, $flags | O_NONBLOCK) or croak ($!);
 521
 522   my $childslot = $freeslot[0];
 523   my $childnode = $slot[$childslot]->{node};
 524   my $childslotname = join (".",
 525                             $slot[$childslot]->{node}->{name},
 526                             $slot[$childslot]->{cpu});
 527   my $childpid = fork();
 528   if ($childpid == 0)
 529   {
 530     $SIG{'INT'} = 'DEFAULT';
 531     $SIG{'QUIT'} = 'DEFAULT';
 532     $SIG{'TERM'} = 'DEFAULT';
 533
 534     foreach (values (%reader))
 535     {
 536       close($_);
 537     }
 538     fcntl ("writer", F_SETFL, 0) or croak ($!); # no close-on-exec
 539     open(STDOUT,">&writer");
 540     open(STDERR,">&writer");
 541
 542     undef $dbh;
 543     undef $sth;
 544
 545     delete $ENV{"GNUPGHOME"};
 546     $ENV{"TASK_UUID"} = $Jobstep->{'arvados_task'}->{'uuid'};
 547     $ENV{"TASK_QSEQUENCE"} = $id;
 548     $ENV{"TASK_SEQUENCE"} = $level;
 549     $ENV{"JOB_SCRIPT"} = $Job->{script};
 550     while (my ($param, $value) = each %{$Job->{script_parameters}}) {
 551       $param =~ tr/a-z/A-Z/;
 552       $ENV{"JOB_PARAMETER_$param"} = $value;
 553     }
 554     $ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
 555     $ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
 556     $ENV{"TASK_WORK"} = $ENV{"JOB_WORK"}."/".$slot[$childslot]->{cpu};
 557     $ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}."/keep";
 558     $ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
 559     $ENV{"CRUNCH_NODE_SLOTS"} = $slot[$childslot]->{node}->{ncpus};
 560     $ENV{"PATH"} = $ENV{"CRUNCH_INSTALL"} . "/bin:" . $ENV{"PATH"};
 561
 562     $ENV{"GZIP"} = "-n";
 563
 564     my @srunargs = (
 565       "srun",
 566       "--nodelist=".$childnode->{name},
 567       qw(-n1 -c1 -N1 -D), $ENV{'TMPDIR'},
 568       "--job-name=$job_id.$id.$$",
 569         );
 570     my @execargs = qw(sh);
 571     my $build_script_to_send = "";
 572     my $command =
 573         "if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
 574         ."mkdir -p $ENV{JOB_WORK} $ENV{CRUNCH_TMP} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
 575         ."&& cd $ENV{CRUNCH_TMP} ";
 576     if ($build_script)
 577     {
 578       $build_script_to_send = $build_script;
 579       $command .=
 580           "&& perl -";
 581     }
 582     $ENV{"PYTHONPATH"} =~ s{^}{:} if $ENV{"PYTHONPATH"};
 583     $ENV{"PYTHONPATH"} =~ s{^}{$ENV{CRUNCH_SRC}/sdk/python}; # xxx hack
 584     $ENV{"PYTHONPATH"} =~ s{^}{$ENV{CRUNCH_SRC}/arvados/sdk/python:}; # xxx hack
 585     $ENV{"PYTHONPATH"} =~ s{$}{:/usr/local/arvados/src/sdk/python}; # xxx hack
 586     $command .=
 587         "&& exec arv-mount $ENV{TASK_KEEPMOUNT} --exec $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 588     my @execargs = ('bash', '-c', $command);
 589     srun (\@srunargs, \@execargs, undef, $build_script_to_send);
 590     exit (111);
 591   }
 592   close("writer");
 593   if (!defined $childpid)
 594   {
 595     close $reader{$id};
 596     delete $reader{$id};
 597     next;
 598   }
 599   shift @freeslot;
 600   $proc{$childpid} = { jobstep => $id,
 601                        time => time,
 602                        slot => $childslot,
 603                        jobstepname => "$job_id.$id.$childpid",
 604                      };
 605   croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
 606   $slot[$childslot]->{pid} = $childpid;
 607
 608   Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
 609   Log ($id, "child $childpid started on $childslotname");
 610   $Jobstep->{starttime} = time;
 611   $Jobstep->{node} = $childnode->{name};
 612   $Jobstep->{slotindex} = $childslot;
 613   delete $Jobstep->{stderr};
 614   delete $Jobstep->{finishtime};
 615
 616   splice @jobstep_todo, $todo_ptr, 1;
 617   --$todo_ptr;
 618
 619   $progress_is_dirty = 1;
 620
 621   while (!@freeslot
 622          ||
 623          (@slot > @freeslot && $todo_ptr+1 > $#jobstep_todo))
 624   {
 625     last THISROUND if $main::please_freeze;
 626     if ($main::please_info)
 627     {
 628       $main::please_info = 0;
 629       freeze();
 630       collate_output();
 631       save_meta(1);
 632       update_progress_stats();
 633     }
 634     my $gotsome
 635         = readfrompipes ()
 636         + reapchildren ();
 637     if (!$gotsome)
 638     {
 639       check_refresh_wanted();
 640       check_squeue();
 641       update_progress_stats();
 642       select (undef, undef, undef, 0.1);
 643     }
 644     elsif (time - $progress_stats_updated >= 30)
 645     {
 646       update_progress_stats();
 647     }
 648     if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
 649         ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
 650     {
 651       my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
 652           .($thisround_failed+$thisround_succeeded)
 653           .") -- giving up on this round";
 654       Log (undef, $message);
 655       last THISROUND;
 656     }
 657
 658     # move slots from freeslot to holdslot (or back to freeslot) if necessary
 659     for (my $i=$#freeslot; $i>=0; $i--) {
 660       if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
 661         push @holdslot, (splice @freeslot, $i, 1);
 662       }
 663     }
 664     for (my $i=$#holdslot; $i>=0; $i--) {
 665       if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
 666         push @freeslot, (splice @holdslot, $i, 1);
 667       }
 668     }
 669
 670     # give up if no nodes are succeeding
 671     if (!grep { $_->{node}->{losing_streak} == 0 &&
 672                     $_->{node}->{hold_count} < 4 } @slot) {
 673       my $message = "Every node has failed -- giving up on this round";
 674       Log (undef, $message);
 675       last THISROUND;
 676     }
 677   }
 678 }
 679
 680
 681 push @freeslot, splice @holdslot;
 682 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
 683
 684
 685 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
 686 while (%proc)
 687 {
 688   if ($main::please_continue) {
 689     $main::please_continue = 0;
 690     goto THISROUND;
 691   }
 692   $main::please_info = 0, freeze(), collate_output(), save_meta(1) if $main::please_info;
 693   readfrompipes ();
 694   if (!reapchildren())
 695   {
 696     check_refresh_wanted();
 697     check_squeue();
 698     update_progress_stats();
 699     select (undef, undef, undef, 0.1);
 700     killem (keys %proc) if $main::please_freeze;
 701   }
 702 }
 703
 704 update_progress_stats();
 705 freeze_if_want_freeze();
 706
 707
 708 if (!defined $main::success)
 709 {
 710   if (@jobstep_todo &&
 711       $thisround_succeeded == 0 &&
 712       ($thisround_failed == 0 || $thisround_failed > 4))
 713   {
 714     my $message = "stop because $thisround_failed tasks failed and none succeeded";
 715     Log (undef, $message);
 716     $main::success = 0;
 717   }
 718   if (!@jobstep_todo)
 719   {
 720     $main::success = 1;
 721   }
 722 }
 723
 724 goto ONELEVEL if !defined $main::success;
 725
 726
 727 release_allocation();
 728 freeze();
 729 if ($job_has_uuid) {
 730   $Job->update_attributes('output' => &collate_output(),
 731                           'running' => 0,
 732                           'success' => $Job->{'output'} && $main::success,
 733                           'finished_at' => scalar gmtime)
 734 }
 735
 736 if ($Job->{'output'})
 737 {
 738   eval {
 739     my $manifest_text = `arv keep get $Job->{'output'}`;
 740     $arv->{'collections'}->{'create'}->execute('collection' => {
 741       'uuid' => $Job->{'output'},
 742       'manifest_text' => $manifest_text,
 743     });
 744   };
 745   if ($@) {
 746     Log (undef, "Failed to register output manifest: $@");
 747   }
 748 }
 749
 750 Log (undef, "finish");
 751
 752 save_meta();
 753 exit 0;
 754
 755
 756
 757 sub update_progress_stats
 758 {
 759   $progress_stats_updated = time;
 760   return if !$progress_is_dirty;
 761   my ($todo, $done, $running) = (scalar @jobstep_todo,
 762                                  scalar @jobstep_done,
 763                                  scalar @slot - scalar @freeslot - scalar @holdslot);
 764   $Job->{'tasks_summary'} ||= {};
 765   $Job->{'tasks_summary'}->{'todo'} = $todo;
 766   $Job->{'tasks_summary'}->{'done'} = $done;
 767   $Job->{'tasks_summary'}->{'running'} = $running;
 768   if ($job_has_uuid) {
 769     $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
 770   }
 771   Log (undef, "status: $done done, $running running, $todo todo");
 772   $progress_is_dirty = 0;
 773 }
 774
 775
 776
 777 sub reapchildren
 778 {
 779   my $pid = waitpid (-1, WNOHANG);
 780   return 0 if $pid <= 0;
 781
 782   my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
 783                   . "."
 784                   . $slot[$proc{$pid}->{slot}]->{cpu});
 785   my $jobstepid = $proc{$pid}->{jobstep};
 786   my $elapsed = time - $proc{$pid}->{time};
 787   my $Jobstep = $jobstep[$jobstepid];
 788
 789   my $childstatus = $?;
 790   my $exitvalue = $childstatus >> 8;
 791   my $exitinfo = sprintf("exit %d signal %d%s",
 792                          $exitvalue,
 793                          $childstatus & 127,
 794                          ($childstatus & 128 ? ' core dump' : ''));
 795   $Jobstep->{'arvados_task'}->reload;
 796   my $task_success = $Jobstep->{'arvados_task'}->{success};
 797
 798   Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
 799
 800   if (!defined $task_success) {
 801     # task did not indicate one way or the other --> fail
 802     $Jobstep->{'arvados_task'}->{success} = 0;
 803     $Jobstep->{'arvados_task'}->save;
 804     $task_success = 0;
 805   }
 806
 807   if (!$task_success)
 808   {
 809     my $temporary_fail;
 810     $temporary_fail ||= $Jobstep->{node_fail};
 811     $temporary_fail ||= ($exitvalue == 111);
 812
 813     ++$thisround_failed;
 814     ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
 815
 816     # Check for signs of a failed or misconfigured node
 817     if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
 818         2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
 819       # Don't count this against jobstep failure thresholds if this
 820       # node is already suspected faulty and srun exited quickly
 821       if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
 822           $elapsed < 5) {
 823         Log ($jobstepid, "blaming failure on suspect node " .
 824              $slot[$proc{$pid}->{slot}]->{node}->{name});
 825         $temporary_fail ||= 1;
 826       }
 827       ban_node_by_slot($proc{$pid}->{slot});
 828     }
 829
 830     Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
 831                              ++$Jobstep->{'failures'},
 832                              $temporary_fail ? 'temporary ' : 'permanent',
 833                              $elapsed));
 834
 835     if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
 836       # Give up on this task, and the whole job
 837       $main::success = 0;
 838       $main::please_freeze = 1;
 839     }
 840     else {
 841       # Put this task back on the todo queue
 842       push @jobstep_todo, $jobstepid;
 843     }
 844     $Job->{'tasks_summary'}->{'failed'}++;
 845   }
 846   else
 847   {
 848     ++$thisround_succeeded;
 849     $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
 850     $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
 851     push @jobstep_done, $jobstepid;
 852     Log ($jobstepid, "success in $elapsed seconds");
 853   }
 854   $Jobstep->{exitcode} = $childstatus;
 855   $Jobstep->{finishtime} = time;
 856   process_stderr ($jobstepid, $task_success);
 857   Log ($jobstepid, "output " . $Jobstep->{'arvados_task'}->{output});
 858
 859   close $reader{$jobstepid};
 860   delete $reader{$jobstepid};
 861   delete $slot[$proc{$pid}->{slot}]->{pid};
 862   push @freeslot, $proc{$pid}->{slot};
 863   delete $proc{$pid};
 864
 865   # Load new tasks
 866   my $newtask_list = $arv->{'job_tasks'}->{'list'}->execute(
 867     'where' => {
 868       'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
 869     },
 870     'order' => 'qsequence'
 871   );
 872   foreach my $arvados_task (@{$newtask_list->{'items'}}) {
 873     my $jobstep = {
 874       'level' => $arvados_task->{'sequence'},
 875       'failures' => 0,
 876       'arvados_task' => $arvados_task
 877     };
 878     push @jobstep, $jobstep;
 879     push @jobstep_todo, $#jobstep;
 880   }
 881
 882   $progress_is_dirty = 1;
 883   1;
 884 }
 885
 886 sub check_refresh_wanted
 887 {
 888   my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
 889   if (@stat && $stat[9] > $latest_refresh) {
 890     $latest_refresh = scalar time;
 891     if ($job_has_uuid) {
 892       my $Job2 = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec);
 893       for my $attr ('cancelled_at',
 894                     'cancelled_by_user_uuid',
 895                     'cancelled_by_client_uuid') {
 896         $Job->{$attr} = $Job2->{$attr};
 897       }
 898       if ($Job->{'cancelled_at'}) {
 899         Log (undef, "Job cancelled at " . $Job->{cancelled_at} .
 900              " by user " . $Job->{cancelled_by_user_uuid});
 901         $main::success = 0;
 902         $main::please_freeze = 1;
 903       }
 904     }
 905   }
 906 }
 907
 908 sub check_squeue
 909 {
 910   # return if the kill list was checked <4 seconds ago
 911   if (defined $squeue_kill_checked && $squeue_kill_checked > time - 4)
 912   {
 913     return;
 914   }
 915   $squeue_kill_checked = time;
 916
 917   # use killem() on procs whose killtime is reached
 918   for (keys %proc)
 919   {
 920     if (exists $proc{$_}->{killtime}
 921         && $proc{$_}->{killtime} <= time)
 922     {
 923       killem ($_);
 924     }
 925   }
 926
 927   # return if the squeue was checked <60 seconds ago
 928   if (defined $squeue_checked && $squeue_checked > time - 60)
 929   {
 930     return;
 931   }
 932   $squeue_checked = time;
 933
 934   if (!$have_slurm)
 935   {
 936     # here is an opportunity to check for mysterious problems with local procs
 937     return;
 938   }
 939
 940   # get a list of steps still running
 941   my @squeue = `squeue -s -h -o '%i %j' && echo ok`;
 942   chop @squeue;
 943   if ($squeue[-1] ne "ok")
 944   {
 945     return;
 946   }
 947   pop @squeue;
 948
 949   # which of my jobsteps are running, according to squeue?
 950   my %ok;
 951   foreach (@squeue)
 952   {
 953     if (/^(\d+)\.(\d+) (\S+)/)
 954     {
 955       if ($1 eq $ENV{SLURM_JOBID})
 956       {
 957         $ok{$3} = 1;
 958       }
 959     }
 960   }
 961
 962   # which of my active child procs (>60s old) were not mentioned by squeue?
 963   foreach (keys %proc)
 964   {
 965     if ($proc{$_}->{time} < time - 60
 966         && !exists $ok{$proc{$_}->{jobstepname}}
 967         && !exists $proc{$_}->{killtime})
 968     {
 969       # kill this proc if it hasn't exited in 30 seconds
 970       $proc{$_}->{killtime} = time + 30;
 971     }
 972   }
 973 }
 974
 975
 976 sub release_allocation
 977 {
 978   if ($have_slurm)
 979   {
 980     Log (undef, "release job allocation");
 981     system "scancel $ENV{SLURM_JOBID}";
 982   }
 983 }
 984
 985
 986 sub readfrompipes
 987 {
 988   my $gotsome = 0;
 989   foreach my $job (keys %reader)
 990   {
 991     my $buf;
 992     while (0 < sysread ($reader{$job}, $buf, 8192))
 993     {
 994       print STDERR $buf if $ENV{CRUNCH_DEBUG};
 995       $jobstep[$job]->{stderr} .= $buf;
 996       preprocess_stderr ($job);
 997       if (length ($jobstep[$job]->{stderr}) > 16384)
 998       {
 999         substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
1000       }
1001       $gotsome = 1;
1002     }
1003   }
1004   return $gotsome;
1005 }
1006
1007
1008 sub preprocess_stderr
1009 {
1010   my $job = shift;
1011
1012   while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1013     my $line = $1;
1014     substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1015     Log ($job, "stderr $line");
1016     if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1017       # whoa.
1018       $main::please_freeze = 1;
1019     }
1020     elsif ($line =~ /srun: error: (Node failure on|Unable to create job step) /) {
1021       $jobstep[$job]->{node_fail} = 1;
1022       ban_node_by_slot($jobstep[$job]->{slotindex});
1023     }
1024   }
1025 }
1026
1027
1028 sub process_stderr
1029 {
1030   my $job = shift;
1031   my $task_success = shift;
1032   preprocess_stderr ($job);
1033
1034   map {
1035     Log ($job, "stderr $_");
1036   } split ("\n", $jobstep[$job]->{stderr});
1037 }
1038
1039 sub fetch_block
1040 {
1041   my $hash = shift;
1042   my ($child_out, $child_in, $output_block);
1043
1044   my $pid = open2($child_out, $child_in, 'arv', 'keep', 'get', $hash);
1045   sysread($child_out, $output_block, 64 * 1024 * 1024);
1046   waitpid($pid, 0);
1047   return $output_block;
1048 }
1049
1050 sub collate_output
1051 {
1052   Log (undef, "collate");
1053
1054   my ($child_out, $child_in);
1055   my $pid = open2($child_out, $child_in, 'arv', 'keep', 'put', '--raw');
1056   my $joboutput;
1057   for (@jobstep)
1058   {
1059     next if (!exists $_->{'arvados_task'}->{output} ||
1060              !$_->{'arvados_task'}->{'success'} ||
1061              $_->{'exitcode'} != 0);
1062     my $output = $_->{'arvados_task'}->{output};
1063     if ($output !~ /^[0-9a-f]{32}(\+\S+)*$/)
1064     {
1065       $output_in_keep ||= $output =~ / [0-9a-f]{32}\S*\+K/;
1066       print $child_in $output;
1067     }
1068     elsif (@jobstep == 1)
1069     {
1070       $joboutput = $output;
1071       last;
1072     }
1073     elsif (defined (my $outblock = fetch_block ($output)))
1074     {
1075       $output_in_keep ||= $outblock =~ / [0-9a-f]{32}\S*\+K/;
1076       print $child_in $outblock;
1077     }
1078     else
1079     {
1080       print $child_in "XXX fetch_block($output) failed XXX\n";
1081       $main::success = 0;
1082     }
1083   }
1084   if (!defined $joboutput) {
1085     my $s = IO::Select->new($child_out);
1086     sysread($child_out, $joboutput, 64 * 1024 * 1024) if $s->can_read(0);
1087   }
1088   $child_in->close;
1089   waitpid($pid, 0);
1090
1091   if ($joboutput)
1092   {
1093     Log (undef, "output $joboutput");
1094     $Job->update_attributes('output' => $joboutput) if $job_has_uuid;
1095   }
1096   else
1097   {
1098     Log (undef, "output undef");
1099   }
1100   return $joboutput;
1101 }
1102
1103
1104 sub killem
1105 {
1106   foreach (@_)
1107   {
1108     my $sig = 2;                # SIGINT first
1109     if (exists $proc{$_}->{"sent_$sig"} &&
1110         time - $proc{$_}->{"sent_$sig"} > 4)
1111     {
1112       $sig = 15;                # SIGTERM if SIGINT doesn't work
1113     }
1114     if (exists $proc{$_}->{"sent_$sig"} &&
1115         time - $proc{$_}->{"sent_$sig"} > 4)
1116     {
1117       $sig = 9;                 # SIGKILL if SIGTERM doesn't work
1118     }
1119     if (!exists $proc{$_}->{"sent_$sig"})
1120     {
1121       Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1122       kill $sig, $_;
1123       select (undef, undef, undef, 0.1);
1124       if ($sig == 2)
1125       {
1126         kill $sig, $_;     # srun wants two SIGINT to really interrupt
1127       }
1128       $proc{$_}->{"sent_$sig"} = time;
1129       $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1130     }
1131   }
1132 }
1133
1134
1135 sub fhbits
1136 {
1137   my($bits);
1138   for (@_) {
1139     vec($bits,fileno($_),1) = 1;
1140   }
1141   $bits;
1142 }
1143
1144
1145 sub Log                         # ($jobstep_id, $logmessage)
1146 {
1147   if ($_[1] =~ /\n/) {
1148     for my $line (split (/\n/, $_[1])) {
1149       Log ($_[0], $line);
1150     }
1151     return;
1152   }
1153   my $fh = select STDERR; $|=1; select $fh;
1154   my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1155   $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1156   $message .= "\n";
1157   my $datetime;
1158   if ($metastream || -t STDERR) {
1159     my @gmtime = gmtime;
1160     $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1161                          $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1162   }
1163   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1164
1165   # return if !$metastream;
1166   # $metastream->write_data ($datetime . " " . $message);
1167 }
1168
1169
1170 sub croak
1171 {
1172   my ($package, $file, $line) = caller;
1173   my $message = "@_ at $file line $line\n";
1174   Log (undef, $message);
1175   freeze() if @jobstep_todo;
1176   collate_output() if @jobstep_todo;
1177   cleanup();
1178   save_meta() if $metastream;
1179   die;
1180 }
1181
1182
1183 sub cleanup
1184 {
1185   return if !$job_has_uuid;
1186   $Job->update_attributes('running' => 0,
1187                           'success' => 0,
1188                           'finished_at' => scalar gmtime);
1189 }
1190
1191
1192 sub save_meta
1193 {
1194 #  my $justcheckpoint = shift; # false if this will be the last meta saved
1195 #  my $m = $metastream;
1196 #  $m = $m->copy if $justcheckpoint;
1197 #  $m->write_finish;
1198 #  my $whc = Warehouse->new;
1199 #  my $loglocator = $whc->store_block ($m->as_string);
1200 #  $arv->{'collections'}->{'create'}->execute('collection' => {
1201 #    'uuid' => $loglocator,
1202 #    'manifest_text' => $m->as_string,
1203 #  });
1204 #  undef $metastream if !$justcheckpoint; # otherwise Log() will try to use it
1205 #  Log (undef, "log manifest is $loglocator");
1206 #  $Job->{'log'} = $loglocator;
1207 #  $Job->update_attributes('log', $loglocator) if $job_has_uuid;
1208 }
1209
1210
1211 sub freeze_if_want_freeze
1212 {
1213   if ($main::please_freeze)
1214   {
1215     release_allocation();
1216     if (@_)
1217     {
1218       # kill some srun procs before freeze+stop
1219       map { $proc{$_} = {} } @_;
1220       while (%proc)
1221       {
1222         killem (keys %proc);
1223         select (undef, undef, undef, 0.1);
1224         my $died;
1225         while (($died = waitpid (-1, WNOHANG)) > 0)
1226         {
1227           delete $proc{$died};
1228         }
1229       }
1230     }
1231     freeze();
1232     collate_output();
1233     cleanup();
1234     save_meta();
1235     exit 0;
1236   }
1237 }
1238
1239
1240 sub freeze
1241 {
1242   Log (undef, "Freeze not implemented");
1243   return;
1244 }
1245
1246
1247 sub thaw
1248 {
1249   croak ("Thaw not implemented");
1250
1251   # my $whc;
1252   # my $key = shift;
1253   # Log (undef, "thaw from $key");
1254
1255   # @jobstep = ();
1256   # @jobstep_done = ();
1257   # @jobstep_todo = ();
1258   # @jobstep_tomerge = ();
1259   # $jobstep_tomerge_level = 0;
1260   # my $frozenjob = {};
1261
1262   # my $stream = new Warehouse::Stream ( whc => $whc,
1263   #                                    hash => [split (",", $key)] );
1264   # $stream->rewind;
1265   # while (my $dataref = $stream->read_until (undef, "\n\n"))
1266   # {
1267   #   if ($$dataref =~ /^job /)
1268   #   {
1269   #     foreach (split ("\n", $$dataref))
1270   #     {
1271   #     my ($k, $v) = split ("=", $_, 2);
1272   #     $frozenjob->{$k} = freezeunquote ($v);
1273   #     }
1274   #     next;
1275   #   }
1276
1277   #   if ($$dataref =~ /^merge (\d+) (.*)/)
1278   #   {
1279   #     $jobstep_tomerge_level = $1;
1280   #     @jobstep_tomerge
1281   #       = map { freezeunquote ($_) } split ("\n", freezeunquote($2));
1282   #     next;
1283   #   }
1284
1285   #   my $Jobstep = { };
1286   #   foreach (split ("\n", $$dataref))
1287   #   {
1288   #     my ($k, $v) = split ("=", $_, 2);
1289   #     $Jobstep->{$k} = freezeunquote ($v) if $k;
1290   #   }
1291   #   $Jobstep->{'failures'} = 0;
1292   #   push @jobstep, $Jobstep;
1293
1294   #   if ($Jobstep->{exitcode} eq "0")
1295   #   {
1296   #     push @jobstep_done, $#jobstep;
1297   #   }
1298   #   else
1299   #   {
1300   #     push @jobstep_todo, $#jobstep;
1301   #   }
1302   # }
1303
1304   # foreach (qw (script script_version script_parameters))
1305   # {
1306   #   $Job->{$_} = $frozenjob->{$_};
1307   # }
1308   # $Job->save if $job_has_uuid;
1309 }
1310
1311
1312 sub freezequote
1313 {
1314   my $s = shift;
1315   $s =~ s/\\/\\\\/g;
1316   $s =~ s/\n/\\n/g;
1317   return $s;
1318 }
1319
1320
1321 sub freezeunquote
1322 {
1323   my $s = shift;
1324   $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1325   return $s;
1326 }
1327
1328
1329 sub srun
1330 {
1331   my $srunargs = shift;
1332   my $execargs = shift;
1333   my $opts = shift || {};
1334   my $stdin = shift;
1335   my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1336   print STDERR (join (" ",
1337                       map { / / ? "'$_'" : $_ }
1338                       (@$args)),
1339                 "\n")
1340       if $ENV{CRUNCH_DEBUG};
1341
1342   if (defined $stdin) {
1343     my $child = open STDIN, "-|";
1344     defined $child or die "no fork: $!";
1345     if ($child == 0) {
1346       print $stdin or die $!;
1347       close STDOUT or die $!;
1348       exit 0;
1349     }
1350   }
1351
1352   return system (@$args) if $opts->{fork};
1353
1354   exec @$args;
1355   warn "ENV size is ".length(join(" ",%ENV));
1356   die "exec failed: $!: @$args";
1357 }
1358
1359
1360 sub ban_node_by_slot {
1361   # Don't start any new jobsteps on this node for 60 seconds
1362   my $slotid = shift;
1363   $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1364   $slot[$slotid]->{node}->{hold_count}++;
1365   Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1366 }
1367
1368 __DATA__
1369 #!/usr/bin/perl
1370
1371 # checkout-and-build
1372
1373 use Fcntl ':flock';
1374
1375 my $destdir = $ENV{"CRUNCH_SRC"};
1376 my $commit = $ENV{"CRUNCH_SRC_COMMIT"};
1377 my $repo = $ENV{"CRUNCH_SRC_URL"};
1378
1379 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
1380 flock L, LOCK_EX;
1381 if (readlink ("$destdir.commit") eq $commit && -d $destdir) {
1382     exit 0;
1383 }
1384
1385 unlink "$destdir.commit";
1386 open STDOUT, ">", "$destdir.log";
1387 open STDERR, ">&STDOUT";
1388
1389 mkdir $destdir;
1390 my @git_archive_data = <DATA>;
1391 if (@git_archive_data) {
1392   open TARX, "|-", "tar", "-C", $destdir, "-xf", "-";
1393   print TARX @git_archive_data;
1394   if(!close(TARX)) {
1395     die "'tar -C $destdir -xf -' exited $?: $!";
1396   }
1397 }
1398
1399 my $pwd;
1400 chomp ($pwd = `pwd`);
1401 my $install_dir = $ENV{"CRUNCH_INSTALL"} || "$pwd/opt";
1402 mkdir $install_dir;
1403
1404 for my $src_path ("$destdir/arvados/sdk/python") {
1405   if (-d $src_path) {
1406     shell_or_die ("virtualenv", $install_dir);
1407     shell_or_die ("cd $src_path && ./build.sh && $install_dir/bin/python setup.py install");
1408   }
1409 }
1410
1411 if (-e "$destdir/crunch_scripts/install") {
1412     shell_or_die ("$destdir/crunch_scripts/install", $install_dir);
1413 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
1414     # Old version
1415     shell_or_die ("./tests/autotests.sh", $install_dir);
1416 } elsif (-e "./install.sh") {
1417     shell_or_die ("./install.sh", $install_dir);
1418 }
1419
1420 if ($commit) {
1421     unlink "$destdir.commit.new";
1422     symlink ($commit, "$destdir.commit.new") or die "$destdir.commit.new: $!";
1423     rename ("$destdir.commit.new", "$destdir.commit") or die "$destdir.commit: $!";
1424 }
1425
1426 close L;
1427
1428 exit 0;
1429
1430 sub shell_or_die
1431 {
1432   if ($ENV{"DEBUG"}) {
1433     print STDERR "@_\n";
1434   }
1435   system (@_) == 0
1436       or die "@_ failed: $! exit 0x".sprintf("%x",$?);
1437 }
1438
1439 __DATA__