Index: /trunk/ippScripts/scripts/sc_mk_stack_mdc.pl =================================================================== --- /trunk/ippScripts/scripts/sc_mk_stack_mdc.pl (revision 36844) +++ /trunk/ippScripts/scripts/sc_mk_stack_mdc.pl (revision 36844) @@ -0,0 +1,19 @@ +#! /usr/bin/env perl + +use Carp; +use warnings; +use strict; + +my $counter = 0; + +foreach my $path_base (@ARGV) { + print "INPUT${counter} METADATA\n"; + print " IMAGE STR ${path_base}.fits\n"; + print " MASK STR ${path_base}.mask.fits\n"; + print " VARIANCE STR ${path_base}.wt.fits\n"; + print " PSF STR ${path_base}.psf\n"; + print " SOURCES STR ${path_base}.cmf\n"; + print " BKGMODEL STR ${path_base}.mdl.fits\n"; + print "END\n\n"; + $counter++; +} Index: /trunk/ippScripts/scripts/sc_prepare_camera.pl =================================================================== --- /trunk/ippScripts/scripts/sc_prepare_camera.pl (revision 36844) +++ /trunk/ippScripts/scripts/sc_prepare_camera.pl (revision 36844) @@ -0,0 +1,359 @@ +#! /usr/bin/env perl + +use Carp; +use warnings; +use strict; + +use IPC::Cmd 0.36 qw( can_run run ); +use PS::IPP::Metadata::Config; +use PS::IPP::Metadata::List qw( parse_md_list ); +use PS::IPP::Config 1.01 qw( :standard ); +use DateTime; +use Data::Dumper; +use File::Basename; + +use Getopt::Long qw( GetOptions :config auto_help auto_version gnu_getopt ); +use Pod::Usage qw( pod2usage ); + + +# Hard coded values +my $remote_root = '/scratch3/watersc1/'; # Far side destination base location +my $threads = 2; # How many threads are we going to use? +my $job_cost = 1700 / 60 / 60; # Estimate of how long a job runs, in hours. +my $proc_per_node = 24; # processors per node +my $min_nodes = 1; # smallest allocation to ask for +my $max_nodes = 1000; # largest allocation to ask for +my $min_time = 1; # shortest allocation to ask for +my $max_time = 8; # longest allocation to ask for +my $remote_hostname= "LANL/Mustang"; # Name of the remote node. +my @return_component_list = ("DBINFO.EXP", "PSASTRO.CONFIG", "PSASTRO.OUTPUT", "LOG.EXP","TRACE.EXP", "PSASTRO.STATS"); +# Look for programs we need +my $missing_tools; + +my $remotetool = can_run('remotetool') or (warn "Can't find remotetool" and $missing_tools = 1); +my $camtool = can_run('camtool') or (warn "Can't find camtool" and $missing_tools = 1); + +if ($missing_tools) { + warn("Can't find required tools."); + exit($PS_EXIT_CONFIG_ERROR); +} + +my ($remote_id,$cam_id,,$camera,$dbname,$verbose,$path_base,$no_update); +GetOptions( + 'remote_id=s' => \$remote_id, + 'camera|c=s' => \$camera, + 'dbname|d=s' => \$dbname, + 'path_base=s' => \$path_base, + 'no_update' => \$no_update, + 'verbose' => \$verbose, + ) or pod2usage( 2 ); + +pod2usage( -msg => "Unknown option: @ARGV", -exitval => 2 ) if @ARGV; +pod2usage( -msg => "Required options: --remote_id --camera --dbname --path_base", -exitval => 3) unless + defined($remote_id) and + defined($camera) and + defined($path_base) and + defined($dbname); +my $ipprc = PS::IPP::Config->new( $camera ) or my_die( "Unable to set up", $remote_id); + +my $mdcParser = PS::IPP::Metadata::Config->new; # Parser for metadata config files + +# +# Step 1: Get a list of the components that make up this remoteRun + +# SHould this call listrun to ensure we're in state new? +my $rt_cmd = "$remotetool -listcomponent -remote_id $remote_id"; +$rt_cmd .= " -dbname $dbname " if defined($dbname); + +my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $rt_cmd, verbose => $verbose); +unless ($success) { + $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); + &my_die("Unable to run remotetool to determine stage parameters.", + $remote_id,$error_code); +} +my $compData = $mdcParser->parse(join "", @$stdout_buf) or + &my_die("Unable to determine component information.", + $remote_id,$PS_EXIT_PROG_ERROR); +my $compData2= parse_md_list($compData); + +# +# Step 1b: Open output files +my $uri_command = $path_base . ".cmd"; +my $uri_transfer= $path_base . ".transfer"; +my $uri_check = $path_base . ".check"; +my $uri_config = $path_base . ".config"; +my $uri_generate= $path_base . ".generate"; +my $uri_return = $path_base . ".return"; + +my $disk_command = $ipprc->file_resolve($uri_command,1); +my $disk_transfer= $ipprc->file_resolve($uri_transfer,1); +my $disk_check = $ipprc->file_resolve($uri_check,1); +my $disk_config = $ipprc->file_resolve($uri_config,1); +my $disk_generate= $ipprc->file_resolve($uri_generate,1); +my $disk_return = $ipprc->file_resolve($uri_return,1); + +my (undef, $remote_config) = uri_convert($uri_config); # Needs to be done after we've created it. + +open(TRANSFER, ">$disk_transfer") || &my_die("Couldn't open file? $disk_transfer",$remote_id,$PS_EXIT_SYS_ERROR); +open(CHECK, ">$disk_check") || &my_die("Couldn't open file? $disk_check",$remote_id,$PS_EXIT_SYS_ERROR); +open(CONFIG, ">$disk_config") || &my_die("Couldn't open file? $disk_config",$remote_id,$PS_EXIT_SYS_ERROR); +open(GENERATE, ">$disk_generate") || &my_die("Couldn't open file? $disk_generate",$remote_id,$PS_EXIT_SYS_ERROR); +open(RETURN, ">$disk_return") || &my_die("Couldn't open file? $disk_return", $remote_id, $PS_EXIT_SYS_ERROR); +my %file_filter = (); + +# +# Step 2: Iterate over all componenets in this remote run. +my $job_index = 0; +my @pre_commands = (); +my @main_commands = (); +my @post_commands = (); + +foreach my $compEntry (@$compData2) { + my $cam_id = $compEntry->{stage_id}; + +# Get exposure level information from the camRun we're working from. + + my $command = "$camtool -pendingexp -cam_id $cam_id "; + $command .= " -dbname $dbname " if defined($dbname); + + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => 0); + unless ($success) { + $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); + + &my_die("Unable to run camtool to determine stage parameters.", + $cam_id,$error_code); + } + + my $camData = $mdcParser->parse(join "", @$stdout_buf) or + &my_die("Unable to determine cam component information.", + $cam_id,$PS_EXIT_PROG_ERROR); + my $camData2= parse_md_list($camData); + my $camEntry = ${ $camData2 }[0]; + + my $workdir = $camEntry->{workdir}; + my $exp_tag = $camEntry->{exp_tag}; + unless (defined($workdir)) { + while( my ($k, $v) = each %$camEntry ) { + print "key: $k, value: $v.\n"; + } + + print "%{ $camEntry }\n"; + die; + } + my $ipp_outroot = "${workdir}/${exp_tag}/${exp_tag}.cm.${cam_id}"; + my $remote_outroot = uri_local_to_remote($ipp_outroot); + + +# +# Step 3: Iterate over the sub-components + + my $reg_command = "$camtool -pendingimfile -cam_id $cam_id "; + $reg_command .= " -dbname $dbname " if defined($dbname); + + ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $reg_command, verbose => 0); + unless ($success) { + $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); + &my_die("Unable to run camtool -pendingimfile ", + $cam_id,$error_code); + } + + + # We don't actually care about the input cam data other than to know which mask files to instantiate. + my $inpData = $mdcParser->parse(join "", @$stdout_buf) or + &my_die("Unable to determine cam component information.", + $cam_id, $PS_EXIT_PROG_ERROR); + my $inpData2=parse_md_list($inpData); + my $chipProto= ${ $inpData2 }[0]; + my $chip_path = $chipProto->{path_base}; + my $remote_chip_path = uri_local_to_remote( $chip_path ); + my $pre_cmd_cmfs = "ls -1 ${remote_chip_path}*.cmf > ${remote_outroot}.cmflist"; + my $pre_cmd_masks= "ls -1 ${remote_chip_path}*.mk.fits > ${remote_outroot}.masklist"; + + my $reduction = 'DEFAULT' unless defined $camEntry->{reduction}; + my $recipe_psastro = $ipprc->reduction($reduction, 'PSASTRO'); # Recipe to use + + my $psastro_command = " psastro -list ${remote_outroot}.cmflist "; + $psastro_command .= " -masklist ${remote_outroot}.masklist ${remote_outroot} "; +# $psastro_command .= " -refmasklist ${remote_outroot}.masklist ${remote_outroot} "; # This is used to do edge calculations. It should probably be smart enough to know that it can just use the masks I just specified, but it's not. + $psastro_command .= " -refmasklist /turquoise/usr/projects/ps1/watersc1/references/gpc1.refmask.list "; + $psastro_command .= " -astrommodel /turquoise/usr/projects/ps1/watersc1/references/gpc1.20080909.asm "; + $psastro_command .= " -recipe PSASTRO $recipe_psastro "; + $psastro_command .= " -tracedest ${remote_outroot}.trace -log ${remote_outroot}.log "; + $psastro_command .= " -dumpconfig ${remote_outroot}.mdc -stats ${remote_outroot}.stats "; + $psastro_command .= " -recipe PPSTATS CAMSTATS "; + + my $camtool_post_cmd = "camtool -cam_id $cam_id -addprocessedexp -uri UNKNOWN "; + $camtool_post_cmd .= " -dbname $dbname " if defined $dbname; $camtool_post_cmd .= " -path_base $ipp_outroot -hostname $remote_hostname -dtime_script 0 "; + my $post_cmd_echo = " echo -n \"$camtool_post_cmd\" > ${remote_outroot}.dbinfo "; + my $post_cmd_SfM = " ppStatsFromMetadata ${remote_outroot}.stats - CAMERA_EXP_FPA >> ${remote_outroot}.dbinfo "; + + print CONFIG "${pre_cmd_cmfs} && ${pre_cmd_masks} && ${psastro_command} && ${post_cmd_echo} && ${post_cmd_SfM} "; + $job_index++; + + # Determine which output files need to be returned + foreach my $component(@return_component_list) { + my $filename = $ipprc->filename($component,$ipp_outroot); + my ($ipp_disk, $remote_disk) = uri_to_outputs_for_return( $filename); + my $remote_outroot_dir = dirname($ipp_disk); + print CONFIG " && mkdir -p ${remote_root}/tmp/${remote_outroot_dir} && ln -sf $remote_disk ${remote_root}/tmp/${ipp_disk} "; + } + foreach my $chipInfo (@{ $inpData2 }) { + my $filename = $ipprc->filename("PSASTRO.OUTPUT.MASK",$ipp_outroot,$chipInfo->{class_id}); + my ($ipp_disk, $remote_disk) = uri_to_outputs_for_return( $filename); + my $remote_outroot_dir = dirname($ipp_disk); + print CONFIG " && mkdir -p ${remote_root}/tmp/${remote_outroot_dir} && ln -sf $remote_disk ${remote_root}/tmp/${ipp_disk} "; + } + print CONFIG "\n"; +} +close(CONFIG); +close(TRANSFER); +close(CHECK); +close(RETURN); +close(GENERATE); + +# +# Construct the moab command last, so we can use the job_index counter to estimate resources. Somehow. +my $proc_need = $job_index * $threads; # how many total processors do we need? +my $node_need = $proc_need / $proc_per_node; # this equals how many nodes? +my $time_need = $job_index * $job_cost; # How many seconds will this take? + +my $fill_factor = 0.8; # This is the factor of how much of the time allocation we'd like to fill +my ($time_req,$node_req); +if ($node_need * $job_cost < $fill_factor * $min_nodes * $min_time) { + $time_req = $min_time; + $node_req = $min_nodes; +} +elsif ($node_need * $job_cost > $fill_factor * $max_nodes * $max_time) { + $time_req = $max_time; + $node_req = $max_nodes; + print STDERR "You've requested the construction of a bundle that appears to need $node_need nodes and $job_cost time per job. This exceeds the max limits ($max_nodes, $max_time). Using those max values instead. Good luck.\n"; +} +else { + $time_req = int(($node_need * $job_cost) / ($fill_factor * $max_nodes)) + 1; + $node_req = int(($node_need * $job_cost) / ($fill_factor * $time_req)) + 1; +} + +open(COMMAND, ">$disk_command") || &my_die("Couldn't open file? $disk_command",$remote_id,$PS_EXIT_SYS_ERROR); +print COMMAND "#!/bin/tcsh\n"; +print COMMAND "##### Moab controll lines\n"; +print COMMAND "#MSUB -l nodes=${node_req}:ppn=${proc_per_node},walltime=${time_req}:00:00\n"; ## CHECK RESOURCES +print COMMAND "#MSUB -j oe\n"; +print COMMAND "#MSUB -V\n"; +print COMMAND "#MSUB -o ${remote_root}cam.${remote_id}.out\n"; +print COMMAND "date\n"; +print COMMAND 'srun -n $SLURM_JOB_NUM_NODES -m cyclic -l /bin/hostname | sort -n | awk \'{printf "%s\n", $2}\' > hosts.${SLURM_JOB_ID}' . "\n"; +print COMMAND "${remote_root}/stask_chip.py $remote_config " . 'hosts.${SLURM_JOB_ID} 1' . "\n"; +print COMMAND "date\n"; +close(COMMAND); + +unless($no_update) { + my $command = "remotetool -updaterun -remote_id $remote_id "; + $command .= " -set_state pending "; + $command .= " -dbname $dbname " if defined $dbname; + + system($command); +} + + +## Common SC routines + +sub uri_convert { # (ipp_disk,remote_disk) = uri_convert(neb_uri); + my $neb_uri = shift; + my $ipp_disk= $ipprc->file_resolve( $neb_uri ); + my $remote_disk = $ipp_disk; + + unless(defined($ipp_disk)) { + my_die(); + } + + $remote_disk =~ s%^.*/%%; # Remove nebulous path + $remote_disk =~ s%^\d+\.%%; # Remove ins_id + $remote_disk =~ s%:%/%g; # Replace colons with directories + $remote_disk = "${remote_root}/${remote_disk}"; + return($ipp_disk,$remote_disk); +} + +sub uri_convert_and_create { # (ipp_disk,remote_disk) = uri_convert_and_create(neb_uri); ipp_disk is created if it doesn't exist + my $neb_uri = shift; + my $ipp_disk= $ipprc->file_resolve( $neb_uri , 1); + my $remote_disk = $ipp_disk; + + unless(defined($ipp_disk)) { + my_die(); + } + + $remote_disk =~ s%^.*/%%; # Remove nebulous path + $remote_disk =~ s%^\d+\.%%; # Remove ins_id + $remote_disk =~ s%:%/%g; # Replace colons with directories + $remote_disk = "${remote_root}/${remote_disk}"; + return($ipp_disk,$remote_disk); +} + +sub uri_to_outputs { # (ipp_disk,remote_disk) = uri_to_output(neb_uri); Appends to TRANSFER and CHECK filehandles + my $neb_uri = shift; + my ($ipp_disk, $remote_disk) = uri_convert( $neb_uri ); + + unless (exists($file_filter{$neb_uri})) { + $file_filter{$neb_uri} = 1; + print TRANSFER "$ipp_disk\n"; + print CHECK "$remote_disk\n"; + } + return($ipp_disk,$remote_disk); +} + +sub uri_to_outputs_for_return { # (ipp_disk,remote_disk) = uri_to_outputs_for_return(neb_uri); create ipp_disk, append to RETURN and GENERATE + my $neb_uri = shift; + my ($ipp_disk, $remote_disk) = uri_convert_and_create( $neb_uri ); + unless (exists($file_filter{$neb_uri})) { + $file_filter{$neb_uri} = 1; + print RETURN "$ipp_disk\n"; + print GENERATE "$remote_disk\n"; + } + return($ipp_disk,$remote_disk); +} + +sub uri_local_to_remote { #(remote_uri) = uri_local_to_remote(local_neb_uri); + # This needs to replace the nebulous tag with the remote root. + my $local_uri = shift; + $local_uri =~ s%^.*?/%%; # neb:/ + $local_uri =~ s%^.*?/%%; # / + $local_uri =~ s%^.*?/%%; # @HOST@.0/ + my $remote_uri = "${remote_root}/" . $local_uri; + + return($remote_uri); +} + +sub uri_remote_to_local { #(local_neb_uri) = uri_remote_to_local(remote_uri); + # This needs to replace the remote root directory with the nebulous tag. + my $remote_uri = shift; + $remote_uri =~ s%${remote_root}%%; + my $local_uri = "neb:///" . $remote_uri; + + return($local_uri); +} + + +sub my_die { # exit with status; my_die(message,stage_id,exit_code,exit_status); + my $msg = shift; + my $id = shift; + my $exit_code = shift; + my $exit_state = shift; + + $exit_code = $PS_EXIT_PROG_ERROR unless defined $exit_code; + + carp($msg); + + if (defined $id and not $no_update) { + my $command = "remotetool -updaterun -stage_id $id"; + $command .= " -fault $exit_code " if defined $exit_code; + $command .= " -set_state $exit_state " if defined $exit_state; + $command .= " -dbname $dbname " if defined $dbname; + + system($command); + } + + exit($exit_code); +} Index: /trunk/ippScripts/scripts/sc_prepare_chip.pl =================================================================== --- /trunk/ippScripts/scripts/sc_prepare_chip.pl (revision 36843) +++ /trunk/ippScripts/scripts/sc_prepare_chip.pl (revision 36844) @@ -19,4 +19,5 @@ # Hard coded values my $remote_root = '/scratch3/watersc1/'; # Far side destination base location +my $remote_raw = "${remote_root}/raw/"; # Directory to find raw data in. my $threads = 2; # How many threads are we going to use? my $job_cost = 150 / 60 / 60; # Estimate of how long a job runs, in hours. @@ -150,4 +151,5 @@ #print Dumper(%detrends); +my @return_component_list = ("DBINFO.IMFILE","PPIMAGE.STATS","LOG.IMFILE","TRACE.IMFILE","PPIMAGE.BACKMDL","PPIMAGE.PATTERN"); # @@ -157,4 +159,6 @@ my $uri_check = $path_base . ".check"; my $uri_config = $path_base . ".config"; +my $uri_generate= $path_base . ".generate"; +my $uri_return = $path_base . ".return"; my $disk_command = $ipprc->file_resolve($uri_command,1); @@ -162,4 +166,6 @@ my $disk_check = $ipprc->file_resolve($uri_check,1); my $disk_config = $ipprc->file_resolve($uri_config,1); +my $disk_generate= $ipprc->file_resolve($uri_generate,1); +my $disk_return = $ipprc->file_resolve($uri_return,1); my (undef, $remote_config) = uri_convert($uri_config); # Needs to be done after we've created it. @@ -168,4 +174,7 @@ open(CHECK, ">$disk_check") || &my_die("Couldn't open file? $disk_check",$chip_id,$PS_EXIT_SYS_ERROR); open(CONFIG, ">$disk_config") || &my_die("Couldn't open file? $disk_config",$chip_id,$PS_EXIT_SYS_ERROR); +open(GENERATE, ">$disk_generate") || &my_die("Couldn't open file? $disk_generate",$chip_id,$PS_EXIT_SYS_ERROR); +open(RETURN, ">$disk_return") || &my_die("Couldn't open file? $disk_return", $chip_id, $PS_EXIT_SYS_ERROR); + # @@ -185,5 +194,5 @@ ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = - run(command => $command, verbose => $verbose); + run(command => $command, verbose => 0); unless ($success) { $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); @@ -254,6 +263,7 @@ my $chipData = $mdcParser->parse(join "", @$stdout_buf) or - &my_die("Unable to parse chiptool -pendingimfile information.", - $chip_id,$error_code); + next; +# &my_die("Unable to parse chiptool -pendingimfile information.", +# $chip_id,$error_code); my $chipData2= parse_md_list($chipData); @@ -270,8 +280,8 @@ # Process the image and burntool table - my (undef,$remote_uri) = uri_to_outputs($uri); + my ($ipp_uri,$remote_uri) = uri_to_outputs_raw($uri); my $btt = $uri; $btt =~ s/fits$/burn.tbl/; - my (undef,$remote_btt) = uri_to_outputs($btt); + my ($ipp_btt,$remote_btt) = uri_to_outputs_raw($btt); # Initialize the ppI command @@ -287,5 +297,5 @@ my $det_code = $det_types{$det}; my $duri = $detrends_to_use{$det}->{$class_id}; - my (undef, $remote_det_uri) = uri_to_outputs($duri); + my ($ipp_det_uri, $remote_det_uri) = uri_to_outputs_raw($duri); $ppImage_command .= " $det_code $remote_det_uri "; } @@ -310,9 +320,9 @@ $ppImage_command .= " -log ${remote_outroot}.${class_id}.log "; - push @main_commands, $ppImage_command; +# push @main_commands, $ppImage_command; # Calculate pre and post commands my $remote_outroot_dir = dirname($remote_outroot); - push @pre_commands, "mkdir -p $remote_outroot_dir"; + my $pre_command = "mkdir -p $remote_outroot_dir"; my $post_commandA = "chiptool -addprocessedimfile -exp_id $exp_id -chip_id $chip_id -class_id $class_id "; @@ -321,37 +331,55 @@ $post_commandA .= " -dbname $dbname " if defined $dbname; - my $post_commandB = "echo -n $post_commandA > ${remote_outroot}.${class_id}.dbinfo && "; + my $post_commandB = "echo -n \"$post_commandA\" > ${remote_outroot}.${class_id}.dbinfo "; my $post_commandC = "ppStatsFromMetadata ${remote_outroot}.${class_id}.stats - CHIP_IMFILE >> ${remote_outroot}.${class_id}.dbinfo"; - push @post_commands, "$post_commandB $post_commandC"; +# push @post_commands, "$post_commandB $post_commandC"; $job_index++; + + print CONFIG "${pre_command} && ${ppImage_command} && ${post_commandB} && ${post_commandC}"; + + + # Determine which output files need to be returned. + foreach my $component (@return_component_list) { +# uri_convert_and_create( $ipprc->filename($component, $ipp_outroot, $class_id) ); + my $filename = $ipprc->filename($component, $ipp_outroot, $class_id); + my ($ipp_disk, $remote_disk) = uri_to_outputs_for_return( $filename ); + my $remote_outroot_dir = dirname($ipp_disk); + print CONFIG " && mkdir -p ${remote_root}/tmp/${remote_outroot_dir} && ln -sf $remote_disk ${remote_root}/tmp/${ipp_disk} "; +# print " $filename $ipp_disk $remote_disk\n"; +# die; + } + + print CONFIG "\n"; } } # Actually put the commands into the output file. -my %unique_cmds = (); -foreach my $cmd (@pre_commands) { - unless(exists($unique_cmds{$cmd})) { - print CONFIG $cmd . "\n"; - $unique_cmds{$cmd} = 1; - } -} -foreach my $cmd (@main_commands) { - unless(exists($unique_cmds{$cmd})) { - print CONFIG $cmd . "\n"; - $unique_cmds{$cmd} = 1; - } -} -foreach my $cmd (@post_commands) { - unless(exists($unique_cmds{$cmd})) { - print CONFIG $cmd . "\n"; - $unique_cmds{$cmd} = 1; - } -} +# my %unique_cmds = (); +# foreach my $cmd (@pre_commands) { +# unless(exists($unique_cmds{$cmd})) { +# print CONFIG $cmd . "\n"; +# $unique_cmds{$cmd} = 1; +# } +# } +# foreach my $cmd (@main_commands) { +# unless(exists($unique_cmds{$cmd})) { +# print CONFIG $cmd . "\n"; +# $unique_cmds{$cmd} = 1; +# } +# } +# foreach my $cmd (@post_commands) { +# unless(exists($unique_cmds{$cmd})) { +# print CONFIG $cmd . "\n"; +# $unique_cmds{$cmd} = 1; +# } +# } close(CONFIG); close(TRANSFER); close(CHECK); +close(RETURN); +close(GENERATE); # @@ -425,4 +453,20 @@ } +sub uri_convert_and_create { + my $neb_uri = shift; + my $ipp_disk= $ipprc->file_resolve( $neb_uri , 1); + my $remote_disk = $ipp_disk; + + unless(defined($ipp_disk)) { + my_die(); + } + + $remote_disk =~ s%^.*/%%; # Remove nebulous path + $remote_disk =~ s%^\d+\.%%; # Remove ins_id + $remote_disk =~ s%:%/%g; # Replace colons with directories + $remote_disk = "${remote_root}/${remote_disk}"; + return($ipp_disk,$remote_disk); +} + sub uri_to_outputs { my $neb_uri = shift; @@ -433,4 +477,27 @@ print TRANSFER "$ipp_disk\n"; print CHECK "$remote_disk\n"; + } + return($ipp_disk,$remote_disk); +} + +sub uri_to_outputs_raw { + my $neb_uri = shift; + my ($ipp_disk, $remote_disk) = uri_convert( $neb_uri ); + $remote_disk = $remote_raw . $ipp_disk; + unless (exists($file_filter{$neb_uri})) { + $file_filter{$neb_uri} = 1; + print TRANSFER "$ipp_disk\n"; + print CHECK "$remote_disk\n"; + } + return($ipp_disk,$remote_disk); +} + +sub uri_to_outputs_for_return { + my $neb_uri = shift; + my ($ipp_disk, $remote_disk) = uri_convert_and_create( $neb_uri ); + unless (exists($file_filter{$neb_uri})) { + $file_filter{$neb_uri} = 1; + print RETURN "$ipp_disk\n"; + print GENERATE "$remote_disk\n"; } return($ipp_disk,$remote_disk); Index: /trunk/ippScripts/scripts/sc_prepare_warp.pl =================================================================== --- /trunk/ippScripts/scripts/sc_prepare_warp.pl (revision 36844) +++ /trunk/ippScripts/scripts/sc_prepare_warp.pl (revision 36844) @@ -0,0 +1,377 @@ +#! /usr/bin/env perl + +use Carp; +use warnings; +use strict; + +use IPC::Cmd 0.36 qw( can_run run ); +use PS::IPP::Metadata::Config; +use PS::IPP::Metadata::List qw( parse_md_list ); +use PS::IPP::Config 1.01 qw( :standard ); +use DateTime; +use Data::Dumper; +use File::Basename; + +use Getopt::Long qw( GetOptions :config auto_help auto_version gnu_getopt ); +use Pod::Usage qw( pod2usage ); + + +# Hard coded values +my $remote_root = '/scratch3/watersc1/'; # Far side destination base location +my $threads = 2; # How many threads are we going to use? +my $job_cost = 70 / 60 / 60; # Estimate of how long a job runs, in hours. +my $proc_per_node = 24; # processors per node +my $min_nodes = 1; # smallest allocation to ask for +my $max_nodes = 1000; # largest allocation to ask for +my $min_time = 1; # shortest allocation to ask for +my $max_time = 8; # longest allocation to ask for +my $remote_hostname= "LANL/Mustang"; # Name of the remote node. +my @return_component_list = ("DBINFO.EXP", "PSWARP.CONFIG", "PSWARP.OUTPUT", "PSWARP.OUTPUT.MASK", "PSWARP.OUTPUT.VARIANCE", "PSWARP.OUTPUT.SOURCES","LOG.EXP","TRACE.EXP"); +# Look for programs we need +my $missing_tools; + +my $remotetool = can_run('remotetool') or (warn "Can't find remotetool" and $missing_tools = 1); +my $warptool = can_run('warptool') or (warn "Can't find warptool" and $missing_tools = 1); + +if ($missing_tools) { + warn("Can't find required tools."); + exit($PS_EXIT_CONFIG_ERROR); +} + +my ($remote_id,$warp_id,,$camera,$dbname,$verbose,$path_base,$no_update); +GetOptions( + 'remote_id=s' => \$remote_id, + 'camera|c=s' => \$camera, + 'dbname|d=s' => \$dbname, + 'path_base=s' => \$path_base, + 'no_update' => \$no_update, + 'verbose' => \$verbose, + ) or pod2usage( 2 ); + +pod2usage( -msg => "Unknown option: @ARGV", -exitval => 2 ) if @ARGV; +pod2usage( -msg => "Required options: --remote_id --camera --dbname --path_base", -exitval => 3) unless + defined($remote_id) and + defined($camera) and + defined($path_base) and + defined($dbname); +my $ipprc = PS::IPP::Config->new( $camera ) or my_die( "Unable to set up", $warp_id); + +my $mdcParser = PS::IPP::Metadata::Config->new; # Parser for metadata config files + +# +# Step 1: Get a list of the components that make up this remoteRun + +# SHould this call listrun to ensure we're in state new? +my $rt_cmd = "$remotetool -listcomponent -remote_id $remote_id"; +$rt_cmd .= " -dbname $dbname " if defined($dbname); + +my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $rt_cmd, verbose => $verbose); +unless ($success) { + $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); + &my_die("Unable to run remotetool to determine stage parameters.", + $remote_id,$error_code); +} +my $compData = $mdcParser->parse(join "", @$stdout_buf) or + &my_die("Unable to determine component information.", + $remote_id,$PS_EXIT_PROG_ERROR); +my $compData2= parse_md_list($compData); + +# +# Step 1b: Open output files +my $uri_command = $path_base . ".cmd"; +my $uri_transfer= $path_base . ".transfer"; +my $uri_check = $path_base . ".check"; +my $uri_config = $path_base . ".config"; +my $uri_generate= $path_base . ".generate"; +my $uri_return = $path_base . ".return"; + +my $disk_command = $ipprc->file_resolve($uri_command,1); +my $disk_transfer= $ipprc->file_resolve($uri_transfer,1); +my $disk_check = $ipprc->file_resolve($uri_check,1); +my $disk_config = $ipprc->file_resolve($uri_config,1); +my $disk_generate= $ipprc->file_resolve($uri_generate,1); +my $disk_return = $ipprc->file_resolve($uri_return,1); + +my (undef, $remote_config) = uri_convert($uri_config); # Needs to be done after we've created it. + +open(TRANSFER, ">$disk_transfer") || &my_die("Couldn't open file? $disk_transfer",$remote_id,$PS_EXIT_SYS_ERROR); +open(CHECK, ">$disk_check") || &my_die("Couldn't open file? $disk_check",$remote_id,$PS_EXIT_SYS_ERROR); +open(CONFIG, ">$disk_config") || &my_die("Couldn't open file? $disk_config",$remote_id,$PS_EXIT_SYS_ERROR); +open(GENERATE, ">$disk_generate") || &my_die("Couldn't open file? $disk_generate",$remote_id,$PS_EXIT_SYS_ERROR); +open(RETURN, ">$disk_return") || &my_die("Couldn't open file? $disk_return", $remote_id, $PS_EXIT_SYS_ERROR); + + +# +# Step 2: Iterate over all componenets in this remote run. +my $job_index = 0; +my @pre_commands = (); +my @main_commands = (); +my @post_commands = (); + +my %file_filter = (); + +foreach my $compEntry (@$compData2) { + my $warp_id = $compEntry->{stage_id}; + +# Get exposure level information from the warpRun we're working from. + # This actually returns all the individual warp/skyfiles that comprise this run. Because consistency. + my $command = "$warptool -towarped -warp_id $warp_id "; + ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => 0); + unless ($success) { + $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); + &my_die("Unable to run warptool -pendingimfile ", + $warp_id,$error_code); + } + my $warpData = $mdcParser->parse(join "", @$stdout_buf) or + &my_die("Unable to determine warp component information.", + $warp_id,$PS_EXIT_PROG_ERROR); + my $warpData2= parse_md_list($warpData); + +# +# Step 3: Iterate over the sub-components + foreach my $warpEntry ( @{ $warpData2 } ) { + my $workdir = $warpEntry->{workdir}; + my $exp_tag = $warpEntry->{exp_tag}; + my $skycell_id = $warpEntry->{skycell_id}; + my $tess_id = $warpEntry->{tess_id}; + my $reduction = 'DEFAULT' unless defined $warpEntry->{reduction}; + my $recipe_pswarp = $ipprc->reduction($reduction, 'WARP_PSWARP'); # Recipe to use + my $ipp_outroot = "${workdir}/${exp_tag}/${exp_tag}.wrp.${warp_id}.${skycell_id}"; + print "$ipp_outroot\n"; + my $remote_outroot = uri_local_to_remote($ipp_outroot); + + my $reg_command = "$warptool -scmap -warp_id $warp_id -skycell_id ${skycell_id}"; + $reg_command .= " -dbname $dbname " if defined($dbname); + + ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $reg_command, verbose => 0); + unless ($success) { + $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); + &my_die("Unable to run warptool -pendingimfile: $reg_command ", + $warp_id,$error_code); + } + + my $inpData = $mdcParser->parse(join "", @$stdout_buf) or + &my_die("Unable to determine warp component information.", + $warp_id, $PS_EXIT_PROG_ERROR); + my $inpData2=parse_md_list($inpData); + + my $pre_cmd_ims = "ls -1 "; + my $pre_cmd_masks= "ls -1 "; + my $pre_cmd_vars = "ls -1 "; + my $pre_cmd_astrom = ""; + + foreach my $imfile (@{ $inpData2 }) { + my $image = $ipprc->filename("PPIMAGE.CHIP", $imfile->{chip_path_base}, $imfile->{class_id}); + my $mask = $ipprc->filename("PSASTRO.OUTPUT.MASK", $imfile->{cam_path_base}, $imfile->{class_id}); + my $var = $ipprc->filename("PPIMAGE.CHIP.VARIANCE", $imfile->{chip_path_base}, $imfile->{class_id}); + + my $astrom= $ipprc->filename("PSASTRO.OUTPUT", $imfile->{cam_path_base}); + + my $remote_image = uri_local_to_remote($image); + my $remote_mask = uri_local_to_remote($mask); + my $remote_var = uri_local_to_remote($var); + my $remote_astrom= uri_local_to_remote($astrom); + + $pre_cmd_ims .= " $remote_image "; + $pre_cmd_masks .= " $remote_mask "; + $pre_cmd_vars .= " $remote_var "; + $pre_cmd_astrom = "echo $remote_astrom > ${remote_outroot}.astrom "; + } + $pre_cmd_ims .= " > ${remote_outroot}.imlist "; + $pre_cmd_masks .= " > ${remote_outroot}.masklist "; + $pre_cmd_vars .= " > ${remote_outroot}.varlist "; + + my $skycell_command = " dvoImageExtract -D CATDIR /turquoise/usr/projects/ps1/watersc1/tess/${tess_id} $skycell_id -o ${remote_outroot}.skyfile "; + + my $pswarp_command = " pswarp -list ${remote_outroot}.imlist "; + $pswarp_command .= " -masklist ${remote_outroot}.masklist -variancelist ${remote_outroot}.varlist "; + $pswarp_command .= " -astromlist ${remote_outroot}.astrom "; + $pswarp_command .= " ${remote_outroot} ${remote_outroot}.skyfile "; + $pswarp_command .= " -F PSPHOT.PSF.SAVE PSPHOT.PSF.SKY.SAVE "; + $pswarp_command .= " -F PSPHOT.OUTPUT PSPHOT.OUT.CMF.MEF "; + $pswarp_command .= " -F PSPHOT.BACKMDL PSPHOT.BACKMDL.MEF "; + $pswarp_command .= " -F SOURCE.PLOT.MOMENTS SOURCE.PLOT.SKY.MOMENTS "; + $pswarp_command .= " -F SOURCE.PLOT.PSFMODEL SOURCE.PLOT.SKY.PSFMODEL "; + $pswarp_command .= " -F SOURCE.PLOT.APRESID SOURCE.PLOT.SKY.APRESID "; + $pswarp_command .= " -recipe PSWARP $recipe_pswarp "; + $pswarp_command .= " -tracedest ${remote_outroot}.trace -log ${remote_outroot}.log "; + $pswarp_command .= " -threads $threads "; # -image_id ${image_id} -source_id ${source_id} "; + $pswarp_command .= " -recipe PPSTATS WARPSTATS "; + $pswarp_command .= " -dumpconfig ${remote_outroot}.mdc -stats ${remote_outroot}.stats "; + +# print "$pswarp_command \n"; + + my $post_cmd_echo = " echo -n \"warptool -addwarped -warp_id $warp_id -skycell_id $skycell_id -tess_id $tess_id "; + $post_cmd_echo .= " -path_base $remote_outroot -uri UNKNOWN "; + $post_cmd_echo .= " -dbname $dbname " if defined $dbname; + $post_cmd_echo .= " -path_base $ipp_outroot -hostname $remote_hostname -dtime_script 0 \" > ${remote_outroot}.dbinfo "; + + my $post_cmd_SfM = " ppStatsFromMetadata ${remote_outroot}.stats - WARP_SKYCELL >> ${remote_outroot}.dbinfo "; + + print CONFIG "${pre_cmd_ims} && ${pre_cmd_masks} && ${pre_cmd_vars} && ${pre_cmd_astrom} && ${skycell_command} && ${pswarp_command} && ${post_cmd_echo} && ${post_cmd_SfM} "; + $job_index++; + + # Determine which output files need to be returned + foreach my $component(@return_component_list) { + my $filename = $ipprc->filename($component,$ipp_outroot); + my ($ipp_disk, $remote_disk) = uri_to_outputs_for_return( $filename); + my $remote_outroot_dir = dirname($ipp_disk); + print CONFIG " && mkdir -p ${remote_root}/tmp/${remote_outroot_dir} && ln -sf $remote_disk ${remote_root}/tmp/${ipp_disk} "; + } + + print CONFIG "\n"; +# die(); + } +} +close(CONFIG); +close(TRANSFER); +close(CHECK); +close(RETURN); +close(GENERATE); + +# +# Construct the moab command last, so we can use the job_index counter to estimate resources. Somehow. +my $proc_need = $job_index * $threads; # how many total processors do we need? +my $node_need = $proc_need / $proc_per_node; # this equals how many nodes? +my $time_need = $job_index * $job_cost; # How many seconds will this take? + +my $fill_factor = 0.8; # This is the factor of how much of the time allocation we'd like to fill +my ($time_req,$node_req); +if ($node_need * $job_cost < $fill_factor * $min_nodes * $min_time) { + $time_req = $min_time; + $node_req = $min_nodes; +} +elsif ($node_need * $job_cost > $fill_factor * $max_nodes * $max_time) { + $time_req = $max_time; + $node_req = $max_nodes; + print STDERR "You've requested the construction of a bundle that appears to need $node_need nodes and $job_cost time per job. This exceeds the max limits ($max_nodes, $max_time). Using those max values instead. Good luck.\n"; +} +else { + $time_req = int(($node_need * $job_cost) / ($fill_factor * $max_nodes)) + 1; + $node_req = int(($node_need * $job_cost) / ($fill_factor * $time_req)) + 1; +} + +open(COMMAND, ">$disk_command") || &my_die("Couldn't open file? $disk_command",$remote_id,$PS_EXIT_SYS_ERROR); +print COMMAND "#!/bin/tcsh\n"; +print COMMAND "##### Moab controll lines\n"; +print COMMAND "#MSUB -l nodes=${node_req}:ppn=${proc_per_node},walltime=${time_req}:00:00\n"; ## CHECK RESOURCES +print COMMAND "#MSUB -j oe\n"; +print COMMAND "#MSUB -V\n"; +print COMMAND "#MSUB -o ${remote_root}warp.${remote_id}.out\n"; +print COMMAND "date\n"; +print COMMAND 'srun -n $SLURM_JOB_NUM_NODES -m cyclic -l /bin/hostname | sort -n | awk \'{printf "%s\n", $2}\' > hosts.${SLURM_JOB_ID}' . "\n"; +print COMMAND "${remote_root}/stask_chip.py $remote_config " . 'hosts.${SLURM_JOB_ID} 1' . "\n"; +print COMMAND "date\n"; +close(COMMAND); + +unless($no_update) { + my $command = "remotetool -updaterun -remote_id $remote_id "; + $command .= " -set_state pending "; + $command .= " -dbname $dbname " if defined $dbname; + + system($command); +} + + +## Common SC routines + +sub uri_convert { # (ipp_disk,remote_disk) = uri_convert(neb_uri); + my $neb_uri = shift; + my $ipp_disk= $ipprc->file_resolve( $neb_uri ); + my $remote_disk = $ipp_disk; + + unless(defined($ipp_disk)) { + my_die(); + } + + $remote_disk =~ s%^.*/%%; # Remove nebulous path + $remote_disk =~ s%^\d+\.%%; # Remove ins_id + $remote_disk =~ s%:%/%g; # Replace colons with directories + $remote_disk = "${remote_root}/${remote_disk}"; + return($ipp_disk,$remote_disk); +} + +sub uri_convert_and_create { # (ipp_disk,remote_disk) = uri_convert_and_create(neb_uri); ipp_disk is created if it doesn't exist + my $neb_uri = shift; + my $ipp_disk= $ipprc->file_resolve( $neb_uri , 1); + my $remote_disk = $ipp_disk; + + unless(defined($ipp_disk)) { + my_die(); + } + + $remote_disk =~ s%^.*/%%; # Remove nebulous path + $remote_disk =~ s%^\d+\.%%; # Remove ins_id + $remote_disk =~ s%:%/%g; # Replace colons with directories + $remote_disk = "${remote_root}/${remote_disk}"; + return($ipp_disk,$remote_disk); +} + +sub uri_to_outputs { # (ipp_disk,remote_disk) = uri_to_output(neb_uri); Appends to TRANSFER and CHECK filehandles + my $neb_uri = shift; + my ($ipp_disk, $remote_disk) = uri_convert( $neb_uri ); + + unless (exists($file_filter{$neb_uri})) { + $file_filter{$neb_uri} = 1; + print TRANSFER "$ipp_disk\n"; + print CHECK "$remote_disk\n"; + } + return($ipp_disk,$remote_disk); +} + +sub uri_to_outputs_for_return { # (ipp_disk,remote_disk) = uri_to_outputs_for_return(neb_uri); create ipp_disk, append to RETURN and GENERATE + my $neb_uri = shift; + my ($ipp_disk, $remote_disk) = uri_convert_and_create( $neb_uri ); + unless (exists($file_filter{$neb_uri})) { + $file_filter{$neb_uri} = 1; + print RETURN "$ipp_disk\n"; + print GENERATE "$remote_disk\n"; + } + return($ipp_disk,$remote_disk); +} + +sub uri_local_to_remote { #(remote_uri) = uri_local_to_remote(local_neb_uri); + # This needs to replace the nebulous tag with the remote root. + my $local_uri = shift; + $local_uri =~ s%^.*?/%%; # neb:/ + $local_uri =~ s%^.*?/%%; # / + $local_uri =~ s%^.*?/%%; # @HOST@.0/ + my $remote_uri = "${remote_root}/" . $local_uri; + + return($remote_uri); +} + +sub uri_remote_to_local { #(local_neb_uri) = uri_remote_to_local(remote_uri); + # This needs to replace the remote root directory with the nebulous tag. + my $remote_uri = shift; + $remote_uri =~ s%${remote_root}%%; + my $local_uri = "neb:///" . $remote_uri; + + return($local_uri); +} + + +sub my_die { # exit with status; my_die(message,stage_id,exit_code,exit_status); + my $msg = shift; + my $id = shift; + my $exit_code = shift; + my $exit_state = shift; + + $exit_code = $PS_EXIT_PROG_ERROR unless defined $exit_code; + + carp($msg); + + if (defined $id and not $no_update) { + my $command = "remotetool -updaterun -stage_id $id"; + $command .= " -fault $exit_code " if defined $exit_code; + $command .= " -set_state $exit_state " if defined $exit_state; + $command .= " -dbname $dbname " if defined $dbname; + + system($command); + } + + exit($exit_code); +} Index: /trunk/ippScripts/scripts/sc_remote_exec.pl =================================================================== --- /trunk/ippScripts/scripts/sc_remote_exec.pl (revision 36843) +++ /trunk/ippScripts/scripts/sc_remote_exec.pl (revision 36844) @@ -16,5 +16,5 @@ # Hard coded values my $DMZ_HOST = 'wtrw'; -my $SEC_HOST = 'mu-fe'; +my $SEC_HOST = 'mu-fe4'; my $IPP_PATH = '/turquoise/usr/projects/cosmo/mswarren/ipp/'; my $remote_root = '/scratch3/watersc1/'; @@ -24,5 +24,5 @@ my $ssh = can_run('ssh') or (warn "Can't find ssh" and $missing_tools = 1); my $scp = can_run('scp') or (warn "Can't find scp" and $missing_tools = 1); -my $remotetool = can_run('echo') or (warn "Can't find remotetool" and $missing_tools = 1); +my $remotetool = can_run('remotetool') or (warn "Can't find remotetool" and $missing_tools = 1); if ($missing_tools) { @@ -72,4 +72,6 @@ my $uri_check = $path_base . ".check"; my $uri_config = $path_base . ".config"; + my $uri_generate= $path_base . ".generate"; + my $uri_return = $path_base . ".return"; my $disk_command = $ipprc->file_resolve($uri_command); @@ -77,4 +79,6 @@ my $disk_check = $ipprc->file_resolve($uri_check); my $disk_config = $ipprc->file_resolve($uri_config); + my $disk_generate= $ipprc->file_resolve($uri_generate); + my $disk_return = $ipprc->file_resolve($uri_return); scp_put($disk_command,uri_local_to_remote($uri_command)); @@ -82,52 +86,16 @@ scp_put($disk_check,uri_local_to_remote($uri_check)); scp_put($disk_config,uri_local_to_remote($uri_config)); + scp_put($disk_generate,uri_local_to_remote($uri_generate)); + scp_put($disk_return, uri_local_to_remote($uri_return)); # Run check command my (undef,$remote_check) = uri_convert($uri_check); - my $ssh_check_stdout = ssh_exec_command("${remote_root}/sc_validate_files.pl --check_list $remote_check"); - -# Parse stdout for files to transfer. - if ($#{ $ssh_check_stdout } != -1) { # There were files not found. - # Someone was missing. Use the check and transfer lists to identify the local version, and send it. - # What was missing? - my %missing_files = (); - foreach my $l (split /\n/, (join '', @$ssh_check_stdout)) { - print "Missing!: $l\n"; - $missing_files{$l} = 1; - } - - # Do we have that file? - my @check_files = (); - open(CHECK,$disk_check); - while() { - chomp; - push @check_files, $_; - } - close(CHECK); - - my @trans_files = (); - open(TRANS,$disk_transfer); - while() { - chomp; - push @trans_files, $_; - } - close(TRANS); - - if ($#trans_files != $#check_files) { &my_die("The transfer file list and check file list are different sizes", - $remote_id,$PS_EXIT_DATA_ERROR); } - - for (my $i = 0; $i <= $#check_files; $i++) { - if (exists($missing_files{$check_files[$i]})) { - if ($trans_files[$i] ne "REMOTE_ONLY") { - scp_put($trans_files[$i],"$check_files[$i]"); - } - else { - &my_die("There is a file that should be there, but the file isn't there, and I don't have a copy of that file here. The file is: $check_files[$i]", - $remote_id,$PS_EXIT_DATA_ERROR); # This seems a bit harsh. - } - } - } - } - + my (undef,$remote_transfer)= uri_convert($uri_transfer); +# my $ssh_check_stdout = ssh_exec_command("${remote_root}/sc_validate_files.pl --check_list $remote_check --transfer_list $remote_transfer"); + my $ssh_check_stdout = ssh_exec_command("${remote_root}/sc_transfer_tool.pl --input $remote_check --fetch $remote_transfer"); + # We no longer need to parse this output, as it retrieves files itself. + foreach my $l (split /\n/, (join '', @$ssh_check_stdout)) { + print "$l\n"; + } # Run real command @@ -147,10 +115,10 @@ my $command = "$remotetool -updaterun -remote_id $remote_id "; $command .= " -set_state run "; - $command .= " -set_job_id $job_id "; + $command .= " -job_id $job_id "; $command .= " -dbname $dbname " if defined $dbname; system($command); -} +} # End of exec phase # Check that we have a valid job_id, either from the command line, or from the msub command. @@ -185,55 +153,48 @@ print "Stopped Polling code. $poll_count of $poll_max $poll_response\n"; } +if ($poll_response != 1) { + &my_die("Job hasn't completed. Ending polling.", $remote_id, 0, "run", $job_id); +} # Retrieve validation summary my $fault = 0; -exit(1); -# Read this as a metadata object -my $ssh_exec_stdout = ssh_exec_command("sc_validate_processing.pl ${path_base}.out"); -my $retrieveData = $mdcParser->parse(join "", @$ssh_exec_stdout); -my $retrieveData2= parse_md_list($retrieveData); - -foreach my $retrieveEntry (@$retrieveData2) { - my $process_fault = $retrieveEntry->{fault}; - my $dbcommand = $retrieveEntry->{dbcommand}; - my $filelist = $retrieveEntry->{filelist}; - - scp_get("${filelist}",${filelist}); - open(FL,${filelist}); - while() { - chomp; - my $f = $_; - scp_get("${f}",$f); - } - close(FL); - - my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = - run(command => $dbcommand, verbose => $verbose); - unless ($success) { - $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); - &my_die("The database command returned did not execute correctly. $dbcommand $process_fault", - $remote_id,$error_code); - } - - if ($process_fault != 0) { - $fault = 1; - } -} - -if (($policy ne 'ignore')&&($fault != 0)) { - &my_die("This job exited without all components finishing successfully. Halting.", - $remote_id,$fault,"halt"); -} -else { - print "This job exited without all components finishing successfully. $fault\n" if $fault; - &my_die("This job finished. Setting remote job to full.", - $remote_id,0,"full"); -} - -# Promote to finished. -&my_die(); # Not really, but I don't have the tool interface finished, so these are all just placeholders. -# Quick review: -# new -> pending -> run -> full -# auth +# Initialize the remote side to sync things + +# Once sync completes, execute all the database commands +# Grab the return list +my $uri_return = $path_base . ".return"; +my ($ipp_return,$remote_return) = uri_convert($uri_return); +scp_get($remote_return,$ipp_return); +# Feed teh return list into the +my $return_push_output = ssh_exec_command("${remote_root}/sc_transfer_tool.pl --input $remote_return"); +print "$return_push_output\n"; + + +open(RETURN, "$ipp_return") || &my_die("Couldn't open file? $ipp_return", $remote_id,$PS_EXIT_SYS_ERROR); +while() { + chomp; + if ($_ =~ /dbinfo/) { + my $file = $_; +# $file =~ s/${remote_root}.tmp//; THis shouldn't be needed, as the transfer tool munges the files on that side. + open(DBF,"$file") || warn "Missing file $file\n"; + my $cmd = ; + chomp($cmd); + $cmd =~ s/-/ -/g; # This is just a safety check for missing space. + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $cmd, verbose => $verbose); + unless ($success) { + # This shouldn't fail, but we also can't suddenly abort if something does fail. Maybe retry if this looks to be a continuing issue? + $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); + warn("The command that shouldn't fail has failed: $cmd $remote_id $error_code");##,$remote_id,$error_code); + } + close(DBF); + } +} +close(RETURN); + +# We're done, so set the state and exit. +&my_die("Finished",$remote_id,0,"full"); + + # END PROGRAM @@ -294,5 +255,5 @@ unless ($success) { $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); - &my_die(); + &my_die("Failed to execute command: >>$cmd<<"); } return ($stdout_buf); @@ -368,5 +329,5 @@ my $exit_code = shift; my $exit_state = shift; - + my $jobid = shift; $exit_code = $PS_EXIT_PROG_ERROR unless defined $exit_code; @@ -374,6 +335,7 @@ if (defined $id and not $no_update) { - my $command = "$remotetool -updaterun -stage_id $id"; + my $command = "$remotetool -updaterun -remote_id $id"; $command .= " -fault $exit_code " if defined $exit_code; + $command .= " -job_id $job_id " if defined $jobid; $command .= " -set_state $exit_state " if defined $exit_state; $command .= " -dbname $dbname " if defined $dbname; @@ -384,2 +346,53 @@ exit($exit_code); } + +# I don't think I need this anymore, but I don't want to burn it until I"m sure. + +# exit(1); +# # Read this as a metadata object +# my $ssh_exec_stdout = ssh_exec_command("sc_validate_processing.pl ${path_base}.out"); +# my $retrieveData = $mdcParser->parse(join "", @$ssh_exec_stdout); +# my $retrieveData2= parse_md_list($retrieveData); + +# foreach my $retrieveEntry (@$retrieveData2) { +# my $process_fault = $retrieveEntry->{fault}; +# my $dbcommand = $retrieveEntry->{dbcommand}; +# my $filelist = $retrieveEntry->{filelist}; + +# scp_get("${filelist}",${filelist}); +# open(FL,${filelist}); +# while() { +# chomp; +# my $f = $_; +# scp_get("${f}",$f); +# } +# close(FL); + +# my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = +# run(command => $dbcommand, verbose => $verbose); +# unless ($success) { +# $error_code = (($error_code >> 8) or $PS_EXIT_PROG_ERROR); +# &my_die("The database command returned did not execute correctly. $dbcommand $process_fault", +# $remote_id,$error_code); +# } + +# if ($process_fault != 0) { +# $fault = 1; +# } +# } + +# if (($policy ne 'ignore')&&($fault != 0)) { +# &my_die("This job exited without all components finishing successfully. Halting.", +# $remote_id,$fault,"halt"); +# } +# else { +# print "This job exited without all components finishing successfully. $fault\n" if $fault; +# &my_die("This job finished. Setting remote job to full.", +# $remote_id,0,"full"); +# } + +# Promote to finished. +#&my_die(); # Not really, but I don't have the tool interface finished, so these are all just placeholders. +# Quick review: +# new -> pending -> run -> full +# auth Index: /trunk/ippScripts/scripts/sc_transfer_tool.pl =================================================================== --- /trunk/ippScripts/scripts/sc_transfer_tool.pl (revision 36844) +++ /trunk/ippScripts/scripts/sc_transfer_tool.pl (revision 36844) @@ -0,0 +1,159 @@ +#!/usr/bin/env perl + +use Carp; +use warnings; +use strict; +use Getopt::Long qw( GetOptions :config auto_help auto_version gnu_getopt ); +use Pod::Usage qw( pod2usage ); +use File::Basename; +use IPC::Cmd 0.36 qw( can_run run ); +use Sys::Hostname; + +my $local_raw = '/scratch3/watersc1/raw/'; +my $local_tmp = '/scratch3/watersc1/tmp/'; +my $threads = 8; +my @hosts = ('ippc20.ipp.ifa.hawaii.edu','ippc24.ipp.ifa.hawaii.edu','ippc28.ipp.ifa.hawaii.edu', + 'ippc21.ipp.ifa.hawaii.edu','ippc25.ipp.ifa.hawaii.edu','ippc29.ipp.ifa.hawaii.edu', + 'ippc22.ipp.ifa.hawaii.edu','ippc26.ipp.ifa.hawaii.edu', + 'ippc23.ipp.ifa.hawaii.edu','ippc27.ipp.ifa.hawaii.edu'); +my $input_file; +my $verbose = 0; +my $fetch = 0; + +GetOptions( + 'threads=s' => \$threads, + 'input=s' => \$input_file, + 'fetch=s' => \$fetch, + 'verbose' => \$verbose, + ) or pod2usage( 2 ); +pod2usage( -msg => "Unknown option: @ARGV", -exitval => 2) if @ARGV; +pod2usage( -msg => "Required options: --input", -exitval => 3) unless + defined($input_file); + +my $hostname = hostname; +print STDERR "Running on $hostname\n"; + +# split in input file list +unless(-d $local_tmp) { system("mkdir -p $local_tmp"); } +unless(-d $local_raw) { system("mkdir -p $local_raw"); } + +open(I,$input_file) || die "Couldn't find input file specified\n"; +if ($fetch) { + open(F,$fetch) || die "Couldn't find fetch file specified\n"; +} +my $input_base = basename($input_file); +my @filehandles; +my $i; +for ($i = 0; $i < $threads; $i++) { + open($filehandles[$i], ">${local_tmp}/${input_base}.${i}"); +} +$i = 0; +while() { + chomp; + my $fline; + if ($fetch) { + $fline = ; + } + if (($fetch)&&(!(-e $_))) { # We are fetching, and do not already have this file. + print { $filehandles[$i] } $fline; + $i++; + } + elsif (!($fetch)) { # We are pushing +# if (-e "${local_tmp}/$_") { + print { $filehandles[$i] } "${local_tmp}/$_" . "\n"; + $i++; +# } +# else { # But somehow couldn't find the file we expected +# print STDERR "Expected file ${local_tmp}/$_ not found!\n"; +# die; +# } + } + if ($i >= $threads) { $i = 0; } +} +close(I); + +for ($i = 0; $i < $threads; $i++) { + close($filehandles[$i]); +} + +# fork the tars +my @pids = (); +for ($i = 0; $i < $threads; $i++) { + $pids[$i] = fork(); + if ($pids[$i] == 0) { + my $host = $hosts[rand($#hosts)]; + my $code; + if ($fetch) { + $code = fetch_task($host,"${input_base}.${i}", 0); + } + else { + $code = transfer_task($host,"${local_tmp}/${input_base}.${i}", 0); + } + exit($code); + } +} +for ($i = 0; $i < $threads; $i++) { + waitpid($pids[$i],0); +} +# die; +# if ($fetch) { +# system("parallel -j $threads ln -sf ${local_tmp}/{2} {1} :::: $input_file :::: $fetch"); +# } + +for ($i = 0; $i < $threads; $i++) { +# unlink("${local_tmp}/${input_base}.${i}"); +} + + +# distribute bundles to nodes + +# execute bundle transfer + +sub transfer_task { + my $destination_host = shift; + my $transfer_filelist= shift; + my $error = shift; + + my $command = "tar cf - --dereference --files-from=${transfer_filelist} | /usr/projects/cosmo/amd6100/bin/ssh -o NoneSwitch=yes -o NoneEnabled=yes $destination_host tar xf - -C /data/ --transform '" . 's,^.\*/data/,,' . "' --dereference"; # the transform bit is there because it looks like the ' gets dropped, so the * is interpreted, and why is our tar so out of date? + print STDERR "$command\n"; +# die; + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => $verbose); + unless ($success) { + $error_code = (($error_code >> 8) or 4); + warn("Transfer of $transfer_filelist to $destination_host failed with error $error_code."); + $error++; + if ($error < 4) { + $error_code = transfer_task($destination_host,$transfer_filelist,$error); + } + else { + die("Failed too many times $error $destination_host $transfer_filelist"); + } + } + return($error_code); +} + +sub fetch_task { + my $destination_host = shift; + my $transfer_filelist= shift; + my $error = shift; + + system("/usr/projects/cosmo/amd6100/bin/scp -o NoneSwitch=yes -o NoneEnabled=yes ${local_tmp}/${transfer_filelist} ${destination_host}:/tmp/"); + my $command = "/usr/projects/cosmo/amd6100/bin/ssh -o NoneSwitch=yes -o NoneEnabled=yes $destination_host tar cf - --dereference --files-from=/tmp/${transfer_filelist} | tar xf - -C ${local_raw} --skip-old-files --warning=existing-file --dereference "; + print STDERR "$command\n"; + + my ( $success, $error_code, $full_buf, $stdout_buf, $stderr_buf ) = + run(command => $command, verbose => $verbose); + unless ($success) { + $error_code = (($error_code >> 8) or 4); + warn("Transfer of $transfer_filelist to $destination_host failed with error $error_code."); + $error++; + if ($error < 4) { + $error_code = fetch_task($destination_host,$transfer_filelist,$error); + } + else { + die("Failed too many times $error $destination_host $transfer_filelist"); + } + } + return($error_code); +}