root/MPI/mpi_evaluator

#!/usr/bin/perl -w

eval 'exec /usr/bin/perl -w -S $0 ${1+"$@"}'
    if 0; # not running under some shell

use strict "vars";
use strict "refs";

use FindBin;
use lib "$FindBin::Bin/../lib";
use lib "$FindBin::Bin/../perl/lib";
use vars qw($RANK $LOG $CMD_ARGS);
use Storable qw(freeze thaw);
use threads;
use threads::shared;

BEGIN{
   $main::eva = 1; #signals that this is evaluator

   if (not ($ENV{CGL_SO_SOURCE})) {
      $ENV{CGL_SO_SOURCE} = "$FindBin::Bin/../lib/CGL/so.obo";
   }
   if (not ($ENV{CGL_GO_SOURCE})) {
      $ENV{CGL_GO_SOURCE} = "$FindBin::Bin/../lib/CGL/gene_ontology.obo"
   }
   
   $CMD_ARGS = join(' ', @ARGV);
   
   #what to do on ^C
   $SIG{'INT'} = sub {
      print STDERR "\n\nAborted by user!!\n\n";
      my @threads = threads->list;
      foreach my $thr (@threads){
         $thr->detach;
      }
      exit (1);
   };    
   
   #supress warnings from storable module
   $SIG{'__WARN__'} = sub {
      warn $_[0] if ( $_[0] !~ /Not a CODE reference/ &&
                      $_[0] !~ /Can\'t store item CODE/
                    );
   };

   #output to log file of seq that caused rank to die
   $SIG{'__DIE__'} =
   sub {
      if (defined ($LOG) && defined $_[0]) {
         my $die_count = $LOG->get_die_count();
         $die_count++;
         
         $LOG->add_entry("DIED","RANK",$RANK);
         $LOG->add_entry("DIED","COUNT",$die_count);
      }

      my @threads = threads->list;
      foreach my $thr (@threads){
         $thr->detach;
      }

      die "#----------------------\n",
          "FATAL: failed!!\n",
          "#----------------------\n",
          $_[0] . "\n";
   };
}

use Cwd;
use Storable;
use FileHandle;
use File::Path;
use Getopt::Long qw(:config no_ignore_case);
use File::Temp qw(tempfile tempdir);
use Bio::DB::Fasta;
use GI;
use Dumper::GFF::GFFV3;
use Iterator::Any;
use Iterator::Fasta;
use Iterator::GFF3;
use Fasta;
use FastaChunker;
use maker::auto_annotator;
use cluster;
use repeat_mask_seq;
use runlog;
use ds_utility;
use GFFDB;
use Error qw(:try);
use Error::Simple;
use Process::MpiChunk;
use Process::MpiTiers;
use Parallel::MPIcar qw(:all);

unless($threads::VERSION >= 1.67){
   die "Program requires threads version 1.67 or greater\n",
       "You have version ". $threads::VERSION ."\n";
}

#--MPI_Init requires there to be arguments on @ARGV
#--This is a logic problem by the Package Authors
#--This is a hack to solve the problem
if (not @ARGV) {
   push (@ARGV, 'null');
   MPI_Init();                  #initiate the MPI
   shift @ARGV;
}
else {
   MPI_Init();                  #initiate the MPI
}

$| = 1;

my $usage = "
Usage:

     mpi_evaluator [options] <eval_opts> <eval_bopts> <eval_exe>


Options:

     -genome_gff <file>  Specify the maker gff file to evaluate.

     -model_gff  <file>  Specify the external gff file to evaluate.

     -genome     <file>  Specify the genome fasta file.  This if optional if the
                         fasta entries are also found in the gff file.

     -RM_off|R           Turns all repeat masking off.

     -retry   <integer>  Rerun failed contigs up to the specified count.

     -cpus|c  <integer>  Tells how many cpus to use for BLAST analysis.

     -force|f            Forces program to delete old files before running again.
                         This will require all blast analyses to be rerun.

     -again|a            Caculate all output files again even if no settings have
                         changed.

     -quiet|q            Silences most of the status messages.

     -CTL                Generate empty control files in the current directory.

     -help|?             Prints this usage statement.


";

#-------------------------------------------------------------------------------
#------------------------------------ MAIN -------------------------------------
#-------------------------------------------------------------------------------

