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