Index: trunk/pstamp/scripts/pstamp_checkdependent.pl =================================================================== --- trunk/pstamp/scripts/pstamp_checkdependent.pl (revision 27703) +++ trunk/pstamp/scripts/pstamp_checkdependent.pl (revision 27751) @@ -48,4 +48,5 @@ my $warptool = can_run('warptool') or (warn "Can't find warptool" and $missing_tools = 1); my $difftool = can_run('difftool') or (warn "Can't find difftool" and $missing_tools = 1); +my $stacktool = can_run('stacktool') or (warn "Can't find stacktool" and $missing_tools = 1); my $magicdstool = can_run('magicdstool') or (warn "Can't find magicdstool" and $missing_tools = 1); my $pstamptool = can_run('pstamptool') or (warn "Can't find pstamptool" and $missing_tools = 1); @@ -86,5 +87,5 @@ } else { if ($stage eq "chip") { - $cmd = "$chiptool -processedimfile -chip_id $stage_id -class_id $component"; + $cmd = "$chiptool -processedimfile -allfiles -chip_id $stage_id -class_id $component"; } elsif ($stage eq "warp") { $cmd = "$warptool -warped -warp_id $stage_id -skycell_id $component"; @@ -96,11 +97,9 @@ } -my $components = runToolAndParse($cmd, $verbose); - -my $n = scalar @$components; -my_die("Unexpected number of components found $n", $PS_EXIT_PROG_ERROR) if ($n != 1); +my $it = runToolAndParseExpectOne($cmd, $verbose); + +my_die("no components found", $PS_EXIT_PROG_ERROR) if ( !$it); # Got "it" -my $it = $components->[0]; my $status = 0; @@ -130,6 +129,5 @@ # need to queue the updates if ($stage eq 'chip') { - # queue_update_chip takes an array so that queue_update_warp can pass a set of - # chips to it + # queue_update_chip takes an array so that queue_update_warp can pass it set of chips my $chips = [$it]; $status = queue_update_chip($it->{chip_id}, $whole_run, $chips, $rlabel, $need_magic); @@ -145,8 +143,22 @@ # detect states that cannot be updated and update the job state - # XXX: we should be more assertive here and check for the specific states where we can continue + # XXX: Perhaps I should be more assertive here and check for the specific states that we know + # that we can continue. + my $state = $it->{state}; my $job_fault = 0; - if (($state =~ /scrub/) or ($state =~ /purge/)) { + + # temporary hack to deal with data with old burntool tables. This check is now done by pstampparse.pl + # so runs in this state will not get queued + if ($stage eq 'chip') { + my $burntool_state = $it->{burntool_state}; + if ($burntool_state and (abs($burntool_state) < 14)) { + print STDERR "chip $it->{chip_id} $it->{class_id} has burntool_state $burntool_state. Not avaiable.\n"; + $job_fault = $PSTAMP_NOT_AVAILABLE; + } + } + if ($state eq 'error_cleaned') { + $job_fault = $PSTAMP_NOT_AVAILABLE; + } elsif (($state =~ /scrub/) or ($state =~ /purge/)) { # jobs must have changed state since depenency was made print STDERR "Dependency cannot be satisfied\n"; @@ -155,4 +167,7 @@ # Dependency never should have been inserted my_die ("Unexpected state for ${stage}Run $stage_id $state", $PS_EXIT_PROG_ERROR); + } + if (!$job_fault and ($stage eq 'chip')) { + $job_fault = queue_update_magicDSRun($stage, $stage_id, $rlabel, $it->{raw_magicked}, $it->{dsRun_state}); } if ($job_fault) { @@ -213,7 +228,334 @@ } + my $status = queue_update_magicDSRun('chip', $chip_id, $rlabel, $raw_all_magicked, $dsRun_state); + + return $status; +} + +sub queue_update_warp { + # check status of input chips. If state is not updatable set error code for job + + # if chipProcessedImfile.state is cleaned call queue_update_chip + + # need to code warptool -setskyfiletoupdate + my $metadata = shift; + my $whole_run = shift; # if true queue entire run for update + my $rlabel = shift; # if defined a new label for the chipRun + my $need_magic = shift; + + my $raw_all_magicked = 1; # this gets cleared if any of the inputs aren't destreaked + + my $warp_id = $metadata->{warp_id}; + if (!$whole_run) { + my $skycell = $metadata; + my $skycell_id = $skycell->{skycell_id}; + + my $command = "$warptool -scmap -warp_id $warp_id -skycell_id $skycell_id"; + my $data = runToolAndParse($command, $verbose); + my_die("failed to find warpSkyCelllMap for warpRun $warp_id skycell_id $skycell_id", $PS_EXIT_UNKNOWN_ERROR) + if !$data or scalar @$data == 0; + + my $good_to_go = 1; + my @chipsToUpdate; + my $chip_id; + foreach my $chip (@$data) { + $chip_id = $chip->{chip_id}; + if (($chip->{data_state} ne 'full') or ($need_magic and ($chip->{magicked} <= 0))) { + $good_to_go = 0; + push @chipsToUpdate, $chip; + } else { + # this chip is good to go + } + } + + if ($good_to_go) { + my $command = "$warptool -setskyfiletoupdate -warp_id $warp_id -skycell_id $skycell->{skycell_id}"; + $command .= " -set_label $rlabel" if $rlabel; + + if (!$no_update) { + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => $verbose); + unless ($success) { + my_die("failed to queue ${stage}Run $stage_id $component for update", $PS_EXIT_UNKNOWN_ERROR); + } + } else { + print "skipping $command\n"; + } + } elsif (scalar @chipsToUpdate > 0) { + return queue_update_chip($chip_id, 0, \@chipsToUpdate, $rlabel, $need_magic); + } + } else { + my $warpRun = $metadata; + my $command = "$chiptool -listrun -chip_id $warpRun->{chip_id}"; + my $data = runToolAndParse($command, $verbose); + my_die("failed to find chipRun $warpRun->{chip_id} for warpRun $warp_id", $PS_EXIT_UNKNOWN_ERROR) + if !$data or scalar @$data != 1; + + my $chipRun = $data->[0]; + + my $chipRunState = $chipRun->{state}; + if (($chipRunState =~ /purge/) or ($chipRunState =~ /scrub/)) { + print STDERR "warpRun $warp_id depends on chipRun $chipRun->{chip_id} which is in state $chipRunState\n"; + faultJobs('stop', 'warp', $warp_id, $PSTAMP_GONE); + return 0; + } + my $warpRunState = $warpRun->{state}; + if (($chipRunState eq 'full') and (! $need_magic or ($chipRun->{magicked} > 0)) and ($warpRunState eq 'cleaned')) { + # The inputs and outputs are ready. Queue the warpRun for update. + + # providing no -skycell_id arguments changes all skyfiles with data_state = 'cleaned' to 'update' + my $command = "$warptool -setskyfiletoupdate -warp_id $warp_id"; + $command .= " -set_label $rlabel" if $rlabel; + + if (!$no_update) { + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => $verbose); + unless ($success) { + my_die("failed to queue ${stage}Run $stage_id $component for update", $PS_EXIT_UNKNOWN_ERROR); + } + } else { + print "skipping $command\n"; + } + } elsif ($chipRunState eq 'cleaned' or + (($chipRun->{state} eq 'full') and ($need_magic and ($chipRun->{magicked} < 0) and ($chipRun->{dsRun_state} ne 'new')))) { + my $data = [$chipRun]; + return queue_update_chip($chipRun->{chip_id}, 1, $data, $rlabel, $need_magic); + } + } + + # return value may be the return status of script so zero is good + return 0; +} +sub queue_update_diff { + my $metadata = shift; + my $whole_run = shift; # if true queue entire run for update + my $rlabel = shift; # if defined a new label for the chipRun + my $need_magic = shift; + +# XXXX This is not ready to go yet XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXxx + + my $diff_id = $metadata->{diff_id}; + my $diff_mode = $metadata->{diff_mode}; + if (!$whole_run) { + my $skycell = $metadata; + my $skycell_id = $skycell->{skycell_id}; + + # put this in a module somewhere + my $IPP_DIFF_MODE_WARP_WARP = 1; + my $IPP_DIFF_MODE_WARP_STACK = 2; + my $IPP_DIFF_MODE_STACK_WARP = 2; + my $IPP_DIFF_MODE_STACK_STACK = 4; + if ($diff_mode == $IPP_DIFF_MODE_WARP_STACK ) { + # check the state of the stack + my $command = "$stacktool -sumkskyfile -stack_id $skycell->{stack2}"; + my $stack = runToolAndParseExpectOne($command, $verbose); + my_die("failed to find stackSumSkyfile for stack_id $skycell->{stack2}", $PS_EXIT_UNKNOWN_ERROR) if !$stack; + + if ($stack->{state} ne 'full') { + print STDERR "template stack for diffRun $diff_id $skycell_id is not in full state faulting jobs\n"; + # this faults all jobs depending on this dep_id + faultJobs('stop', 'diff', $diff_id, $PSTAMP_GONE); + return $PSTAMP_GONE; + } + + # now check the warp + $command = "$warptool -warped -warp_id $skycell->{warp1} -skycell_id $skycell_id"; + my $warp = runToolAndExpectOne($command, $verbose); + my_die("failed to find warpSkyfile for warpRun $skycell->{warp_id} skycell_id $skycell_id", $PS_EXIT_UNKNOWN_ERROR) + if !$warp; + } + + my $good_to_go = 1; + my @chipsToUpdate; + my $chip_id; + my $warp_id; + my $data ; #XXXX + foreach my $chip (@$data) { + $chip_id = $chip->{chip_id}; + if (($chip->{data_state} ne 'full') or ($need_magic and ($chip->{magicked} <= 0))) { + $good_to_go = 0; + push @chipsToUpdate, $chip; + } else { + # this chip is good to go + } + } + + if ($good_to_go) { + my $command = "$warptool -setskyfiletoupdate -warp_id $warp_id -skycell_id $skycell->{skycell_id}"; + $command .= " -set_label $rlabel" if $rlabel; + + if (!$no_update) { + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => $verbose); + unless ($success) { + my_die("failed to queue ${stage}Run $stage_id $component for update", $PS_EXIT_UNKNOWN_ERROR); + } + } else { + print "skipping $command\n"; + } + } elsif (scalar @chipsToUpdate > 0) { + return queue_update_chip($chip_id, 0, \@chipsToUpdate, $rlabel, $need_magic); + } + } else { + my $warpRun = $metadata; + my $warp_id = $warpRun->{warp_id}; + my $command = "$chiptool -listrun -chip_id $warpRun->{chip_id}"; + my $data = runToolAndParse($command, $verbose); + my_die("failed to find chipRun $warpRun->{chip_id} for warpRun $warp_id", $PS_EXIT_UNKNOWN_ERROR) + if !$data or scalar @$data != 1; + + my $chipRun = $data->[0]; + + my $chipRunState = $chipRun->{state}; + if (($chipRunState =~ /purge/) or ($chipRunState =~ /scrub/)) { + print STDERR "warpRun $warp_id depends on chipRun $chipRun->{chip_id} which is in state $chipRunState\n"; + faultJobs('stop', 'warp', $warp_id, $PSTAMP_GONE); + return 0; + } + my $warpRunState = $warpRun->{state}; + if (($chipRunState eq 'full') and (! $need_magic or ($chipRun->{magicked} > 0)) and ($warpRunState eq 'cleaned')) { + # The inputs and outputs are ready. Queue the warpRun for update. + + # providing no -skycell_id arguments changes all skyfiles with data_state = 'cleaned' to 'update' + my $command = "$warptool -setskyfiletoupdate -warp_id $warp_id"; + $command .= " -set_label $rlabel" if $rlabel; + + if (!$no_update) { + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => $verbose); + unless ($success) { + my_die("failed to queue ${stage}Run $stage_id $component for update", $PS_EXIT_UNKNOWN_ERROR); + } + } else { + print "skipping $command\n"; + } + } elsif ($chipRunState eq 'cleaned' or + (($chipRun->{state} eq 'full') and ($need_magic and ($chipRun->{magicked} < 0) and ($chipRun->{dsRun_state} ne 'new')))) { + my $data = [$chipRun]; + return queue_update_chip($chipRun->{chip_id}, 1, $data, $rlabel, $need_magic); + } + } + + # return value may be the return status of script so zero is good + return 0; +} + + +# run a command that produces metadata output and parse the results into an array of objects +sub runToolAndParse { + my $command = shift; + my $verbose = shift; + + my ($program) = split " ", $command; + $program = basename($program); + + print "Running $command\n" if !$verbose; + my $start_tool = DateTime->now->mjd; + # run the command and parse the output + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => $verbose); + unless ($success) { + print STDERR @$stderr_buf if !$verbose; + return undef; + } + + my $now = DateTime->now->mjd; + my $dtime_tool = (DateTime->now->mjd - $start_tool) * 86400.; + print "Time to run $program: $dtime_tool\n"; + + my $buf = join "", @$stdout_buf; + if (!$buf) { + return undef; + } + + my $start_parse = DateTime->now->mjd; + + my $mdcParser = PS::IPP::Metadata::Config->new; + my $results = parse_md_fast($mdcParser, $buf) + or my_die ("Unable to parse metadata config doc", $PS_EXIT_UNKNOWN_ERROR); + + my $dtime_parse = (DateTime->now->mjd - $start_parse) * 86400.; + print "Time to parse results from $program: $dtime_parse\n"; + + return $results; +} + +# run an command returning metadata where we expect 1 entry +sub runToolAndParseExpectOne { + my $command = shift; + my $verbose = shift; + + my $data = runToolAndParse($command, $verbose); + + if (!$data) { + return undef; + } + + my $n = scalar @$data; + if ($n > 1) { + my_die("Unexpected number of components $n returned by $command", $PS_EXIT_PROG_ERROR); + } + + return $data->[0]; +} + +# splits meta data config input stream into single units to work around the pathalogically +# slow parser. This is similar to and adapted from code in various ippScripts. +sub parse_md_fast { + my $mdcParser = shift; + my $input = shift; + my $output = (); + + my @whole = split /\n/, $input; + my @single = (); + + my $n; + while ( ($n = @whole) > 0) { + my $value = shift @whole; + push @single, $value; + if ($value =~ /^\s*END\s*$/) { + push @single, "\n"; + + my $list = parse_md_list( $mdcParser->parse( join("\n", @single ) ) ) or + print STDERR "Unable to parse metdata config doc" and return undef; + push @$output, $list->[0]; + + @single = (); + } + } + return $output; +} + +sub faultJobs { + my ($state, $stage, $stage_id, $job_fault) = @_; + + my $command = "$pstamptool -updatejob -state stop -fault $job_fault -dep_id $dep_id"; + $command .= " -dbname $dbname" if $dbname; + $command .= " -dbserver $dbserver" if $dbserver; + if (!$no_update) { + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => $verbose); + unless ($success) { + my_die("failed to set pstampJob.fault for dep_id: $dep_id", + $PS_EXIT_UNKNOWN_ERROR); + } + } else { + print "skipping $command\n"; + } +} + +sub queue_update_magicDSRun { + my $stage = shift; + my $stage_id = shift; + my $rlabel = shift; + my $input_magicked = shift; + my $dsRun_state = shift; + + # XXX: this code is only fully coded and tested for chip stage + my_die ("queue_update_magicDSRun only works for stage chip", $PS_EXIT_PROG_ERROR) if $stage ne 'chip'; + # if called from queue_update_warp dsRun_state is unknown. Go find it. if (!$dsRun_state) { - my $command = "$chiptool -listrun -chip_id $chip_id"; + my $command = "$chiptool -listrun -chip_id $stage_id"; my $data = runToolAndParse($command, $verbose); my $chipRun = $data->[0]; @@ -222,5 +564,5 @@ # if the input file is already magicked no need to queue destreaking for this chipRun - if ($need_magic and !$raw_all_magicked) { + if ($need_magic and !$input_magicked) { if ($dsRun_state eq 'cleaned') { my $command = "$magicdstool -updaterun -set_state new -stage $stage -stage_id $stage_id"; @@ -242,177 +584,14 @@ } } - - # return value may be the return status of script so zero is good return 0; } -sub queue_update_warp { - # check status of input chips. If state is not updatable set error code for job - - # if chipProcessedImfile.state is cleaned call queue_update_chip - - # need to code warptool -setskyfiletoupdate - my $metadata = shift; - my $whole_run = shift; # if true queue entire run for update - my $rlabel = shift; # if defined a new label for the chipRun - my $need_magic = shift; - - my $raw_all_magicked = 1; # this gets cleared if any of the inputs aren't destreaked - - my $warp_id = $metadata->{warp_id}; - if (!$whole_run) { - my $skycell = $metadata; - my $skycell_id = $skycell->{skycell_id}; - - my $command = "$warptool -scmap -warp_id $warp_id -skycell_id $skycell_id"; - my $data = runToolAndParse($command, $verbose); - my_die("failed to find warpSkyCelllMap for warpRun $warp_id skycell_id $skycell_id", $PS_EXIT_UNKNOWN_ERROR) - if !$data or scalar @$data == 0; - - my $good_to_go = 1; - my @chipsToUpdate; - my $chip_id; - foreach my $chip (@$data) { - $chip_id = $chip->{chip_id}; - if (($chip->{data_state} ne 'full') or ($need_magic and ($chip->{magicked} <= 0))) { - $good_to_go = 0; - push @chipsToUpdate, $chip; - } else { - # this chip is good to go - } - } - - if ($good_to_go) { - my $command = "$warptool -setskyfiletoupdate -warp_id $warp_id -skycell_id $skycell->{skycell_id}"; - $command .= " -set_label $rlabel" if $rlabel; - - if (!$no_update) { - my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = - run(command => $command, verbose => $verbose); - unless ($success) { - my_die("failed to queue ${stage}Run $stage_id $component for update", $PS_EXIT_UNKNOWN_ERROR); - } - } else { - print "skipping $command\n"; - } - } elsif (scalar @chipsToUpdate > 0) { - return queue_update_chip($chip_id, 0, \@chipsToUpdate, $rlabel, $need_magic); - } - } else { - my $warpRun = $metadata; - my $command = "$chiptool -listrun -chip_id $warpRun->{chip_id}"; - my $data = runToolAndParse($command, $verbose); - my_die("failed to find chipRun $warpRun->{chip_id} for warpRun $warp_id", $PS_EXIT_UNKNOWN_ERROR) - if !$data or scalar @$data != 1; - - my $chipRun = $data->[0]; - - my $chipRunState = $chipRun->{state}; - if (($chipRunState =~ /purge/) or ($chipRunState =~ /scrub/)) { - print STDERR "warpRun $warp_id depends on chipRun $chipRun->{chip_id} which is in state $chipRunState\n"; - faultJobs('stop', 'warp', $warp_id, $PSTAMP_GONE); - return 0; - } - my $warpRunState = $warpRun->{state}; - if (($chipRunState eq 'full') and (! $need_magic or ($chipRun->{magicked} > 0)) and ($warpRunState eq 'cleaned')) { - # The inputs and outputs are ready. Queue the warpRun for update. - - # providing no -skycell_id arguments changes all skyfiles with data_state = 'cleaned' to 'update' - my $command = "$warptool -setskyfiletoupdate -warp_id $warp_id"; - $command .= " -set_label $rlabel" if $rlabel; - - if (!$no_update) { - my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = - run(command => $command, verbose => $verbose); - unless ($success) { - my_die("failed to queue ${stage}Run $stage_id $component for update", $PS_EXIT_UNKNOWN_ERROR); - } - } else { - print "skipping $command\n"; - } - } elsif ($chipRunState eq 'cleaned' or - (($chipRun->{state} eq 'full') and ($need_magic and ($chipRun->{magicked} < 0) and ($chipRun->{dsRun_state} ne 'new')))) { - my $data = [$chipRun]; - return queue_update_chip($chipRun->{chip_id}, 1, $data, $rlabel, $need_magic); - } - } - - # return value may be the return status of script so zero is good - return 0; -} -sub queue_update_diff { - return 0; -} - - -# run a command that produces metadata output and parse the results into an array of objects -sub runToolAndParse { - my $command = shift; - my $verbose = shift; - - my ($program) = split " ", $command; - $program = basename($program); - - print "Running $command\n" if !$verbose; - my $start_tool = DateTime->now->mjd; - # run the command and parse the output - my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = - run(command => $command, verbose => $verbose); - unless ($success) { - print STDERR @$stderr_buf if !$verbose; - return undef; - } - - my $now = DateTime->now->mjd; - my $dtime_tool = (DateTime->now->mjd - $start_tool) * 86400.; - print "Time to run $program: $dtime_tool\n"; - - my $buf = join "", @$stdout_buf; - if (!$buf) { - return undef; - } - - my $start_parse = DateTime->now->mjd; - - my $mdcParser = PS::IPP::Metadata::Config->new; - my $results = parse_md_fast($mdcParser, $buf) - or my_die ("Unable to parse metadata config doc", $PS_EXIT_UNKNOWN_ERROR); - - my $dtime_parse = (DateTime->now->mjd - $start_parse) * 86400.; - print "Time to parse results from $program: $dtime_parse\n"; - - return $results; -} - -# splits meta data config input stream into single units to work around the pathalogically -# slow parser. This is similar to and adapted from code in various ippScripts. -sub parse_md_fast { - my $mdcParser = shift; - my $input = shift; - my $output = (); - - my @whole = split /\n/, $input; - my @single = (); - - my $n; - while ( ($n = @whole) > 0) { - my $value = shift @whole; - push @single, $value; - if ($value =~ /^\s*END\s*$/) { - push @single, "\n"; - - my $list = parse_md_list( $mdcParser->parse( join("\n", @single ) ) ) or - print STDERR "Unable to parse metdata config doc" and return undef; - push @$output, $list->[0]; - - @single = (); - } - } - return $output; -} -sub faultJobs { - my ($state, $stage, $stage_id, $job_fault) = @_; - - my $command = "$pstamptool -updatejob -state stop -fault $job_fault -dep_id $dep_id"; +sub my_die +{ + my $msg = shift; + my $fault = shift; + carp $msg; + + my $command = "$pstamptool -updatedepedent -fault $fault -dep_id $dep_id"; $command .= " -dbname $dbname" if $dbname; $command .= " -dbserver $dbserver" if $dbserver; @@ -421,5 +600,5 @@ run(command => $command, verbose => $verbose); unless ($success) { - my_die("failed to set pstamJob.fault for dep_id: $dep_id", + my_die("failed to set pstampDependent.fault for dep_id: $dep_id", $PS_EXIT_UNKNOWN_ERROR); } @@ -427,13 +606,5 @@ print "skipping $command\n"; } -} - -sub my_die -{ - my $msg = shift; - my $fault = shift; - - carp $msg; - - return $fault; -} + + exit $fault; +}