Inheritance diagram for Bio::EnsEMBL::Hive::AnalysisJob:

Collaboration diagram for Bio::EnsEMBL::Hive::AnalysisJob:

Public Member Functions
public	input_id ()

public	param_id_stack ()

public	accu_id_stack ()

public	role_id ()

public	status ()

public	retry_count ()

public	when_completed ()

public	runtime_msec ()

public	query_count ()

public	semaphore_count ()

public	semaphored_job_id ()

public	set_and_update_status ()

public	stdout_file ()

public	stderr_file ()

public	accu_hash ()

public Boolean	autoflow ()

public	lethal_for_worker ()

public	transient_error ()

public	incomplete ()

public	died_somewhere ()

public	load_parameters ()

public	fan_cache ()

public	dataflow_output_id ()

public	toString ()

Public Member Functions inherited from Bio::EnsEMBL::Hive::Storable
public Bio::EnsEMBL::Hive::Storable	new ()

public Int	dbID ()

public Bio::EnsEMBL::Hive::DBSQL::BaseAdaptor	adaptor ()

public	DESTROY ()

public	AUTOLOAD ()

Public Member Functions inherited from Bio::EnsEMBL::Hive::Params
public	new ()

public	param_init ()

protected	_param_possibly_overridden ()

protected	_param_silent ()

public Any	param_required ()

public Boolean	param_exists ()

public Boolean	param_is_defined ()

public Any	param ()

public	param_substitute ()

public	mysql_conn ()

public	mysql_dbname ()

public	csvq ()

protected	_subst_one_hashpair ()

Detailed Description

Description

    An AnalysisJob is the link between the input_id control data, the analysis and
    the rule system.  It also tracks the state of the job as it is processed

Member Function Documentation

◆ accu_hash()

public Bio::EnsEMBL::Hive::AnalysisJob::accu_hash ( )

Undocumented method

Code:

click to view

◆ accu_id_stack()

public Bio::EnsEMBL::Hive::AnalysisJob::accu_id_stack ( )

Undocumented method

Code:

click to view

◆ autoflow()

public Boolean Bio::EnsEMBL::Hive::AnalysisJob::autoflow ( )

    Title   :  autoflow
    Function:  Gets/sets flag for whether the job should
               be automatically dataflowed on branch 1 when the job completes.
               If the subclass manually sends a job along branch 1 with dataflow_output_id,
               the autoflow will turn itself off.
    Returns :  boolean (1=default|0)

Code:

click to view

◆ dataflow_output_id()

public Bio::EnsEMBL::Hive::AnalysisJob::dataflow_output_id ( )

    Title        :  dataflow_output_id
    Arg[1](req)  :  <string> $output_id 
    Arg[2](opt)  :  <int> $branch_name_or_code (optional, defaults to 1)
    Usage        :  $self->dataflow_output_id($output_id, $branch_name_or_code);
    Function:  
      If a RunnableDB(Process) needs to create jobs, this allows it to have jobs 
      created and flowed through the dataflow rules of the workflow graph.
      This 'output_id' becomes the 'input_id' of the newly created job at
      the ends of the dataflow pipes.  The optional 'branch_name_or_code' determines
      which dataflow pipe(s) to flow the job through.

Code:

click to view