#--set object variables for serialization of data
$Storable::forgive_me = 1; #allows serializaion of objects with code refs

#------INITIATE MPI VARIABLES------
my $rank = MPI_Comm_rank(MPI_COMM_WORLD); #my proccess number
my $size = MPI_Comm_size(MPI_COMM_WORLD); #how many proccesses
$RANK = $rank;

#MPI SIGNAL CODES
#--mpi message tags
my $who_I_am       = 1111;
my $what_I_want    = 2222;
my $result_status  = 3333;
my $request_status = 4444;
my $c_res_status   = 5555;
my $chunk_status   = 6666;
my $work_order     = 7777; #generic data tag
my $mpi_data       = 8888;
my $message_length = 9999;

#--what_I_want type signals
my $need_tier   = 1;
my $need_helper = 2;
my $have_c_res  = 3;
my $need_c_res  = 4;

#--request_status signals
my $wait_as_helper = 1;
my $yes_tier       = 2;
my $yes_helper     = 3;
my $no_helper      = 4;
my $go_chunk       = 5;
my $reset          = 6;
my $terminate      = 0;

#--results_status signals
my $yes_result = 1;
my $no_result  = 0;

#--c_res_status signal
my $yes_c_res      = 1;
my $no_c_res      = 0;

#--chunk_status signals
my $yes_chunk = 1;
my $no_chunk  = 0;

#---variables for thread and the root node
my @c_results;
my @failed;
my @res_loc;
my @helper_stack;
my @active;
my @chunks : shared;
my @returned_chunks :shared;
my $t_need_flag :shared;
my $t_tier :shared;
my $t_tier_result :shared;
my $t_chunk :shared;
my $t_chunk_result :shared;
my $t_terminate :shared;

#---global variables
my %OPT;
my $root = 0; #define root node (only changed for debugging)

#---Process options on the command line 
try{
    GetOptions("RM_off|R" => \$OPT{R},
               "force|f" => \$OPT{force},
               "genome_gff=s" => \$OPT{genome_gff},
               "genome=s" => \$OPT{genome},
               "model_gff=s" => \$OPT{model_gff},
               "cpus|c=i" => \$OPT{cpus},
               "retry=i" =>\$OPT{retry},
               "again|a" =>\$OPT{again},
               "quiet" =>\$main::quiet,
               "CTL" => sub {GI::generate_control_files() if($rank == $root); MPI_Finalize(); exit(0);},
               "help|?" => sub {print $usage if($rank == $root); MPI_Finalize(); exit(0)}
               );
}
catch Error::Simple with{
    my $E = shift;
    
    print STDERR $E->{-text};
    die "\n\nFailed parsing command line options!!\n\n";
};

#--------------------------------------
#---------PRIMARY MPI PROCCESS---------
#--------------------------------------

