use strict;
use POSIX ':sys_wait_h';
+use POSIX qw(strftime);
use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
use Arvados;
use Digest::MD5 qw(md5_hex);
{
$Job = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec);
if (!$force_unlock) {
- # If some other crunch-job process has grabbed this job (or we see
- # other evidence that the job is already underway) we exit
- # EX_TEMPFAIL so crunch-dispatch (our parent process) doesn't
- # mark the job as failed.
- if ($Job->{'is_locked_by_uuid'}) {
- Log(undef, "Job is locked by " . $Job->{'is_locked_by_uuid'});
+ # Claim this job, and make sure nobody else does
+ eval {
+ # lock() sets is_locked_by_uuid and changes state to Running.
+ $arv->{'jobs'}->{'lock'}->execute('uuid' => $Job->{'uuid'})
+ };
+ if ($@) {
+ Log(undef, "Error while locking job, exiting ".EX_TEMPFAIL);
exit EX_TEMPFAIL;
- }
- if ($Job->{'success'} ne undef) {
- Log(undef, "Job 'success' flag (" . $Job->{'success'} . ") is not null");
- exit EX_TEMPFAIL;
- }
- if ($Job->{'running'}) {
- Log(undef, "Job 'running' flag is already set");
- exit EX_TEMPFAIL;
- }
- if ($Job->{'started_at'}) {
- Log(undef, "Job 'started_at' time is already set (" . $Job->{'started_at'} . ")");
- exit EX_TEMPFAIL;
- }
+ };
}
}
else
@slot = sort { $a->{cpu} <=> $b->{cpu} } @slot;
-
-my $jobmanager_id;
-if ($job_has_uuid)
-{
- # Claim this job, and make sure nobody else does
- unless ($Job->update_attributes('is_locked_by_uuid' => $User->{'uuid'}) &&
- $Job->{'is_locked_by_uuid'} == $User->{'uuid'}) {
- Log(undef, "Error while updating / locking job, exiting ".EX_TEMPFAIL);
- exit EX_TEMPFAIL;
- }
- $Job->update_attributes('started_at' => scalar gmtime,
- 'running' => 1,
- 'success' => undef,
- 'tasks_summary' => { 'failed' => 0,
- 'todo' => 1,
- 'running' => 0,
- 'done' => 0 });
-}
-
+$Job->update_attributes(
+ 'tasks_summary' => { 'failed' => 0,
+ 'todo' => 1,
+ 'running' => 0,
+ 'done' => 0 });
Log (undef, "start");
$SIG{'INT'} = sub { $main::please_freeze = 1; };
if ($treeish =~ m{^\d{1,4}$}) {
my $gitlog = `git --git-dir=\Q$repo\E log --pretty="format:%H" --grep="git-svn-id:.*\@"\Q$treeish\E" " master`;
chomp $gitlog;
- if ($gitlog =~ /^[a-f0-9]{40}$/) {
+ Log(undef, "git Subversion search exited $?");
+ if (($? == 0) && ($gitlog =~ /^[a-f0-9]{40}$/)) {
$commit = $gitlog;
- Log (undef, "Using commit $commit for script_version $treeish");
+ Log(undef, "Using commit $commit for Subversion revision $treeish");
}
}
if (!defined $commit) {
my $found = `git --git-dir=\Q$repo\E rev-list -1 ''\Q$treeish\E`;
chomp $found;
- if ($found =~ /^[0-9a-f]{40}$/s) {
+ Log(undef, "git rev-list exited $? with result '$found'");
+ if (($? == 0) && ($found =~ /^[0-9a-f]{40}$/s)) {
$commit = $found;
+ Log(undef, "Using commit $commit for tree-ish $treeish");
if ($commit ne $treeish) {
# Make sure we record the real commit id in the database,
# frozentokey, logs, etc. -- instead of an abbreviation or a
# branch name which can become ambiguous or point to a
# different commit in the future.
- $ENV{"CRUNCH_SRC_COMMIT"} = $commit;
- Log (undef, "Using commit $commit for tree-ish $treeish");
- if ($commit ne $treeish) {
- $Job->{'script_version'} = $commit;
- !$job_has_uuid or
- $Job->update_attributes('script_version' => $commit) or
- croak("Error while updating job");
- }
+ $Job->{'script_version'} = $commit;
+ !$job_has_uuid or
+ $Job->update_attributes('script_version' => $commit) or
+ croak("Error while updating job");
}
}
}
@execargs = ("sh", "-c",
"mkdir -p $ENV{CRUNCH_INSTALL} && cd $ENV{CRUNCH_TMP} && perl -");
$git_archive = `git --git-dir=\Q$repo\E archive ''\Q$commit\E`;
+ croak("git archive failed: exit " . ($? >> 8)) if ($? != 0);
}
else {
croak ("could not figure out commit id for $treeish");
if ($build_script)
{
$build_script_to_send = $build_script;
+ $command .=
+ "&& perl -";
}
- $command .= "&& exec arv-mount --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
+ $command .= "&& exec arv-mount --by-id --allow-other $ENV{TASK_KEEPMOUNT} --exec ";
if ($docker_hash)
{
$command .= "crunchstat -cgroup-root=/sys/fs/cgroup -cgroup-parent=docker -cgroup-cid=$ENV{TASK_WORK}/docker.cid -poll=10000 ";
$command .=
q{$(ip -o address show scope global |
gawk 'match($4, /^([0-9\.:]+)\//, x){print "--dns", x[1]}') };
+ $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 =~ /^(ARVADOS|JOB|TASK)_/) {
- if ($env_key eq "TASK_KEEPMOUNT") {
+ 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=\QCRUNCH_SRC=/tmp/crunch-src\E ";
$command .= "\Q$docker_hash\E ";
$command .= "stdbuf --output=0 --error=0 ";
- $command .= "perl - ";
$command .= "/tmp/crunch-src/crunch_scripts/" . $Job->{"script"};
} else {
# Non-docker run
$command .= "crunchstat -cgroup-root=/sys/fs/cgroup -poll=10000 ";
$command .= "stdbuf --output=0 --error=0 ";
- $command .= "perl - ";
$command .= "$ENV{CRUNCH_SRC}/crunch_scripts/" . $Job->{"script"};
}
delete $Jobstep->{stderr};
delete $Jobstep->{finishtime};
+ $Jobstep->{'arvados_task'}->{started_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{starttime});
+ $Jobstep->{'arvados_task'}->save;
+
splice @jobstep_todo, $todo_ptr, 1;
--$todo_ptr;
freeze();
my $collated_output = &collate_output();
-if ($job_has_uuid) {
- $Job->update_attributes('running' => 0,
- 'success' => $collated_output && $main::success,
- 'finished_at' => scalar gmtime)
-}
-
if (!$collated_output) {
Log(undef, "output undef");
}
my $output = $arv->{'collections'}->{'create'}->execute('collection' => {
'manifest_text' => $orig_manifest_text,
});
- Log(undef, "output " . $output->{portable_data_hash});
+ Log(undef, "output uuid " . $output->{uuid});
+ Log(undef, "output hash " . $output->{portable_data_hash});
$Job->update_attributes('output' => $output->{portable_data_hash}) if $job_has_uuid;
};
if ($@) {
Log (undef, "finish");
save_meta();
-exit ($Job->{'success'} ? 1 : 0);
+
+if ($job_has_uuid) {
+ if ($collated_output && $main::success) {
+ $Job->update_attributes('state' => 'Complete')
+ } else {
+ $Job->update_attributes('state' => 'Failed')
+ }
+}
+
+exit ($Job->{'state'} != 'Complete' ? 1 : 0);
}
$Jobstep->{exitcode} = $childstatus;
$Jobstep->{finishtime} = time;
+ $Jobstep->{'arvados_task'}->{finished_at} = strftime "%Y-%m-%dT%H:%M:%SZ", gmtime($Jobstep->{finishtime});
+ $Jobstep->{'arvados_task'}->save;
process_stderr ($jobstepid, $task_success);
Log ($jobstepid, "output " . $Jobstep->{'arvados_task'}->{output});
my $Job2 = $arv->{'jobs'}->{'get'}->execute('uuid' => $jobspec);
for my $attr ('cancelled_at',
'cancelled_by_user_uuid',
- 'cancelled_by_client_uuid') {
+ 'cancelled_by_client_uuid',
+ 'state') {
$Job->{$attr} = $Job2->{$attr};
}
- if ($Job->{'cancelled_at'}) {
- Log (undef, "Job cancelled at " . $Job->{cancelled_at} .
- " by user " . $Job->{cancelled_by_user_uuid});
+ if ($Job->{'state'} ne "Running") {
+ if ($Job->{'state'} eq "Cancelled") {
+ Log (undef, "Job cancelled at " . $Job->{'cancelled_at'} . " by user " . $Job->{'cancelled_by_user_uuid'});
+ } else {
+ Log (undef, "Job state unexpectedly changed to " . $Job->{'state'});
+ }
$main::success = 0;
$main::please_freeze = 1;
}
Log (undef, "collate");
my ($child_out, $child_in);
- my $pid = open2($child_out, $child_in, 'arv-put', '--raw');
+ my $pid = open2($child_out, $child_in, 'arv-put', '--raw',
+ '--retries', put_retry_count());
my $joboutput;
for (@jobstep)
{
sub cleanup
{
return if !$job_has_uuid;
- $Job->update_attributes('running' => 0,
- 'success' => 0,
- 'finished_at' => scalar gmtime);
+ if ($Job->{'state'} eq 'Cancelled') {
+ $Job->update_attributes('finished_at' => scalar gmtime);
+ } else {
+ $Job->update_attributes('state' => 'Failed');
+ }
}
return if $justcheckpoint; # checkpointing is not relevant post-Warehouse.pm
$local_logfile->flush;
- my $cmd = "arv-put --filename ''\Q$keep_logfile\E "
- . quotemeta($local_logfile->filename);
+ my $retry_count = put_retry_count();
+ my $cmd = "arv-put --portable-data-hash --retries $retry_count " .
+ "--filename ''\Q$keep_logfile\E " . quotemeta($local_logfile->filename);
my $loglocator = `$cmd`;
die "system $cmd failed: $?" if $?;
chomp($loglocator);
# If so, return its stream name and Docker hash.
# If not, return undef for both values.
my $locator = shift;
+ my ($streamname, $filename);
if (my $image = $arv->{collections}->{get}->execute(uuid => $locator)) {
- my @file_list = @{$image->{files}};
- if ((scalar(@file_list) == 1) &&
- ($file_list[0][1] =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
- return ($file_list[0][0], $1);
+ foreach my $line (split(/\n/, $image->{manifest_text})) {
+ my @tokens = split(/\s+/, $line);
+ next if (!@tokens);
+ $streamname = shift(@tokens);
+ foreach my $filedata (grep(/^\d+:\d+:/, @tokens)) {
+ if (defined($filename)) {
+ return (undef, undef); # More than one file in the Collection.
+ } else {
+ $filename = (split(/:/, $filedata, 3))[2];
+ }
+ }
}
}
- return (undef, undef);
+ if (defined($filename) and ($filename =~ /^([0-9A-Fa-f]{64})\.tar$/)) {
+ return ($streamname, $1);
+ } else {
+ return (undef, undef);
+ }
+}
+
+sub put_retry_count {
+ # Calculate a --retries argument for arv-put that will have it try
+ # approximately as long as this Job has been running.
+ my $stoptime = shift || time;
+ my $starttime = $jobstep[0]->{starttime};
+ my $timediff = defined($starttime) ? ($stoptime - $starttime) : 1;
+ my $retries = 0;
+ while ($timediff >= 2) {
+ $retries++;
+ $timediff /= 2;
+ }
+ return ($retries > 3) ? $retries : 3;
}
__DATA__
open L, ">", "$destdir.lock" or die "$destdir.lock: $!";
flock L, LOCK_EX;
if (readlink ("$destdir.commit") eq $commit && -d $destdir) {
- exec(@ARGV) if @ARGV;
- exit 0;
+ if (@ARGV) {
+ exec(@ARGV);
+ die "Cannot exec `@ARGV`: $!";
+ } else {
+ exit 0;
+ }
}
unlink "$destdir.commit";
if (@ARGV) {
exec(@ARGV);
+ die "Cannot exec `@ARGV`: $!";
} else {
exit 0;
}
projects / arvados.git / blobdiff