sub dataflow_output_id {
    my ($self, $output_ids, $branch_name_or_code, $create_job_options) = @_;
    my $input_id                = $self->input_id();
    my $param_id_stack          = $self->param_id_stack();
    my $accu_id_stack           = $self->accu_id_stack();
    my $job_adaptor             = $self->adaptor() || 'Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor';
    my $hive_use_param_stack    = ref($job_adaptor) && $job_adaptor->db->hive_use_param_stack();
    if($hive_use_param_stack) {
        if($input_id and ($input_id ne '{}')) {     # add the parent to the param_id_stack if it had non-trivial extra parameters
            $param_id_stack = ($param_id_stack ? $param_id_stack.',' : '').$self->dbID();
        }
        if(scalar(keys %{$self->accu_hash()})) {    # add the parent to the accu_id_stack if it had "own" accumulator
            $accu_id_stack = ($accu_id_stack ? $accu_id_stack.',' : '').$self->dbID();
        }
    }
    $output_ids  ||= [ $hive_use_param_stack ? {} : $input_id ];            # by default replicate the parameters of the parent in the child
    $output_ids    = [ $output_ids ] unless(ref($output_ids) eq 'ARRAY');   # force previously used single values into an arrayref
    if($create_job_options) {
        die "Please consider configuring semaphored dataflow from PipeConfig rather than setting it up manually";
    }
        # map branch names to numbers:
    my $branch_code = Bio::EnsEMBL::Hive::DBSQL::DataflowRuleAdaptor::branch_name_2_code($branch_name_or_code);
        # if branch_code is set to 1 (explicitly or impliticly), turn off automatic dataflow:
    $self->autoflow(0) if($branch_code == 1);
    my @output_job_ids = ();
        # sort rules to make sure the fan rules come before funnel rules for the same branch_code:
    foreach my $rule (sort {($b->funnel_dataflow_rule//0) cmp ($a->funnel_dataflow_rule//0)} @{ $self->analysis->dataflow_rules_by_branch->{$branch_code} || [] }) {
            # parameter substitution into input_id_template is rule-specific
        my $output_ids_for_this_rule;
        if(my $template_string = $rule->input_id_template()) {
            my $template_hash = destringify($template_string);
            $output_ids_for_this_rule = [ map { $self->param_substitute($template_hash, $_) } @$output_ids ];
        } else {
            $output_ids_for_this_rule = $output_ids;
        }
        my $target_analysis_or_table = $rule->to_analysis();
        if($target_analysis_or_table->can('dataflow')) {
            $target_analysis_or_table->dataflow( $output_ids_for_this_rule, $self );
        } else {
            my @common_params = (
                'prev_job'          => $self,
                'analysis'          => $target_analysis_or_table,   # expecting an Analysis
                'param_id_stack'    => $param_id_stack,
                'accu_id_stack'     => $accu_id_stack,
            );
            if( my $funnel_dataflow_rule = $rule->funnel_dataflow_rule ) {    # members of a semaphored fan will have to wait in cache until the funnel is created:
                my $fan_cache_this_branch = $self->fan_cache->{"$funnel_dataflow_rule"} ||= [];
                push @$fan_cache_this_branch, map { Bio::EnsEMBL::Hive::AnalysisJob->new(
                                                        @common_params,
                                                        'input_id'          => $_,
                                                        # semaphored_job_id  => to be set when the $funnel_job has been stored
                                                    ) } @$output_ids_for_this_rule;
            } else {    # either a semaphored funnel or a non-semaphored dataflow:
                my $fan_jobs = delete $self->fan_cache->{"$rule"};   # clear the cache at the same time
                if( $fan_jobs && @$fan_jobs ) { # a semaphored funnel
                    if( (my $funnel_job_count = scalar(@$output_ids_for_this_rule)) !=1 ) {
                        $self->transient_error(0);
                        die "Asked to dataflow into $funnel_job_count funnel jobs instead of 1";
                    } else {
                        my $funnel_job = Bio::EnsEMBL::Hive::AnalysisJob->new(
                                            @common_params,
                                            'input_id'          => $output_ids_for_this_rule->[0],
                                            'semaphore_count'   => scalar(@$fan_jobs),          # "pre-increase" the semaphore count before creating the dependent jobs
                                            'semaphored_job_id' => $self->semaphored_job_id(),  # propagate parent's semaphore if any
                        );
                        my ($funnel_job_id) = @{ $job_adaptor->store_jobs_and_adjust_counters( [ $funnel_job ], 0) };
                        unless($funnel_job_id) {    # apparently it has been created previously, trying to leech to it:
                            if( $funnel_job = $job_adaptor->fetch_by_analysis_id_and_input_id( $funnel_job->analysis->dbID, $funnel_job->input_id) ) {
                                $funnel_job_id = $funnel_job->dbID;
                                if( $funnel_job->status eq 'SEMAPHORED' ) {
                                    $job_adaptor->increase_semaphore_count_for_jobid( $funnel_job_id, scalar(@$fan_jobs) );    # "pre-increase" the semaphore count before creating the dependent jobs
                                    $job_adaptor->db->get_LogMessageAdaptor->store_job_message($self->dbID, "Discovered and using an existing funnel ".$funnel_job->toString, 0);
                                } else {
                                    die "The funnel job (id=$funnel_job_id) fetched from the database was not in SEMAPHORED status";
                                }
                            } else {
                                die "The funnel job could neither be stored nor fetched";
                            }
                        }
                        foreach my $fan_job (@$fan_jobs) {  # set the funnel in every fan's job:
                            $fan_job->semaphored_job_id( $funnel_job_id );
                        }
                        push @output_job_ids, $funnel_job_id, @{ $job_adaptor->store_jobs_and_adjust_counters( $fan_jobs, 1) };
                    }
                } else {    # non-semaphored dataflow (but potentially propagating any existing semaphores)
                    my @non_semaphored_jobs = map { Bio::EnsEMBL::Hive::AnalysisJob->new(
                                                        @common_params,
                                                        'input_id'          => $_,
                                                        'semaphored_job_id' => $self->semaphored_job_id(),  # propagate parent's semaphore if any
                    ) } @$output_ids_for_this_rule;
                    push @output_job_ids, @{ $job_adaptor->store_jobs_and_adjust_counters( \@non_semaphored_jobs, 0) };
                }
            } # /if funnel
        } # /if (table or analysis)
    } # /foreach my $rule
    return \@output_job_ids;
}

◆ died_somewhere()

public Bio::EnsEMBL::Hive::AnalysisJob::died_somewhere ( )

Undocumented method

Code:

click to view

◆ fan_cache()

public Bio::EnsEMBL::Hive::AnalysisJob::fan_cache ( )

Undocumented method

Code:

click to view

◆ incomplete()

public Bio::EnsEMBL::Hive::AnalysisJob::incomplete ( )

Undocumented method

Code:

click to view

◆ input_id()

public Bio::EnsEMBL::Hive::AnalysisJob::input_id ( )

Undocumented method

Code:

click to view

◆ lethal_for_worker()

public Bio::EnsEMBL::Hive::AnalysisJob::lethal_for_worker ( )

Undocumented method

Code:

click to view

◆ load_parameters()

public Bio::EnsEMBL::Hive::AnalysisJob::load_parameters ( )

Undocumented method

Code:

click to view

◆ param_id_stack()

public Bio::EnsEMBL::Hive::AnalysisJob::param_id_stack ( )

Undocumented method

Code:

click to view

◆ query_count()

public Bio::EnsEMBL::Hive::AnalysisJob::query_count ( )

Undocumented method

Code:

click to view

◆ retry_count()

public Bio::EnsEMBL::Hive::AnalysisJob::retry_count ( )

Undocumented method

Code:

click to view

◆ role_id()

public Bio::EnsEMBL::Hive::AnalysisJob::role_id ( )

Undocumented method

Code:

click to view

◆ runtime_msec()

public Bio::EnsEMBL::Hive::AnalysisJob::runtime_msec ( )

Undocumented method

Code:

click to view

◆ semaphore_count()

public Bio::EnsEMBL::Hive::AnalysisJob::semaphore_count ( )

Undocumented method

Code:

click to view

◆ semaphored_job_id()

public Bio::EnsEMBL::Hive::AnalysisJob::semaphored_job_id ( )

Undocumented method

Code:

click to view

◆ set_and_update_status()

public Bio::EnsEMBL::Hive::AnalysisJob::set_and_update_status ( )

Undocumented method

Code:

click to view

◆ status()

public Bio::EnsEMBL::Hive::AnalysisJob::status ( )

Undocumented method

Code:

click to view

◆ stderr_file()

public Bio::EnsEMBL::Hive::AnalysisJob::stderr_file ( )

Undocumented method

Code:

click to view

◆ stdout_file()

public Bio::EnsEMBL::Hive::AnalysisJob::stdout_file ( )

Undocumented method

Code:

click to view

◆ toString()

public Bio::EnsEMBL::Hive::AnalysisJob::toString ( )

Undocumented method

Code:

click to view

◆ transient_error()

public Bio::EnsEMBL::Hive::AnalysisJob::transient_error ( )

Undocumented method

Code:

click to view

◆ when_completed()

public Bio::EnsEMBL::Hive::AnalysisJob::when_completed ( )

Undocumented method

Code:

click to view

The documentation for this class was generated from the following file:

modules/Bio/EnsEMBL/Hive/AnalysisJob.pm

Public Member Functions

Detailed Description

Description

Member Function Documentation

◆ accu_hash()

◆ accu_id_stack()

◆ autoflow()

◆ dataflow_output_id()

◆ died_somewhere()

◆ fan_cache()

◆ incomplete()

◆ input_id()

◆ lethal_for_worker()

◆ load_parameters()

◆ param_id_stack()

◆ query_count()

◆ retry_count()

◆ role_id()

◆ runtime_msec()

◆ semaphore_count()

◆ semaphored_job_id()

◆ set_and_update_status()

◆ status()

◆ stderr_file()

◆ stdout_file()

◆ toString()

◆ transient_error()

◆ when_completed()