#--check if root node
if ($rank == $root) {
   #varibles that are persistent outside of try
   my %CTL_OPT;
   my $iterator;
   my $DS_CTL;
   my $GFF_DB;
   my $build;

   try{
      #get arguments off the command line
      my @ctlfiles = @ARGV;
      
      if (not @ctlfiles) {
         if (-e "eval_opts.ctl" &&
             -e "eval_bopts.ctl" &&
             -e "eval_exe.ctl"
            ) {
            
            @ctlfiles = ("eval_opts.ctl",
                         "eval_bopts.ctl",
                         "eval_exe.ctl"
                        );
         }
         else {
            print STDERR  "ERROR: Control files not found\n";
            print $usage;
            exit(0);
         }
      }
      
      #--Control file processing
      
      #set up control options from control files
      %CTL_OPT = GI::load_control_files(\@ctlfiles, \%OPT, $size);
      
      #--open datastructure controller
      $DS_CTL = ds_utility->new(\%CTL_OPT);
      
      #--set up gff database
      $GFF_DB = new GFFDB(\%CTL_OPT);
      $build = $GFF_DB->next_build;
      
      #---load genome multifasta/GFF3 file
      $iterator = new Iterator::Any( -fasta => $CTL_OPT{'genome'},
                                     -gff => $CTL_OPT{'genome_gff'},
                                   );
   }
   catch Error::Simple with{
      my $E = shift;
      print STDERR $E->{-text};
      print STDERR "\n\nFailed while examining startup data\n",
                   "(control files and input fasta files)!!\n\n";
      my $code = 2;
      $code = $E->{-value} if (defined($E->{-value}));
      
      exit($code);
   };

   #build indexes of databases
   #Shared_Functions::build_all_indexes($CTL_OPT{old_protein},
   #                                   $CTL_OPT{old_est}
   #                                  );

   #====ACTUAL MPI COMMUNICATION
   
   #---main code for distribution of mpi data starts here

   #thread for root node to other things than just manage mpi
   my $thr = threads->create(\&node_thread);
   my $go_mpi_status = 1;
   $t_need_flag = 1;

   while($go_mpi_status){
      #====INTERNAL TIER THREAD
      #check on results from internal thread
      if (defined($t_tier_result)){
         my $t_res = ${thaw($t_tier_result)};
         $t_tier_result = undef;
         $active[$root] = 0;

         $DS_CTL->add_entry($t_res->{-DS});

         if ($t_res->{-failed}){
            push(@failed, $t_res->{-fasta});
         }
      }
      if (defined($t_chunk_result)){
         my $chunk =  ${thaw($t_chunk_result)};
         $t_chunk_result = undef;
         my $id = $chunk->id();
         ($id) = split (":", $id);
         push (@{$c_results[$id]}, $chunk);
         unshift (@{$res_loc[$id]}, $root);
      }

      #see if there are chunks to get from the internal thread
      while((@helper_stack > 0) && (@chunks > 0) && (my $chunk = shift @chunks)){
         my $helper = shift @helper_stack;
         $chunk = ${thaw($chunk)};
         
         #tell helper node I need help
         MPI_Send(\$rank, 1,  MPI_INT, $helper, $who_I_am, MPI_COMM_WORLD);
         
         #tell helper node a chunk is coming
         MPI_Send(\$go_chunk, 1, MPI_INT, $helper, $request_status, MPI_COMM_WORLD );
         
         #send the chunk
         MPI_SendII(\$chunk, $helper, $mpi_data, MPI_COMM_WORLD);
      }

      #get tier for internal thread
      if($t_need_flag > 0){
         my $tier;
         while (my $fasta = $iterator->nextFasta() || shift @failed){
            $tier = Process::MpiTiers->new({fasta =>$fasta,
                                            CTL_OPT => \%CTL_OPT,
                                            DS_CTL  => $DS_CTL,
                                            GFF_DB  => $GFF_DB,
                                            build   => $build},
                                           $root
                                          );

            last if(! $tier->terminated);
         }
         if(defined $tier && ! $tier->terminated){
            $t_need_flag = 0;
            my $t_val = freeze(\$tier);
            $t_tier = $t_val;
            $active[$root] = 1;
         }
         else{
            $t_need_flag = 2; #take tier or chunk
         }
      }


      #take a node out of limbo if there are failed contigs
      if(@helper_stack && @failed){
         my $helper = shift @helper_stack;

         #tell helper node who I am
         MPI_Send(\$rank, 1,  MPI_INT, $helper, $who_I_am, MPI_COMM_WORLD);
         
         #tell helper node that no chunk is coming (resets to ask for tier)
         MPI_Send(\$reset, 1, MPI_INT, $helper, $request_status, MPI_COMM_WORLD );
      }


      #work with mpi nodes or skip mpi if all nodes are waiting in limbo
      if (@helper_stack < $size - 1){
         my $who;
         my $what;
         my $rs_type;
         
         #see who asks for a file
         MPI_Recv(\$who, 1,  MPI_INT, -2, $who_I_am, MPI_COMM_WORLD);
         
         #see what the mpi node wants
         MPI_Recv(\$what, 1, MPI_INT, $who, $what_I_want, MPI_COMM_WORLD);
         
         #if the node wants a tier to process, do this
         if($what == $need_tier){
            #receive result status
            MPI_Recv(\$rs_type, 1,  MPI_INT, $who, $result_status, MPI_COMM_WORLD); 
            
            #get result if available
            if($rs_type == $yes_result){
               my $result;
               MPI_RecvII(\$result, $who, $mpi_data, MPI_COMM_WORLD);
               $DS_CTL->add_entry($result->{-DS});

               if ($result->{-failed}){
                  push(@failed, $result->{-fasta});
               }
            }
            
            #if a contig is available send tier
            my $tier;

            while (my $fasta = $iterator->nextFasta() || shift @failed){
               $tier = Process::MpiTiers->new({fasta => $fasta,
                                               CTL_OPT => \%CTL_OPT,
                                               DS_CTL  => $DS_CTL,
                                               GFF_DB  => $GFF_DB,
                                               build   => $build},
                                              $who
                                             );

               last if(! $tier->terminated);
            }
            if(defined $tier && ! $tier->terminated){
               #say tier is available and send it
               MPI_Send(\$yes_tier, 1, MPI_INT, $who, $request_status, MPI_COMM_WORLD);
               MPI_SendII(\$tier, $who, $mpi_data, MPI_COMM_WORLD );
               $active[$who] = 1;
            }
            else{
               MPI_Send(\$wait_as_helper, 1, MPI_INT, $who, $request_status, MPI_COMM_WORLD);
               push(@helper_stack, $who);
               $active[$who] = 0;
            }
         }
         #if the node wants a helper or needs a chunk result, do this
         elsif($what == $need_helper || $what == $need_c_res){
            #--first send c_res_status
            # send ids of nodes with chunk results
            if(defined ($res_loc[$who])){
               MPI_Send(\$yes_c_res, 1, MPI_INT, $who, $c_res_status, MPI_COMM_WORLD);
               MPI_SendII(\$res_loc[$who], $who, $mpi_data, MPI_COMM_WORLD);
               
               my @locs = @{$res_loc[$who]};
               $res_loc[$who] = undef;
               
               #if primary node has chunk result to send then send them
               while (defined(my $loc = shift @locs)){
                  if ($loc == $root){
                     my $res = shift @{$c_results[$who]};
                     MPI_SendII(\$res, $who, $mpi_data, MPI_COMM_WORLD);
                  }
               }
            }
            #no one has anything yet
            else{
               MPI_Send(\$no_c_res, 1, MPI_INT, $who, $c_res_status, MPI_COMM_WORLD);
            }
            
            #continue the rest if the node needs a helper
            if($what == $need_helper){
               #find the number of helpers required
               my $num_helpers_req;
               MPI_Recv(\$num_helpers_req, 1, MPI_INT, $who, $work_order, MPI_COMM_WORLD);
               
               #number of secondary node helpers available
               my $sec_node_avail = @helper_stack;
               
               #number of primary node threads available
               my $thr_avail = ($t_need_flag == 2 && ! defined $t_chunk) ? 1 : 0;
               
               #signal that no helpers are available
               if($sec_node_avail == 0 && $thr_avail == 0){
                  MPI_Send(\$no_helper, 1, MPI_INT, $who, $request_status, MPI_COMM_WORLD);
               }
               else{#if node helpers are available
                  #helpers to send
                  my $helpers = [];
                  
                  #secondary node helpers
                  if($sec_node_avail > 0){
                     #seperate the helpers
                     while(@{$helpers} < $num_helpers_req && @helper_stack > 0){
                        my $helper = shift @helper_stack;
                        push(@{$helpers}, $helper);
                     }
                     
                     $num_helpers_req -= @{$helpers};
                  }
                  
                  #primary node thread helper
                  my $root_helper_flag = 0;
                  if ($thr_avail && $num_helpers_req > 0){
                     my $helper = $root;
                     #aways make root node first
                     unshift(@{$helpers}, $helper);
                     $root_helper_flag = 1;
                  }
                  
                  #say helper is available and send ids of the helpers
                  MPI_Send(\$yes_helper, 1, MPI_INT, $who, $request_status, MPI_COMM_WORLD);
                  MPI_SendII(\$helpers, $who, $mpi_data, MPI_COMM_WORLD);
               
                  #take chunk as a helper
                  if($root_helper_flag){
                     #see who's one who needs help
                     my $who2;
                     MPI_Recv(\$who2, 1,  MPI_INT, $who, $who_I_am, MPI_COMM_WORLD);
                     
                     #get go_chunk request_status
                     my $req_stat;
                     MPI_Recv(\$req_stat, 1, MPI_INT, $who2, $request_status, MPI_COMM_WORLD );
                     if ($req_stat == $go_chunk){
                        #get the chunk
                        my $chnk;
                        MPI_RecvII(\$chnk, $who2, $mpi_data, MPI_COMM_WORLD);
                        
                        $t_need_flag = 0;
                        $t_chunk = freeze(\$chnk);
                     }
                     elsif($req_stat == $reset){
                        #do nothing
                     }
                     else{
                        die "ERROR: Logic error in getting chunk as a helper\n";
                     }
                  }
               }
            }
         }
         #if the node has a chunk result, do this
         elsif($what == $have_c_res){
            #get the owner of the result
            my $owner;
            MPI_Recv(\$owner, 1, MPI_INT, $who, $work_order, MPI_COMM_WORLD);
            
            if($owner == $root){#if root is owner get result
               my $chunk_res;
               MPI_RecvII(\$chunk_res, $who, $mpi_data, MPI_COMM_WORLD);
               push(@returned_chunks, freeze(\$chunk_res));
            }
            else{#take note of owner to tell him he has a result waiting
               push(@{$res_loc[$owner]}, $who);
            }
         }
         #if what the node wants is something else
         else{
            die "ERROR: Invalid request type\n";
         }
      }
      else{ #take a break if mpi was skipped
         #this keeps the root node from hogging resources if only the thread is active
         sleep 2;
      }

      #see if all contigs are finished
      $go_mpi_status = 0;
      foreach my $n (@active ){
         if(@helper_stack < $size - 1){
            $go_mpi_status = 1;
            last;
         }
         if((defined($n) && $n == 1)){
            $go_mpi_status = 1;
            last;
         }
      }
      if(! $iterator->finished || @failed > 0){
          $go_mpi_status = 1;
      }
   }

   #---tell mpi nodes to terminate
   for(my $i = 1; $i < $size; $i++){
      #tell chunks waiting for helper who I am
      MPI_Send(\$rank, 1,  MPI_INT, $i, $who_I_am, MPI_COMM_WORLD);

      #send termination signal
      MPI_Send(\$terminate, 1, MPI_INT, $i, $request_status, MPI_COMM_WORLD);
   }
   
   #---release thread
   $t_terminate = 1; #signals to thread to clean up
   $thr->detach() unless($thr->is_detached);

   print STDERR "\n\nProgram is now finished!!!\n\n";
}
#------SECONDARY MPI PROCESSES------
else {
   my $go_mpi_status = 1;
   my $tier_result;
   my $tier;
   my $chunk_result;

   while ($go_mpi_status) {
      #tell the  primary process what node it is speaking to
      MPI_Send(\$rank, 1, MPI_INT, $root, $who_I_am, MPI_COMM_WORLD );

      #decide what this node needs
      my $what;
      my $chunk;

      if(defined $chunk_result){
         $what = $have_c_res;
      }
      elsif(!defined($tier) || $tier->terminated){
         $what = $need_tier;
         if(defined($tier)){
            #collect errors and failures if any
            $tier_result->{-error} = $tier->error;
            $tier_result->{-failed} = $tier->failed;
            $tier_result->{-DS} = $tier->DS;
            $tier_result->{-fasta} = $tier->fasta if($tier->failed);
            $tier = undef;
         }
      }
      elsif(($chunk = $tier->next_chunk) && ($tier->num_chunks > 0)){
         $what = $need_helper;
      }
      else{
         $what = $need_c_res;
      }

      #--tell primary node what this node needs
      MPI_Send(\$what, 1, MPI_INT, $root, $what_I_want, MPI_COMM_WORLD );

      #if what I want is a tier do this
      if($what == $need_tier){
         #Send result status
         my $rs_type = (defined($tier_result)) ? $yes_result: $no_result;
         MPI_Send(\$rs_type, 1, MPI_INT, $root, $result_status, MPI_COMM_WORLD );

         #Send result if available
         if($rs_type == $yes_result){
            MPI_SendII(\$tier_result, $root, $mpi_data, MPI_COMM_WORLD);
            $tier_result = undef;
         }

         #get request_status for the tier
         my $req_status;
         MPI_Recv(\$req_status, 1, MPI_INT, $root, $request_status, MPI_COMM_WORLD );

         #get tier and run if it if there is one
         if($req_status == $yes_tier){
            MPI_RecvII(\$tier, $root, $mpi_data, MPI_COMM_WORLD );
            $tier->run;
         }#wait as helper if asked to
         elsif($req_status == $wait_as_helper){
            #see who needs help
            my $who;
            MPI_Recv(\$who, 1,  MPI_INT, -2, $who_I_am, MPI_COMM_WORLD);

            #get request_status for chunk
            my $chunk_status;
            MPI_Recv(\$chunk_status, 1, MPI_INT, $who, $request_status, MPI_COMM_WORLD );

            #if there is a chunk do this
            if($chunk_status == $go_chunk){
               #get chunk to process
               my $chnk;
               MPI_RecvII(\$chnk, $who, $mpi_data, MPI_COMM_WORLD);

               #run chunk
               $chnk->run($rank);
               $chunk_result = $chnk;
            }
            #if the reset signal is received do this
            elsif($chunk_status == $reset){
               #do nothing
            }
            #if the terminate signal is received do this
            elsif($chunk_status == $terminate){
                  $go_mpi_status = 0;
                  last;
            }
            else{
               die "ERROR: Invalid chunk status signal\n;";
            }
         }
         else{
            die "ERROR: Invalid request status type\n";
         }
      }#if what I want is help or the result from a helper do this
      elsif ($what == $need_helper || $what == $need_c_res){
         # check c_result_status
         my $c_res_stat;
         MPI_Recv(\$c_res_stat, 1, MPI_INT, $root, $c_res_status, MPI_COMM_WORLD);

         #if there are chunk results, do this
         my $locs;
         if($c_res_stat == $yes_c_res){
            #get ids of nodes with chunk result
            MPI_RecvII(\$locs, $root, $mpi_data, MPI_COMM_WORLD);

            #get chunk results from only the root node for now
            foreach my $loc (@{$locs}){
               next if ($loc != $root);
               my $c_res;
               MPI_RecvII(\$c_res, $loc, $mpi_data, MPI_COMM_WORLD);
               $tier->update_chunk($c_res);
            }
         }

         #continue the rest if the node needs a helper
         if ($what == $need_helper){
            #send the number of helpers required
            my $num_helpers_req = $tier->num_chunks;
            MPI_Send(\$num_helpers_req, 1, MPI_INT, $root, $work_order, MPI_COMM_WORLD);
            
            #see if helper is available
            my $help_stat;
            MPI_Recv(\$help_stat, 1, MPI_INT, $root, $request_status, MPI_COMM_WORLD);
            
            if($help_stat == $yes_helper){
               my $helpers;
               MPI_RecvII(\$helpers, $root, $mpi_data, MPI_COMM_WORLD);
               
               #send chunk to helper
               foreach my $helper (@{$helpers}){
                  #say I'm the one who needs help
                  MPI_Send(\$rank, 1,  MPI_INT, $helper, $who_I_am, MPI_COMM_WORLD);
                  
                  #send go_chunk request_status
                  MPI_Send(\$go_chunk, 1, MPI_INT, $helper, $request_status, MPI_COMM_WORLD);
                  
                  #send chunk
                  my $chnk = $tier->next_chunk;
                  MPI_SendII(\$chnk, $helper, $mpi_data, MPI_COMM_WORLD);
               }
            }
         }

         #get chunk results from non root nodes since root comm has terminated
         foreach my $loc (@{$locs}){
            next if ($loc == $root);
            my $c_res;
            MPI_RecvII(\$c_res, $loc, $mpi_data, MPI_COMM_WORLD);
            $tier->update_chunk($c_res);
         }

         #run the chunk if there is one
         if(defined($chunk)){
            $chunk->run($rank);
            $tier->update_chunk($chunk);
            $tier->run();
            $chunk = undef;
         }
      }
      #if just finished a helper chunk, inform that it is finished
      elsif($what == $have_c_res){
         #send the owner id of the result
         my $owner = $chunk_result->id();
         ($owner) = split(":", $owner);
         MPI_Send(\$owner, 1, MPI_INT, $root, $work_order, MPI_COMM_WORLD);

         #send the result
         MPI_SendII(\$chunk_result, $owner, $mpi_data, MPI_COMM_WORLD);
         $chunk_result = undef;
      }
   }
}

