use IO::Select;
use File::Temp;
use Fcntl ':flock';
+use File::Path qw( make_path );
$ENV{"TMPDIR"} ||= "/tmp";
unless (defined $ENV{"CRUNCH_TMP"}) {
$ENV{"CRUNCH_TMP"} .= "-$<";
}
}
+
+# Create the tmp directory if it does not exist
+if ( ! -d $ENV{"CRUNCH_TMP"} ) {
+ make_path $ENV{"CRUNCH_TMP"} or die "Failed to create temporary working directory: " . $ENV{"CRUNCH_TMP"};
+}
+
$ENV{"JOB_WORK"} = $ENV{"CRUNCH_TMP"} . "/work";
$ENV{"CRUNCH_INSTALL"} = "$ENV{CRUNCH_TMP}/opt";
$ENV{"CRUNCH_WORK"} = $ENV{"JOB_WORK"}; # deprecated
mkdir ($ENV{"JOB_WORK"});
-my $arv_cli;
-
-if (defined $ENV{"ARV_CLI"}) {
- $arv_cli = $ENV{"ARV_CLI"};
-}
-else {
- $arv_cli = 'arv';
-}
-
my $force_unlock;
my $git_dir;
my $jobspec;
# If this job requires a Docker image, install that.
my $docker_bin = "/usr/bin/docker.io";
-my $docker_image = $Job->{runtime_constraints}->{docker_image} || "";
-if ($docker_image) {
+my ($docker_locator, $docker_hash);
+if ($docker_locator = $Job->{docker_image_locator}) {
+ $docker_hash = find_docker_hash($docker_locator);
+ if (!$docker_hash)
+ {
+ croak("No Docker image hash found from locator $docker_locator");
+ }
+ my $docker_install_script = qq{
+if ! $docker_bin images -q --no-trunc | grep -qxF \Q$docker_hash\E; then
+ arv-get \Q$docker_locator/$docker_hash.tar\E | $docker_bin load
+fi
+};
my $docker_pid = fork();
if ($docker_pid == 0)
{
- srun (["srun", "--nodelist=" . join(' ', @node)],
- [$docker_bin, 'pull', $docker_image]);
+ srun (["srun", "--nodelist=" . join(',', @node)],
+ ["/bin/sh", "-ec", $docker_install_script]);
exit ($?);
}
while (1)
freeze_if_want_freeze ($docker_pid);
select (undef, undef, undef, 0.1);
}
- # If the Docker image was specified as a hash, pull will fail.
- # Ignore that error. We'll see what happens when we try to run later.
- if (($? != 0) && ($docker_image !~ /^[0-9a-fA-F]{5,64}$/))
+ if ($? != 0)
{
- croak("Installing Docker image $docker_image returned exit code $?");
+ croak("Installing Docker image from $docker_locator returned exit code $?");
}
}
$ENV{"TASK_SLOT_NODE"} = $slot[$childslot]->{node}->{name};
$ENV{"TASK_SLOT_NUMBER"} = $slot[$childslot]->{cpu};
$ENV{"TASK_WORK"} = $ENV{"JOB_WORK"}."/$id.$$";
+ $ENV{"HOME"} = $ENV{"TASK_WORK"};
$ENV{"TASK_KEEPMOUNT"} = $ENV{"TASK_WORK"}.".keep";
$ENV{"TASK_TMPDIR"} = $ENV{"TASK_WORK"}; # deprecated
$ENV{"CRUNCH_NODE_SLOTS"} = $slot[$childslot]->{node}->{ncpus};
my $build_script_to_send = "";
my $command =
"if [ -e $ENV{TASK_WORK} ]; then rm -rf $ENV{TASK_WORK}; fi; "
- ."mkdir -p $ENV{JOB_WORK} $ENV{CRUNCH_TMP} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT} "
+ ."mkdir -p $ENV{JOB_WORK} $ENV{CRUNCH_TMP} $ENV{TASK_WORK} $ENV{TASK_KEEPMOUNT}"
."&& cd $ENV{CRUNCH_TMP} ";
if ($build_script)
{
"&& perl -";
}
$command .= "&& exec arv-mount --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
- if ($docker_image)
+ if ($docker_hash)
{
$command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$ENV{TASK_WORK}/docker.cid -poll=10000 ";
- $command .= "$docker_bin run -i -a stdin -a stdout -a stderr --cidfile=$ENV{TASK_WORK}/docker.cid ";
+ $command .= "$docker_bin run --rm=true --attach=stdout --attach=stderr --user=crunch --cidfile=$ENV{TASK_WORK}/docker.cid ";
# Dynamically configure the container to use the host system as its
# DNS server. Get the host's global addresses from the ip command,
# and turn them into docker --dns options using gawk.
$command .=
q{$(ip -o address show scope global |
gawk 'match($4, /^([0-9\.:]+)\//, x){print "--dns", x[1]}') };
- foreach my $env_key (qw(CRUNCH_SRC CRUNCH_TMP TASK_KEEPMOUNT))
- {
- $command .= "-v \Q$ENV{$env_key}:$ENV{$env_key}:rw\E ";
- }
+ $command .= "--volume=\Q$ENV{CRUNCH_SRC}:/tmp/crunch-src:ro\E ";
+ $command .= "--volume=\Q$ENV{TASK_KEEPMOUNT}:/keep:ro\E ";
+ $command .= "--env=\QHOME=/home/crunch\E ";
while (my ($env_key, $env_val) = each %ENV)
{
- if ($env_key =~ /^(JOB|TASK)_/) {
- $command .= "-e \Q$env_key=$env_val\E ";
+ if ($env_key =~ /^(ARVADOS|JOB|TASK)_/) {
+ if ($env_key eq "TASK_WORK") {
+ $command .= "--env=\QTASK_WORK=/tmp/crunch-job\E ";
+ }
+ elsif ($env_key eq "TASK_KEEPMOUNT") {
+ $command .= "--env=\QTASK_KEEPMOUNT=/keep\E ";
+ }
+ else {
+ $command .= "--env=\Q$env_key=$env_val\E ";
+ }
}
}
- $command .= "\Q$docker_image\E ";
+ $command .= "--env=\QCRUNCH_NODE_SLOTS=$ENV{CRUNCH_NODE_SLOTS}\E ";
+ $command .= "--env=\QCRUNCH_SRC=/tmp/crunch-src\E ";
+ $command .= "\Q$docker_hash\E ";
+ $command .= "stdbuf --output=0 --error=0 ";
+ $command .= "/tmp/crunch-src/crunch_scripts/" . $Job->{"script"};
} else {
- $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 "
+ # Non-docker run
+ $command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 ";
+ $command .= "stdbuf --output=0 --error=0 ";
+ $command .= "$ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
}
- $command .= "stdbuf -o0 -e0 ";
- $command .= "$ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
+
my @execargs = ('bash', '-c', $command);
srun (\@srunargs, \@execargs, undef, $build_script_to_send);
exit (111);
if ($collated_output)
{
eval {
- open(my $orig_manifest, '-|', 'arv', 'keep', 'get', $collated_output)
+ open(my $orig_manifest, '-|', 'arv-get', $collated_output)
or die "failed to get collated manifest: $!";
# Read the original manifest, and strip permission hints from it,
# so we can put the result in a Collection.
- my @manifest_lines = ();
+ my @stripped_manifest_lines = ();
+ my $orig_manifest_text = '';
while (my $manifest_line = <$orig_manifest>) {
+ $orig_manifest_text .= $manifest_line;
my @words = split(/ /, $manifest_line, -1);
foreach my $ii (0..$#words) {
if ($words[$ii] =~ /^[0-9a-f]{32}\+/) {
$words[$ii] =~ s/\+A[0-9a-f]{40}@[0-9a-f]{8}\b//;
}
}
- push(@manifest_lines, join(" ", @words));
+ push(@stripped_manifest_lines, join(" ", @words));
}
- my $manifest_text = join("", @manifest_lines);
+ my $stripped_manifest_text = join("", @stripped_manifest_lines);
my $output = $arv->{'collections'}->{'create'}->execute('collection' => {
- 'uuid' => md5_hex($manifest_text),
- 'manifest_text' => $manifest_text,
+ 'uuid' => md5_hex($stripped_manifest_text),
+ 'manifest_text' => $orig_manifest_text,
});
$Job->update_attributes('output' => $output->{uuid});
if ($Job->{'output_is_persistent'}) {
push @freeslot, $proc{$pid}->{slot};
delete $proc{$pid};
- # Load new tasks
- my $newtask_list = [];
- my $newtask_results;
- do {
- $newtask_results = $arv->{'job_tasks'}->{'list'}->execute(
- 'where' => {
- 'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
- },
- 'order' => 'qsequence',
- 'offset' => scalar(@$newtask_list),
- );
- push(@$newtask_list, @{$newtask_results->{items}});
- } while (@{$newtask_results->{items}});
- foreach my $arvados_task (@$newtask_list) {
- my $jobstep = {
- 'level' => $arvados_task->{'sequence'},
- 'failures' => 0,
- 'arvados_task' => $arvados_task
- };
- push @jobstep, $jobstep;
- push @jobstep_todo, $#jobstep;
+ if ($task_success) {
+ # Load new tasks
+ my $newtask_list = [];
+ my $newtask_results;
+ do {
+ $newtask_results = $arv->{'job_tasks'}->{'list'}->execute(
+ 'where' => {
+ 'created_by_job_task_uuid' => $Jobstep->{'arvados_task'}->{uuid}
+ },
+ 'order' => 'qsequence',
+ 'offset' => scalar(@$newtask_list),
+ );
+ push(@$newtask_list, @{$newtask_results->{items}});
+ } while (@{$newtask_results->{items}});
+ foreach my $arvados_task (@$newtask_list) {
+ my $jobstep = {
+ 'level' => $arvados_task->{'sequence'},
+ 'failures' => 0,
+ 'arvados_task' => $arvados_task
+ };
+ push @jobstep, $jobstep;
+ push @jobstep_todo, $#jobstep;
+ }
}
$progress_is_dirty = 1;
my $hash = shift;
my ($keep, $child_out, $output_block);
- my $cmd = "$arv_cli keep get \Q$hash\E";
+ my $cmd = "arv-get \Q$hash\E";
open($keep, '-|', $cmd) or die "fetch_block: $cmd: $!";
- sysread($keep, $output_block, 64 * 1024 * 1024);
+ $output_block = '';
+ while (1) {
+ my $buf;
+ my $bytes = sysread($keep, $buf, 1024 * 1024);
+ if (!defined $bytes) {
+ die "reading from arv-get: $!";
+ } elsif ($bytes == 0) {
+ # sysread returns 0 at the end of the pipe.
+ last;
+ } else {
+ # some bytes were read into buf.
+ $output_block .= $buf;
+ }
+ }
close $keep;
return $output_block;
}
Log (undef, "collate");
my ($child_out, $child_in);
- my $pid = open2($child_out, $child_in, $arv_cli, 'keep', 'put', '--raw');
+ my $pid = open2($child_out, $child_in, 'arv-put', '--raw');
my $joboutput;
for (@jobstep)
{
sysread($child_out, $joboutput, 64 * 1024 * 1024);
chomp($joboutput);
} else {
- Log (undef, "timed out reading from 'arv keep put'");
+ Log (undef, "timed out reading from 'arv-put'");
}
}
waitpid($pid, 0);
return if $justcheckpoint; # checkpointing is not relevant post-Warehouse.pm
$local_logfile->flush;
- my $cmd = "$arv_cli keep put --filename ''\Q$keep_logfile\E "
+ my $cmd = "arv-put --filename ''\Q$keep_logfile\E "
. quotemeta($local_logfile->filename);
my $loglocator = `$cmd`;
die "system $cmd failed: $?" if $?;
}
}
+sub find_docker_hash {
+ # Given a Keep locator, search for a matching link to find the Docker hash
+ # of the stored image.
+ my $locator = shift;
+ my $links_result = $arv->{links}->{list}->execute(
+ filters => [["head_uuid", "=", $locator],
+ ["link_class", "=", "docker_image_hash"]],
+ limit => 1);
+ my $docker_hash;
+ foreach my $link (@{$links_result->{items}}) {
+ $docker_hash = lc($link->{name});
+ }
+ return $docker_hash;
+}
+
__DATA__
#!/usr/bin/perl