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     $command .=
 583         "&& exec arv-mount $ENV{TASK_KEEPMOUNT} --exec $ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
 584     my @execargs = ('bash', '-c', $command);
 585     srun (\@srunargs, \@execargs, undef, $build_script_to_send);
 586     exit (111);
 587   }
 588   close("writer");
 589   if (!defined $childpid)
 590   {
 591     close $reader{$id};
 592     delete $reader{$id};
 593     next;
 594   }
 595   shift @freeslot;
 596   $proc{$childpid} = { jobstep => $id,
 597                        time => time,
 598                        slot => $childslot,
 599                        jobstepname => "$job_id.$id.$childpid",
 600                      };
 601   croak ("assert failed: \$slot[$childslot]->{'pid'} exists") if exists $slot[$childslot]->{pid};
 602   $slot[$childslot]->{pid} = $childpid;
 603
 604   Log ($id, "job_task ".$Jobstep->{'arvados_task'}->{'uuid'});
 605   Log ($id, "child $childpid started on $childslotname");
 606   $Jobstep->{starttime} = time;
 607   $Jobstep->{node} = $childnode->{name};
 608   $Jobstep->{slotindex} = $childslot;
 609   delete $Jobstep->{stderr};
 610   delete $Jobstep->{finishtime};
 611
 612   splice @jobstep_todo, $todo_ptr, 1;
 613   --$todo_ptr;
 614
 615   $progress_is_dirty = 1;
 616
 617   while (!@freeslot
 618          ||
 619          (@slot > @freeslot && $todo_ptr+1 > $#jobstep_todo))
 620   {
 621     last THISROUND if $main::please_freeze;
 622     if ($main::please_info)
 623     {
 624       $main::please_info = 0;
 625       freeze();
 626       collate_output();
 627       save_meta(1);
 628       update_progress_stats();
 629     }
 630     my $gotsome
 631         = readfrompipes ()
 632         + reapchildren ();
 633     if (!$gotsome)
 634     {
 635       check_refresh_wanted();
 636       check_squeue();
 637       update_progress_stats();
 638       select (undef, undef, undef, 0.1);
 639     }
 640     elsif (time - $progress_stats_updated >= 30)
 641     {
 642       update_progress_stats();
 643     }
 644     if (($thisround_failed_multiple >= 8 && $thisround_succeeded == 0) ||
 645         ($thisround_failed_multiple >= 16 && $thisround_failed_multiple > $thisround_succeeded))
 646     {
 647       my $message = "Repeated failure rate too high ($thisround_failed_multiple/"
 648           .($thisround_failed+$thisround_succeeded)
 649           .") -- giving up on this round";
 650       Log (undef, $message);
 651       last THISROUND;
 652     }
 653
 654     # move slots from freeslot to holdslot (or back to freeslot) if necessary
 655     for (my $i=$#freeslot; $i>=0; $i--) {
 656       if ($slot[$freeslot[$i]]->{node}->{hold_until} > scalar time) {
 657         push @holdslot, (splice @freeslot, $i, 1);
 658       }
 659     }
 660     for (my $i=$#holdslot; $i>=0; $i--) {
 661       if ($slot[$holdslot[$i]]->{node}->{hold_until} <= scalar time) {
 662         push @freeslot, (splice @holdslot, $i, 1);
 663       }
 664     }
 665
 666     # give up if no nodes are succeeding
 667     if (!grep { $_->{node}->{losing_streak} == 0 &&
 668                     $_->{node}->{hold_count} < 4 } @slot) {
 669       my $message = "Every node has failed -- giving up on this round";
 670       Log (undef, $message);
 671       last THISROUND;
 672     }
 673   }
 674 }
 675
 676
 677 push @freeslot, splice @holdslot;
 678 map { $slot[$freeslot[$_]]->{node}->{losing_streak} = 0 } (0..$#freeslot);
 679
 680
 681 Log (undef, "wait for last ".(scalar keys %proc)." children to finish");
 682 while (%proc)
 683 {
 684   if ($main::please_continue) {
 685     $main::please_continue = 0;
 686     goto THISROUND;
 687   }
 688   $main::please_info = 0, freeze(), collate_output(), save_meta(1) if $main::please_info;
 689   readfrompipes ();
 690   if (!reapchildren())
 691   {
 692     check_refresh_wanted();
 693     check_squeue();
 694     update_progress_stats();
 695     select (undef, undef, undef, 0.1);
 696     killem (keys %proc) if $main::please_freeze;
 697   }
 698 }
 699
 700 update_progress_stats();
 701 freeze_if_want_freeze();
 702
 703
 704 if (!defined $main::success)
 705 {
 706   if (@jobstep_todo &&
 707       $thisround_succeeded == 0 &&
 708       ($thisround_failed == 0 || $thisround_failed > 4))
 709   {
 710     my $message = "stop because $thisround_failed tasks failed and none succeeded";
 711     Log (undef, $message);
 712     $main::success = 0;
 713   }
 714   if (!@jobstep_todo)
 715   {
 716     $main::success = 1;
 717   }
 718 }
 719
 720 goto ONELEVEL if !defined $main::success;
 721
 722
 723 release_allocation();
 724 freeze();
 725 if ($job_has_uuid) {
 726   $Job->update_attributes('output' => &collate_output(),
 727                           'running' => 0,
 728                           'success' => $Job->{'output'} && $main::success,
 729                           'finished_at' => scalar gmtime)
 730 }
 731
 732 if ($Job->{'output'})
 733 {
 734   eval {
 735     my $manifest_text = `arv keep get $Job->{'output'}`;
 736     $arv->{'collections'}->{'create'}->execute('collection' => {
 737       'uuid' => $Job->{'output'},
 738       'manifest_text' => $manifest_text,
 739     });
 740   };
 741   if ($@) {
 742     Log (undef, "Failed to register output manifest: $@");
 743   }
 744 }
 745
 746 Log (undef, "finish");
 747
 748 save_meta();
 749 exit 0;
 750
 751
 752
 753 sub update_progress_stats
 754 {
 755   $progress_stats_updated = time;
 756   return if !$progress_is_dirty;
 757   my ($todo, $done, $running) = (scalar @jobstep_todo,
 758                                  scalar @jobstep_done,
 759                                  scalar @slot - scalar @freeslot - scalar @holdslot);
 760   $Job->{'tasks_summary'} ||= {};
 761   $Job->{'tasks_summary'}->{'todo'} = $todo;
 762   $Job->{'tasks_summary'}->{'done'} = $done;
 763   $Job->{'tasks_summary'}->{'running'} = $running;
 764   if ($job_has_uuid) {
 765     $Job->update_attributes('tasks_summary' => $Job->{'tasks_summary'});
 766   }
 767   Log (undef, "status: $done done, $running running, $todo todo");
 768   $progress_is_dirty = 0;
 769 }
 770
 771
 772
 773 sub reapchildren
 774 {
 775   my $pid = waitpid (-1, WNOHANG);
 776   return 0 if $pid <= 0;
 777
 778   my $whatslot = ($slot[$proc{$pid}->{slot}]->{node}->{name}
 779                   . "."
 780                   . $slot[$proc{$pid}->{slot}]->{cpu});
 781   my $jobstepid = $proc{$pid}->{jobstep};
 782   my $elapsed = time - $proc{$pid}->{time};
 783   my $Jobstep = $jobstep[$jobstepid];
 784
 785   my $childstatus = $?;
 786   my $exitvalue = $childstatus >> 8;
 787   my $exitinfo = sprintf("exit %d signal %d%s",
 788                          $exitvalue,
 789                          $childstatus & 127,
 790                          ($childstatus & 128 ? ' core dump' : ''));
 791   $Jobstep->{'arvados_task'}->reload;
 792   my $task_success = $Jobstep->{'arvados_task'}->{success};
 793
 794   Log ($jobstepid, "child $pid on $whatslot $exitinfo success=$task_success");
 795
 796   if (!defined $task_success) {
 797     # task did not indicate one way or the other --> fail
 798     $Jobstep->{'arvados_task'}->{success} = 0;
 799     $Jobstep->{'arvados_task'}->save;
 800     $task_success = 0;
 801   }
 802
 803   if (!$task_success)
 804   {
 805     my $temporary_fail;
 806     $temporary_fail ||= $Jobstep->{node_fail};
 807     $temporary_fail ||= ($exitvalue == 111);
 808
 809     ++$thisround_failed;
 810     ++$thisround_failed_multiple if $Jobstep->{'failures'} >= 1;
 811
 812     # Check for signs of a failed or misconfigured node
 813     if (++$slot[$proc{$pid}->{slot}]->{node}->{losing_streak} >=
 814         2+$slot[$proc{$pid}->{slot}]->{node}->{ncpus}) {
 815       # Don't count this against jobstep failure thresholds if this
 816       # node is already suspected faulty and srun exited quickly
 817       if ($slot[$proc{$pid}->{slot}]->{node}->{hold_until} &&
 818           $elapsed < 5) {
 819         Log ($jobstepid, "blaming failure on suspect node " .
 820              $slot[$proc{$pid}->{slot}]->{node}->{name});
 821         $temporary_fail ||= 1;
 822       }
 823       ban_node_by_slot($proc{$pid}->{slot});
 824     }
 825
 826     Log ($jobstepid, sprintf('failure (#%d, %s) after %d seconds',
 827                              ++$Jobstep->{'failures'},
 828                              $temporary_fail ? 'temporary ' : 'permanent',
 829                              $elapsed));
 830
 831     if (!$temporary_fail || $Jobstep->{'failures'} >= 3) {
 832       # Give up on this task, and the whole job
 833       $main::success = 0;
 834       $main::please_freeze = 1;
 835     }
 836     else {
 837       # Put this task back on the todo queue
 838       push @jobstep_todo, $jobstepid;
 839     }
 840     $Job->{'tasks_summary'}->{'failed'}++;
 841   }
 842   else
 843   {
 844     ++$thisround_succeeded;
 845     $slot[$proc{$pid}->{slot}]->{node}->{losing_streak} = 0;
 846     $slot[$proc{$pid}->{slot}]->{node}->{hold_until} = 0;
 847     push @jobstep_done, $jobstepid;
 848     Log ($jobstepid, "success in $elapsed seconds");
 849   }
 850   $Jobstep->{exitcode} = $childstatus;
 851   $Jobstep->{finishtime} = time;
 852   process_stderr ($jobstepid, $task_success);
 853   Log ($jobstepid, "output " . $Jobstep->{'arvados_task'}->{output});
 854
 855   close $reader{$jobstepid};
 856   delete $reader{$jobstepid};
 857   delete $slot[$proc{$pid}->{slot}]->{pid};
 858   push @freeslot, $proc{$pid}->{slot};
 859   delete $proc{$pid};
 860
 861   # Load new tasks
 862   my $newtask_list = $arv->{'job_tasks'}->{'list'}->execute(
 863     'where' => {
 864       'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
 865     },
 866     'order' => 'qsequence'
 867   );
 868   foreach my $arvados_task (@{$newtask_list->{'items'}}) {
 869     my $jobstep = {
 870       'level' => $arvados_task->{'sequence'},
 871       'failures' => 0,
 872       'arvados_task' => $arvados_task
 873     };
 874     push @jobstep, $jobstep;
 875     push @jobstep_todo, $#jobstep;
 876   }
 877
 878   $progress_is_dirty = 1;
 879   1;
 880 }
 881
 882 sub check_refresh_wanted
 883 {
 884   my @stat = stat $ENV{"CRUNCH_REFRESH_TRIGGER"};
 885   if (@stat && $stat[9] > $latest_refresh) {
 886     $latest_refresh = scalar time;
 887     if ($job_has_uuid) {
 888       my $Job2 = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec);
 889       for my $attr ('cancelled_at',
 890                     'cancelled_by_user_uuid',
 891                     'cancelled_by_client_uuid') {
 892         $Job->{$attr} = $Job2->{$attr};
 893       }
 894       if ($Job->{'cancelled_at'}) {
 895         Log (undef, "Job cancelled at " . $Job->{cancelled_at} .
 896              " by user " . $Job->{cancelled_by_user_uuid});
 897         $main::success = 0;
 898         $main::please_freeze = 1;
 899       }
 900     }
 901   }
 902 }
 903
 904 sub check_squeue
 905 {
 906   # return if the kill list was checked <4 seconds ago
 907   if (defined $squeue_kill_checked && $squeue_kill_checked > time - 4)
 908   {
 909     return;
 910   }
 911   $squeue_kill_checked = time;
 912
 913   # use killem() on procs whose killtime is reached
 914   for (keys %proc)
 915   {
 916     if (exists $proc{$_}->{killtime}
 917         && $proc{$_}->{killtime} <= time)
 918     {
 919       killem ($_);
 920     }
 921   }
 922
 923   # return if the squeue was checked <60 seconds ago
 924   if (defined $squeue_checked && $squeue_checked > time - 60)
 925   {
 926     return;
 927   }
 928   $squeue_checked = time;
 929
 930   if (!$have_slurm)
 931   {
 932     # here is an opportunity to check for mysterious problems with local procs
 933     return;
 934   }
 935
 936   # get a list of steps still running
 937   my @squeue = `squeue -s -h -o '%i %j' && echo ok`;
 938   chop @squeue;
 939   if ($squeue[-1] ne "ok")
 940   {
 941     return;
 942   }
 943   pop @squeue;
 944
 945   # which of my jobsteps are running, according to squeue?
 946   my %ok;
 947   foreach (@squeue)
 948   {
 949     if (/^(\d+)\.(\d+) (\S+)/)
 950     {
 951       if ($1 eq $ENV{SLURM_JOBID})
 952       {
 953         $ok{$3} = 1;
 954       }
 955     }
 956   }
 957
 958   # which of my active child procs (>60s old) were not mentioned by squeue?
 959   foreach (keys %proc)
 960   {
 961     if ($proc{$_}->{time} < time - 60
 962         && !exists $ok{$proc{$_}->{jobstepname}}
 963         && !exists $proc{$_}->{killtime})
 964     {
 965       # kill this proc if it hasn't exited in 30 seconds
 966       $proc{$_}->{killtime} = time + 30;
 967     }
 968   }
 969 }
 970
 971
 972 sub release_allocation
 973 {
 974   if ($have_slurm)
 975   {
 976     Log (undef, "release job allocation");
 977     system "scancel $ENV{SLURM_JOBID}";
 978   }
 979 }
 980
 981
 982 sub readfrompipes
 983 {
 984   my $gotsome = 0;
 985   foreach my $job (keys %reader)
 986   {
 987     my $buf;
 988     while (0 < sysread ($reader{$job}, $buf, 8192))
 989     {
 990       print STDERR $buf if $ENV{CRUNCH_DEBUG};
 991       $jobstep[$job]->{stderr} .= $buf;
 992       preprocess_stderr ($job);
 993       if (length ($jobstep[$job]->{stderr}) > 16384)
 994       {
 995         substr ($jobstep[$job]->{stderr}, 0, 8192) = "";
 996       }
 997       $gotsome = 1;
 998     }
 999   }
1000   return $gotsome;
1001 }
1002
1003
1004 sub preprocess_stderr
1005 {
1006   my $job = shift;
1007
1008   while ($jobstep[$job]->{stderr} =~ /^(.*?)\n/) {
1009     my $line = $1;
1010     substr $jobstep[$job]->{stderr}, 0, 1+length($line), "";
1011     Log ($job, "stderr $line");
1012     if ($line =~ /srun: error: (SLURM job $ENV{SLURM_JOB_ID} has expired|Unable to confirm allocation for job $ENV{SLURM_JOB_ID})/) {
1013       # whoa.
1014       $main::please_freeze = 1;
1015     }
1016     elsif ($line =~ /srun: error: (Node failure on|Unable to create job step) /) {
1017       $jobstep[$job]->{node_fail} = 1;
1018       ban_node_by_slot($jobstep[$job]->{slotindex});
1019     }
1020   }
1021 }
1022
1023
1024 sub process_stderr
1025 {
1026   my $job = shift;
1027   my $task_success = shift;
1028   preprocess_stderr ($job);
1029
1030   map {
1031     Log ($job, "stderr $_");
1032   } split ("\n", $jobstep[$job]->{stderr});
1033 }
1034
1035 sub fetch_block
1036 {
1037   my $hash = shift;
1038   my ($child_out, $child_in, $output_block);
1039
1040   my $pid = open2($child_out, $child_in, 'arv', 'keep', 'get', $hash);
1041   sysread($child_out, $output_block, 64 * 1024 * 1024);
1042   waitpid($pid, 0);
1043   return $output_block;
1044 }
1045
1046 sub collate_output
1047 {
1048   Log (undef, "collate");
1049
1050   my ($child_out, $child_in);
1051   my $pid = open2($child_out, $child_in, 'arv', 'keep', 'put', '--raw');
1052   my $joboutput;
1053   for (@jobstep)
1054   {
1055     next if (!exists $_->{'arvados_task'}->{output} ||
1056              !$_->{'arvados_task'}->{'success'} ||
1057              $_->{'exitcode'} != 0);
1058     my $output = $_->{'arvados_task'}->{output};
1059     if ($output !~ /^[0-9a-f]{32}(\+\S+)*$/)
1060     {
1061       $output_in_keep ||= $output =~ / [0-9a-f]{32}\S*\+K/;
1062       print $child_in $output;
1063     }
1064     elsif (@jobstep == 1)
1065     {
1066       $joboutput = $output;
1067       last;
1068     }
1069     elsif (defined (my $outblock = fetch_block ($output)))
1070     {
1071       $output_in_keep ||= $outblock =~ / [0-9a-f]{32}\S*\+K/;
1072       print $child_in $outblock;
1073     }
1074     else
1075     {
1076       print $child_in "XXX fetch_block($output) failed XXX\n";
1077       $main::success = 0;
1078     }
1079   }
1080   if (!defined $joboutput) {
1081     my $s = IO::Select->new($child_out);
1082     sysread($child_out, $joboutput, 64 * 1024 * 1024) if $s->can_read(0);
1083   }
1084   $child_in->close;
1085   waitpid($pid, 0);
1086
1087   if ($joboutput)
1088   {
1089     Log (undef, "output $joboutput");
1090     $Job->update_attributes('output' => $joboutput) if $job_has_uuid;
1091   }
1092   else
1093   {
1094     Log (undef, "output undef");
1095   }
1096   return $joboutput;
1097 }
1098
1099
1100 sub killem
1101 {
1102   foreach (@_)
1103   {
1104     my $sig = 2;                # SIGINT first
1105     if (exists $proc{$_}->{"sent_$sig"} &&
1106         time - $proc{$_}->{"sent_$sig"} > 4)
1107     {
1108       $sig = 15;                # SIGTERM if SIGINT doesn't work
1109     }
1110     if (exists $proc{$_}->{"sent_$sig"} &&
1111         time - $proc{$_}->{"sent_$sig"} > 4)
1112     {
1113       $sig = 9;                 # SIGKILL if SIGTERM doesn't work
1114     }
1115     if (!exists $proc{$_}->{"sent_$sig"})
1116     {
1117       Log ($proc{$_}->{jobstep}, "sending 2x signal $sig to pid $_");
1118       kill $sig, $_;
1119       select (undef, undef, undef, 0.1);
1120       if ($sig == 2)
1121       {
1122         kill $sig, $_;     # srun wants two SIGINT to really interrupt
1123       }
1124       $proc{$_}->{"sent_$sig"} = time;
1125       $proc{$_}->{"killedafter"} = time - $proc{$_}->{"time"};
1126     }
1127   }
1128 }
1129
1130
1131 sub fhbits
1132 {
1133   my($bits);
1134   for (@_) {
1135     vec($bits,fileno($_),1) = 1;
1136   }
1137   $bits;
1138 }
1139
1140
1141 sub Log                         # ($jobstep_id, $logmessage)
1142 {
1143   if ($_[1] =~ /\n/) {
1144     for my $line (split (/\n/, $_[1])) {
1145       Log ($_[0], $line);
1146     }
1147     return;
1148   }
1149   my $fh = select STDERR; $|=1; select $fh;
1150   my $message = sprintf ("%s %d %s %s", $job_id, $$, @_);
1151   $message =~ s{([^ -\176])}{"\\" . sprintf ("%03o", ord($1))}ge;
1152   $message .= "\n";
1153   my $datetime;
1154   if ($metastream || -t STDERR) {
1155     my @gmtime = gmtime;
1156     $datetime = sprintf ("%04d-%02d-%02d_%02d:%02d:%02d",
1157                          $gmtime[5]+1900, $gmtime[4]+1, @gmtime[3,2,1,0]);
1158   }
1159   print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
1160
1161   # return if !$metastream;
1162   # $metastream->write_data ($datetime . " " . $message);
1163 }
1164
1165
1166 sub croak
1167 {
1168   my ($package, $file, $line) = caller;
1169   my $message = "@_ at $file line $line\n";
1170   Log (undef, $message);
1171   freeze() if @jobstep_todo;
1172   collate_output() if @jobstep_todo;
1173   cleanup();
1174   save_meta() if $metastream;
1175   die;
1176 }
1177
1178
1179 sub cleanup
1180 {
1181   return if !$job_has_uuid;
1182   $Job->update_attributes('running' => 0,
1183                           'success' => 0,
1184                           'finished_at' => scalar gmtime);
1185 }
1186
1187
1188 sub save_meta
1189 {
1190 #  my $justcheckpoint = shift; # false if this will be the last meta saved
1191 #  my $m = $metastream;
1192 #  $m = $m->copy if $justcheckpoint;
1193 #  $m->write_finish;
1194 #  my $whc = Warehouse->new;
1195 #  my $loglocator = $whc->store_block ($m->as_string);
1196 #  $arv->{'collections'}->{'create'}->execute('collection' => {
1197 #    'uuid' => $loglocator,
1198 #    'manifest_text' => $m->as_string,
1199 #  });
1200 #  undef $metastream if !$justcheckpoint; # otherwise Log() will try to use it
1201 #  Log (undef, "log manifest is $loglocator");
1202 #  $Job->{'log'} = $loglocator;
1203 #  $Job->update_attributes('log', $loglocator) if $job_has_uuid;
1204 }
1205
1206
1207 sub freeze_if_want_freeze
1208 {
1209   if ($main::please_freeze)
1210   {
1211     release_allocation();
1212     if (@_)
1213     {
1214       # kill some srun procs before freeze+stop
1215       map { $proc{$_} = {} } @_;
1216       while (%proc)
1217       {
1218         killem (keys %proc);
1219         select (undef, undef, undef, 0.1);
1220         my $died;
1221         while (($died = waitpid (-1, WNOHANG)) > 0)
1222         {
1223           delete $proc{$died};
1224         }
1225       }
1226     }
1227     freeze();
1228     collate_output();
1229     cleanup();
1230     save_meta();
1231     exit 0;
1232   }
1233 }
1234
1235
1236 sub freeze
1237 {
1238   Log (undef, "Freeze not implemented");
1239   return;
1240 }
1241
1242
1243 sub thaw
1244 {
1245   croak ("Thaw not implemented");
1246
1247   # my $whc;
1248   # my $key = shift;
1249   # Log (undef, "thaw from $key");
1250
1251   # @jobstep = ();
1252   # @jobstep_done = ();
1253   # @jobstep_todo = ();
1254   # @jobstep_tomerge = ();
1255   # $jobstep_tomerge_level = 0;
1256   # my $frozenjob = {};
1257
1258   # my $stream = new Warehouse::Stream ( whc => $whc,
1259   #                                    hash => [split (",", $key)] );
1260   # $stream->rewind;
1261   # while (my $dataref = $stream->read_until (undef, "\n\n"))
1262   # {
1263   #   if ($$dataref =~ /^job /)
1264   #   {
1265   #     foreach (split ("\n", $$dataref))
1266   #     {
1267   #     my ($k, $v) = split ("=", $_, 2);
1268   #     $frozenjob->{$k} = freezeunquote ($v);
1269   #     }
1270   #     next;
1271   #   }
1272
1273   #   if ($$dataref =~ /^merge (\d+) (.*)/)
1274   #   {
1275   #     $jobstep_tomerge_level = $1;
1276   #     @jobstep_tomerge
1277   #       = map { freezeunquote ($_) } split ("\n", freezeunquote($2));
1278   #     next;
1279   #   }
1280
1281   #   my $Jobstep = { };
1282   #   foreach (split ("\n", $$dataref))
1283   #   {
1284   #     my ($k, $v) = split ("=", $_, 2);
1285   #     $Jobstep->{$k} = freezeunquote ($v) if $k;
1286   #   }
1287   #   $Jobstep->{'failures'} = 0;
1288   #   push @jobstep, $Jobstep;
1289
1290   #   if ($Jobstep->{exitcode} eq "0")
1291   #   {
1292   #     push @jobstep_done, $#jobstep;
1293   #   }
1294   #   else
1295   #   {
1296   #     push @jobstep_todo, $#jobstep;
1297   #   }
1298   # }
1299
1300   # foreach (qw (script script_version script_parameters))
1301   # {
1302   #   $Job->{$_} = $frozenjob->{$_};
1303   # }
1304   # $Job->save if $job_has_uuid;
1305 }
1306
1307
1308 sub freezequote
1309 {
1310   my $s = shift;
1311   $s =~ s/\\/\\\\/g;
1312   $s =~ s/\n/\\n/g;
1313   return $s;
1314 }
1315
1316
1317 sub freezeunquote
1318 {
1319   my $s = shift;
1320   $s =~ s{\\(.)}{$1 eq "n" ? "\n" : $1}ge;
1321   return $s;
1322 }
1323
1324
1325 sub srun
1326 {
1327   my $srunargs = shift;
1328   my $execargs = shift;
1329   my $opts = shift || {};
1330   my $stdin = shift;
1331   my $args = $have_slurm ? [@$srunargs, @$execargs] : $execargs;
1332   print STDERR (join (" ",
1333                       map { / / ? "'$_'" : $_ }
1334                       (@$args)),
1335                 "\n")
1336       if $ENV{CRUNCH_DEBUG};
1337
1338   if (defined $stdin) {
1339     my $child = open STDIN, "-|";
1340     defined $child or die "no fork: $!";
1341     if ($child == 0) {
1342       print $stdin or die $!;
1343       close STDOUT or die $!;
1344       exit 0;
1345     }
1346   }
1347
1348   return system (@$args) if $opts->{fork};
1349
1350   exec @$args;
1351   warn "ENV size is ".length(join(" ",%ENV));
1352   die "exec failed: $!: @$args";
1353 }
1354
1355
1356 sub ban_node_by_slot {
1357   # Don't start any new jobsteps on this node for 60 seconds
1358   my $slotid = shift;
1359   $slot[$slotid]->{node}->{hold_until} = 60 + scalar time;
1360   $slot[$slotid]->{node}->{hold_count}++;
1361   Log (undef, "backing off node " . $slot[$slotid]->{node}->{name} . " for 60 seconds");
1362 }
1363
1364 __DATA__
1365 #!/usr/bin/perl
1366
1367 # checkout-and-build
1368
1369 use Fcntl ':flock';
1370
1371 my $destdir = $ENV{"CRUNCH_SRC"};
1372 my $commit = $ENV{"CRUNCH_SRC_COMMIT"};
1373 my $repo = $ENV{"CRUNCH_SRC_URL"};
1374
1375 open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
1376 flock L, LOCK_EX;
1377 if (readlink ("$destdir.commit") eq $commit && -d $destdir) {
1378     exit 0;
1379 }
1380
1381 unlink "$destdir.commit";
1382 open STDOUT, ">", "$destdir.log";
1383 open STDERR, ">&STDOUT";
1384
1385 mkdir $destdir;
1386 my @git_archive_data = <DATA>;
1387 if (@git_archive_data) {
1388   open TARX, "|-", "tar", "-C", $destdir, "-xf", "-";
1389   print TARX @git_archive_data;
1390   if(!close(TARX)) {
1391     die "'tar -C $destdir -xf -' exited $?: $!";
1392   }
1393 }
1394
1395 my $pwd;
1396 chomp ($pwd = `pwd`);
1397 my $install_dir = $ENV{"CRUNCH_INSTALL"} || "$pwd/opt";
1398 mkdir $install_dir;
1399
1400 for my $src_path ("$destdir/arvados/sdk/python") {
1401   if (-d $src_path) {
1402     shell_or_die ("virtualenv", $install_dir);
1403     shell_or_die ("cd $src_path && ./build.sh && $install_dir/bin/python setup.py install");
1404   }
1405 }
1406
1407 if (-e "$destdir/crunch_scripts/install") {
1408     shell_or_die ("$destdir/crunch_scripts/install", $install_dir);
1409 } elsif (!-e "./install.sh" && -e "./tests/autotests.sh") {
1410     # Old version
1411     shell_or_die ("./tests/autotests.sh", $install_dir);
1412 } elsif (-e "./install.sh") {
1413     shell_or_die ("./install.sh", $install_dir);
1414 }
1415
1416 if ($commit) {
1417     unlink "$destdir.commit.new";
1418     symlink ($commit, "$destdir.commit.new") or die "$destdir.commit.new: $!";
1419     rename ("$destdir.commit.new", "$destdir.commit") or die "$destdir.commit: $!";
1420 }
1421
1422 close L;
1423
1424 exit 0;
1425
1426 sub shell_or_die
1427 {
1428   if ($ENV{"DEBUG"}) {
1429     print STDERR "@_\n";
1430   }
1431   system (@_) == 0
1432       or die "@_ failed: $! exit 0x".sprintf("%x",$?);
1433 }
1434
1435 __DATA__