#---------ALL NODES----------
MPI_Finalize();                 #terminate MPI

exit(0);

#-----------------------------------------------------------------------------
#----------------------------------- SUBS ------------------------------------
#-----------------------------------------------------------------------------
#other things for root node to do
#(thread allows root to process tiers like a secondary node)
sub node_thread {
   my $tier;
   my $chunk;
   $t_need_flag = 1;

   while(not $t_terminate){
      #load serialized tier into tier
      if(! defined ($tier) && defined ($t_tier)){
         $t_need_flag = 0;
         $tier = ${thaw($t_tier)};
         $t_tier = undef;
         next;
      }#process tier
      elsif(defined($tier)){
         #get chunk results from other nodes
         while(my $res = shift @returned_chunks){
            $res = ${thaw($res)};
            $tier->update_chunk($res);
         }

         #run the tier as far as possible
         $tier->run;

         #get all chunks available
         my $chnk = $tier->next_chunk;
         while(my $o_chnk = $tier->next_chunk){
            $o_chnk = freeze(\$o_chnk);
            push (@chunks, $o_chnk);
         }

         #run chunks one at a time
         $chnk->run($rank) if ($chnk);
         $tier->update_chunk($chnk) if ($chnk);
         while($chnk = shift @chunks){
            $chnk = ${thaw($chnk)};
            if($tier->failed){ #skip chunks after failure
                $tier->update_chunk($chnk);
                next;
            }

            $chnk->run($rank);
            $tier->update_chunk($chnk);
         }

         #let tier advance if possible
         $tier->run();

         #terminate tier, wait, or continue
         if($tier->terminated){
            my $tier_result;
            $tier_result->{-error} = $tier->error;
            $tier_result->{-failed} = $tier->failed;
            $tier_result->{-DS} = $tier->DS;
            $tier_result->{-fasta} = $tier->fasta if ($tier->failed);

            sleep 1 while (defined ($t_tier_result)); #pause incase result will be overwritten
            $t_tier_result = freeze(\$tier_result);
            $tier = undef;
            $t_need_flag = 1;
         }#take a break
         elsif($tier->num_chunks == 0){
            #keeps thread from hogging resources while waiting for external results
            sleep 1;
         }

         next;
      }#load serialized chunk into chunk
      elsif(! defined ($chunk) && defined ($t_chunk)){
         $t_need_flag = 0;
         $chunk = ${thaw($t_chunk)};
         $t_chunk = undef;
         next;
      }#process chunk
      elsif(defined($chunk)){
         $chunk->run($rank);
         sleep 1 while (defined ($t_chunk_result)); #pause incase result will be overwritten
         $t_chunk_result = freeze(\$chunk);
         $chunk = undef;
         $t_need_flag = 1;
         next;
      }#take a break
      else{
         #keeps thread form hogging resources when there is nothing to do
         sleep 1;
      }
   }
}
#----------------------------------------------------------------------------
#easy dump of string to a tempfile
sub totemp{
   my $data = shift @_;

   my ($fh, $name) = tempfile();
   print $fh $data;
   close ($fh);

   return $name;
}
#----------------------------------------------------------------------------
#sends scalar variable contents via serialization
#scalar can hold ref to other data structures
sub MPI_SendII{
   my $msg = shift @_;
   my $target = shift @_;
   my $tag = shift @_;
   my $communicator = shift @_;

   my $send = freeze($msg);
   my $length = length($send);

   MPI_Send(\$length, 1, MPI_INT, $target, $message_length, $communicator);
   MPI_Send(\$send, $length, MPI_CHAR, $target, $tag, $communicator);

}
#----------------------------------------------------------------------------
#receives serialized scalar variable
#scalar can hold ref to other data structures
sub MPI_RecvII{
   my $ref = shift @_;
   my $source = shift @_;
   my $tag = shift @_;
   my $communicator = shift @_;

   my $length;
   my $recv;


   MPI_Recv(\$length, 1, MPI_INT, $source, $message_length, $communicator);
   MPI_Recv(\$recv, $length, MPI_CHAR, $source, $tag, $communicator); #receive line

   ${$ref} = ${thaw($recv)};
}