Inheritance diagram for Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor:

Public Member Functions
public	default_table_name ()

public	default_insertion_method ()

public	object_class ()

public	default_overflow_limit ()

public String	job_status_cast ()

public AnalysisJob	fetch_by_analysis_id_and_input_id ()

public	class_specific_execute ()

public Reference	store_jobs_and_adjust_counters ()

public	store_a_semaphored_group_of_jobs ()

public Reference	fetch_all_by_analysis_id_status ()

public	fetch_some_by_analysis_id_limit ()

public	fetch_all_incomplete_jobs_by_role_id ()

public	fetch_all_unfinished_jobs_with_no_roles ()

public	fetch_by_url_query ()

public	fetch_job_counts_hashed_by_status ()

public	semaphore_job_by_id ()

public	unsemaphore_job_by_id ()

public	prelock_semaphore_for_update ()

public	check_in_job ()

public	store_out_files ()

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

public	grab_jobs_for_role ()

public	release_claimed_jobs_from_role ()

public	release_undone_jobs_from_role ()

public	release_and_age_job ()

public	gc_dataflow ()

public	reset_jobs_for_analysis_id ()

public	unblock_jobs_for_analysis_id ()

public	discard_jobs_for_analysis_id ()

public	balance_semaphores ()

public	fetch_input_ids_for_job_ids ()

Public Member Functions inherited from Bio::EnsEMBL::Hive::DBSQL::ObjectAdaptor
public	object_class ()

public	slicer ()

public	objectify ()

public	mark_stored ()

public	keys_to_columns ()

public The	refresh ()

Public Member Functions inherited from Bio::EnsEMBL::Hive::DBSQL::BaseAdaptor
public	default_table_name ()

public	default_insertion_method ()

public	default_overflow_limit ()

public	default_input_column_mapping ()

public	do_not_update_columns ()

public	new ()

public	db ()

public	dbc ()

public	prepare ()

public	overflow_limit ()

public	input_column_mapping ()

public	table_name ()

public	insertion_method ()

public	column_set ()

public	primary_key ()

public	updatable_column_list ()

public	autoinc_id ()

protected	_table_info_loader ()

public	count_all ()

public	fetch_all ()

public	primary_key_constraint ()

public	fetch_by_dbID ()

public	remove_all ()

public	remove ()

public	update ()

public	store_or_update_one ()

public	check_object_present_in_db_by_content ()

public	class_specific_execute ()

public	store ()

protected	_multi_column_filter ()

public	DESTROY ()

public	AUTOLOAD ()

Detailed Description

Synopsis

$analysisJobAdaptor = $db_adaptor->get_AnalysisJobAdaptor;

$analysisJobAdaptor = $analysisJob->adaptor;

Description

    Module to encapsulate all db access for persistent class AnalysisJob.
    There should be just one per application and database connection.

Definition at line 22 of file AnalysisJobAdaptor.pm.

Member Function Documentation

◆ balance_semaphores()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::balance_semaphores ( )

  Description: Reset all semaphore_counts to the numbers of unDONE semaphoring jobs.

Code:

click to view

sub balance_semaphores {
    my ($self, $list_of_analyses) = @_;
 
    my $analysis_filter = $list_of_analyses
        ? "funnel.analysis_id IN (".join(',', map { $_->dbID } @$list_of_analyses).") AND"
        : '';
 
    my $find_sql    = qq{
                        SELECT * FROM (
                            SELECT s.semaphore_id, s.local_jobs_counter AS was, COALESCE(COUNT(CASE WHEN fan.status NOT IN ($ALL_STATUSES_OF_COMPLETE_JOBS) THEN 1 ELSE NULL END),0) AS should
                            FROM semaphore s
                            LEFT JOIN job fan ON (s.semaphore_id=fan.controlled_semaphore_id)
                            LEFT JOIN job funnel ON (s.dependent_job_id=funnel.job_id)
                            WHERE $analysis_filter
                            funnel.status in ('SEMAPHORED', 'READY')
                            GROUP BY s.semaphore_id
                         ) AS internal WHERE was<>should OR should=0
                     };
 
    my $rebalanced_jobs_counter = 0;
 
    # Run in a transaction to ensure we see a consistent state of the job
    # statuses and semaphore counts.
    $self->dbc->run_in_transaction( sub {
 
    my $find_sth            = $self->prepare($find_sql);
    my $semaphore_adaptor   = $self->db->get_SemaphoreAdaptor;
 
    $find_sth->execute();
    while(my ($semaphore_id, $was, $should) = $find_sth->fetchrow_array()) {
        my $msg;
        if($should<$was) {
 
            $msg = "Semaphore $semaphore_id local_jobs_counter has to be decreased $was -> $should, performing it now with a potential release";
            $self->db->get_LogMessageAdaptor->store_hive_message( $msg, 'PIPELINE_CAUTION' );
 
            my $semaphore = $semaphore_adaptor->fetch_by_dbID( $semaphore_id );
            $semaphore->decrease_by( $was-$should );                                # decrease the local_jobs_counter, recursively releasing if ripe
 
            $rebalanced_jobs_counter++;
        } elsif($was<$should) {
 
            $msg = "Semaphore $semaphore_id local_jobs_counter has to be increased $was -> $should, performing it now with a potential reblock";
            $self->db->get_LogMessageAdaptor->store_hive_message( $msg, 'PIPELINE_CAUTION' );
 
            my $semaphore = $semaphore_adaptor->fetch_by_dbID( $semaphore_id );
            $semaphore->reblock_by( $should-$was );                                 # increase the local_jobs_counter, reblock recursively if needed
 
            $rebalanced_jobs_counter++;
        } else {
            my $semaphore = $semaphore_adaptor->fetch_by_dbID( $semaphore_id );
            # check_if_ripe does the same but with an extra call to the database
            if( $semaphore->local_jobs_counter + $semaphore->remote_jobs_counter <= 0) {
                $msg = "Semaphore $semaphore_id is marked as blocked despite nothing blocking it, releasing it now";
                $self->db->get_LogMessageAdaptor->store_hive_message( $msg, 'PIPELINE_CAUTION' );
 
                $semaphore->release_if_ripe();
            }
        }
        warn "[Semaphore $semaphore_id] $msg\n" if $msg;    # TODO: integrate the STDERR diagnostic output with LogMessageAdaptor calls in general
    }
    $find_sth->finish;
 
    } ); # end of transaction
 
    return $rebalanced_jobs_counter;
}

◆ check_in_job()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::check_in_job ( )

  Arg [1]    : $analysis_id
  Example    :

  Description: updates the job.status in the database
  Returntype : 
  Exceptions :
  Caller     : general

Code:

click to view

◆ class_specific_execute()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::class_specific_execute ( )

Undocumented method

Code:

click to view

◆ default_insertion_method()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::default_insertion_method ( )

Undocumented method

Code:

click to view

◆ default_overflow_limit()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::default_overflow_limit ( )

Undocumented method

Code:

click to view

◆ default_table_name()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::default_table_name ( )

Undocumented method

Code:

click to view

◆ discard_jobs_for_analysis_id()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::discard_jobs_for_analysis_id ( )

  Arg [1]    : list-ref of int $analysis_id
  Arg [2]    : filter status
  Description: Resets all $input_status jobs of the matching analyses to DONE.
               Semaphores are updated accordingly.
  Caller     : beekeeper.pl and guiHive

Code:

click to view

◆ fetch_all_by_analysis_id_status()

public Reference Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::fetch_all_by_analysis_id_status ( )

  Arg [1]    : (optional) listref $list_of_analyses
  Arg [2]    : (optional) string $status
  Arg [3]    : (optional) int $retry_at_least
  Example    :

$all_failed_jobs = $adaptor->fetch_all_by_analysis_id_status(undef, 'FAILED');

$analysis_done_jobs = $adaptor->fetch_all_by_analysis_id_status( $list_of_analyses, 'DONE');

  Description: Returns a list of all jobs filtered by given analysis_id (if specified) and given status (if specified).
  Returntype : reference to list of Bio::EnsEMBL::Hive::AnalysisJob objects

Code:

click to view

◆ fetch_all_incomplete_jobs_by_role_id()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::fetch_all_incomplete_jobs_by_role_id ( )

Undocumented method

Code:

click to view

◆ fetch_all_unfinished_jobs_with_no_roles()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::fetch_all_unfinished_jobs_with_no_roles ( )

Undocumented method

Code:

click to view

◆ fetch_by_analysis_id_and_input_id()

public AnalysisJob Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::fetch_by_analysis_id_and_input_id ( )

  Arg [1]    : Integer $analysis_id
  Arg [2]    : String $input_id
  Example    :

$funnel_job = $job_adaptor->fetch_by_analysis_id_and_input_id( $funnel_job->analysis->dbID, $funnel_job->input_id);

  Description: Attempts to find the job by contents, then makes another attempt if the input_id is expected to have overflown into analysis_data
  Returntype : AnalysisJob object

Code:

click to view

◆ fetch_by_url_query()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::fetch_by_url_query ( )

Undocumented method

Code:

click to view

◆ fetch_input_ids_for_job_ids()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::fetch_input_ids_for_job_ids ( )

Undocumented method

Code:

click to view

◆ fetch_job_counts_hashed_by_status()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::fetch_job_counts_hashed_by_status ( )

Undocumented method

Code:

click to view

◆ fetch_some_by_analysis_id_limit()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::fetch_some_by_analysis_id_limit ( )

Undocumented method

Code:

click to view

◆ gc_dataflow()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::gc_dataflow ( )

    Description:    perform automatic dataflow from a dead job that overused resources if a corresponding dataflow rule was provided
                    Should only be called once during garbage collection phase, when the job is definitely 'abandoned' and not being worked on.

Code:

click to view

◆ grab_jobs_for_role()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::grab_jobs_for_role ( )

  Arg [1]           : Bio::EnsEMBL::Hive::Role object $role
  Arg [2]           : int $how_many_this_role
  Example: 
    my $jobs  = $job_adaptor->grab_jobs_for_role( $role, $how_many );
  Description: 
    For the specified Role, it will search available jobs, 
    and using the how_many_this_batch parameter, claim/fetch that
    number of jobs, and then return them.
  Returntype : 
    reference to array of Bio::EnsEMBL::Hive::AnalysisJob objects
  Caller     : Bio::EnsEMBL::Hive::Worker::run

Code:

click to view

◆ job_status_cast()

public String Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::job_status_cast ( )

  Example     :

$job_adaptor->job_status_cast();

  Description : Changes the type of the expression to the type of the
                job-status ENUM.
                This is needed for CASE expressions in PostgreSQL which
                otherwise default to returning a string.
  Returntype  : String
  Exceptions  : none

Code:

click to view

◆ object_class()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::object_class ( )

Undocumented method

Code:

click to view

◆ prelock_semaphore_for_update()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::prelock_semaphore_for_update ( )

Undocumented method

Code:

click to view

◆ release_and_age_job()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::release_and_age_job ( )

Undocumented method

Code:

click to view

◆ release_claimed_jobs_from_role()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::release_claimed_jobs_from_role ( )

Undocumented method

Code:

click to view

◆ release_undone_jobs_from_role()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::release_undone_jobs_from_role ( )

  Arg [1]    : Bio::EnsEMBL::Hive::Role object
  Arg [2]    : optional message to be recorded in 'job_message' table
  Example    :

  Description: If a Worker has died some of its jobs need to be reset back to 'READY'
               so they can be rerun.
               Jobs in state CLAIMED as simply reset back to READY.
               If jobs was 'in progress' (see the $ALL_STATUSES_OF_RUNNING_JOBS variable)
               the retry_count is increased and the status set back to READY.
               If the retry_count >= $max_retry_count (3 by default) the job is set
               to 'FAILED' and not rerun again.
  Exceptions : $role must be defined
  Caller     : Bio::EnsEMBL::Hive::Queen

Code:

click to view

sub release_undone_jobs_from_role {
    my ($self, $role, $msg) = @_;
 
    my $role_id         = $role->dbID;
    my $analysis        = $role->analysis;
    my $max_retry_count = $analysis->max_retry_count;
    my $worker          = $role->worker;
 
        #first just reset the claimed jobs, these don't need a retry_count index increment:
    $self->release_claimed_jobs_from_role( $role );
 
    my $sth = $self->prepare( qq{
        SELECT job_id
          FROM job
         WHERE role_id='$role_id'
           AND status in ($ALL_STATUSES_OF_TAKEN_JOBS)
    } );
    $sth->execute();
 
    my $cod = $worker->cause_of_death() || 'UNKNOWN';
    $msg ||= "GarbageCollector: The worker died because of $cod";
 
    my $resource_overusage = ($cod eq 'MEMLIMIT') || ($cod eq 'RUNLIMIT' and $worker->work_done()==0);
 
    while(my ($job_id) = $sth->fetchrow_array()) {
 
        my $passed_on = 0;  # the flag indicating that the garbage_collection was attempted and was successful
 
        if( $resource_overusage ) {
            if($passed_on = $self->gc_dataflow( $analysis, $job_id, $cod )) {
                $msg .= ', performing gc_dataflow';
            }
        }
        unless($passed_on) {
            if($passed_on = $self->gc_dataflow( $analysis, $job_id, 'ANYFAILURE' )) {
                $msg .= ", performing 'ANYFAILURE' gc_dataflow";
            }
        }
 
        $self->db()->get_LogMessageAdaptor()->store_job_message($job_id, $msg, $passed_on ? 'INFO' : 'WORKER_ERROR');
 
        unless($passed_on) {
            # We can not retry this job if it failed during compilation or due
            # to a resource constraint. If a gc_dataflow is present, it will
            # take care of that and create another job.
            my $worker_status = $worker->status();
            my $no_retry = ($cod eq 'COMPILATION') || $resource_overusage;
            $self->release_and_age_job( $job_id, $max_retry_count, not $no_retry );
        }
 
        $role->register_attempt( 0 );
    }
    $sth->finish();
}

◆ reset_jobs_for_analysis_id()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::reset_jobs_for_analysis_id ( )

  Arg [1]    : arrayref of Analyses
  Arg [2]    : arrayref of job statuses $input_statuses
  Description: Resets all the jobs of the selected analyses that have one of the
               required statuses to 'READY' and their retry_count to 0.
               Semaphores are updated accordingly.
  Caller     : beekeeper.pl and guiHive

Code:

click to view

sub reset_jobs_for_analysis_id {
    my ($self, $list_of_analyses, $input_statuses) = @_;
 
    return if !scalar(@$input_statuses);  # No statuses to reset
 
    my $analyses_filter = 'j.analysis_id IN ('.join(',', map { $_->dbID } @$list_of_analyses).')';
    my $statuses_filter = 'AND j.status IN ('.join(', ', map { "'$_'" } @$input_statuses).')';
 
    # Get the list of semaphores, and by how much their local_jobs_counter should be increased.
    # Only DONE and PASSED_ON jobs of the matching analyses and statuses should be counted
    #
    my $sql1 = qq{
        SELECT COUNT(*) AS local_delta, controlled_semaphore_id
        FROM job j
        WHERE controlled_semaphore_id IS NOT NULL
              AND $analyses_filter $statuses_filter AND status IN ($ALL_STATUSES_OF_COMPLETE_JOBS)
        GROUP BY controlled_semaphore_id
    };
 
    # Run in a transaction to ensure we see a consistent state of the job
    # statuses and semaphore counts.
    $self->dbc->run_in_transaction( sub {
 
        my $semaphore_adaptor = $self->db->get_SemaphoreAdaptor;
 
        # Update all the semaphored jobs one by one
        my $sth1 = $self->prepare($sql1);
        $sth1->execute();
        while (my ($local_delta, $semaphore_id) = $sth1->fetchrow_array()) {
 
            my $semaphore = $semaphore_adaptor->fetch_by_dbID( $semaphore_id );
            $semaphore->reblock_by( $local_delta );                                 # increase the local_jobs_counter, reblock recursively if needed
        }
        $sth1->finish;
 
            # change fan jobs' statuses to 'READY', if they are themselves not SEMAPHORED
        my $sql3 = ($self->dbc->driver eq 'mysql') ? qq{
                UPDATE job j
             LEFT JOIN semaphore s
                    ON (j.job_id=s.dependent_job_id)
                   SET j.retry_count = CASE WHEN j.status='READY' THEN 0 ELSE 1 END,
                       j.status = }.$self->job_status_cast("CASE WHEN s.local_jobs_counter+s.remote_jobs_counter>0 THEN 'SEMAPHORED' ELSE 'READY' END").qq{
                 WHERE $analyses_filter $statuses_filter
        } : ($self->dbc->driver eq 'pgsql') ? qq{
                UPDATE job
                   SET retry_count = CASE WHEN j.status='READY' THEN 0 ELSE 1 END,
                       status = }.$self->job_status_cast("CASE WHEN s.local_jobs_counter+s.remote_jobs_counter>0 THEN 'SEMAPHORED' ELSE 'READY' END").qq{
                  FROM job j
             LEFT JOIN semaphore s
                    ON (j.job_id=s.dependent_job_id)
                 WHERE job.job_id=j.job_id AND $analyses_filter $statuses_filter
        } : qq{
 
            REPLACE INTO job (job_id, prev_job_id, analysis_id, input_id, param_id_stack, accu_id_stack, role_id, status, retry_count, when_completed, runtime_msec, query_count, controlled_semaphore_id)
                  SELECT j.job_id,
                         j.prev_job_id,
                         j.analysis_id,
                         j.input_id,
                         j.param_id_stack,
                         j.accu_id_stack,
                         j.role_id,
                         CASE WHEN s.local_jobs_counter+s.remote_jobs_counter>0 THEN 'SEMAPHORED' ELSE 'READY' END,
                         CASE WHEN j.status='READY' THEN 0 ELSE 1 END,
                         j.when_completed,
                         j.runtime_msec,
                         j.query_count,
                         j.controlled_semaphore_id
                    FROM job j
               LEFT JOIN semaphore s
                      ON (j.job_id=s.dependent_job_id)
                   WHERE $analyses_filter $statuses_filter
        };
 
        $self->dbc->protected_prepare_execute( [$sql3],
            sub { my ($after) = @_; $self->db->get_LogMessageAdaptor->store_hive_message( 'resetting jobs'.$after, 'INFO' ); }

◆ reset_or_grab_job_by_dbID()

public Bio::EnsEMBL::Hive::AnalysisJob Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::reset_or_grab_job_by_dbID ( )

  Arg [1]    : int $job_id
  Arg [2]    : int $role_id (optional)
  Description: resets a job to to 'READY' (if no $role_id given) or directly to 'CLAIMED' so it can be run again, and fetches it.
               NB: Will also reset a previously 'SEMAPHORED' job to READY.
               The retry_count will be set to 1 for previously run jobs (partially or wholly) to trigger PRE_CLEANUP for them,
               but will not change retry_count if a job has never *really* started.
  Returntype : Bio::EnsEMBL::Hive::AnalysisJob or undef

Code:

click to view

◆ semaphore_job_by_id()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::semaphore_job_by_id ( )

Undocumented method

Code:

click to view

◆ store_a_semaphored_group_of_jobs()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::store_a_semaphored_group_of_jobs ( )

  Arg [1]    : Bio::EnsEMBL::Hive::AnalysisJob $funnel_job
  Arg [2]    : arrayref of Bio::EnsEMBL::Hive::AnalysisJob $fan_jobs
  Arg [3]    : (optional) Bio::EnsEMBL::Hive::AnalysisJob $emitting_job
  Arg [4]    : (optional) boolean $no_leeching
  Example    :

my ($funnel_semaphore_id, $funnel_job_id, @fan_job_ids) = $job_adaptor->store_a_semaphored_group_of_jobs( $funnel_job, $fan_jobs, $emitting_job );

  Description: Attempts to store a semaphored group of jobs, returns a list of successfully stored job_ids
  Returntype : ($funnel_semaphore_id, $funnel_job_id, @fan_job_ids)

Code:

click to view

sub store_a_semaphored_group_of_jobs {
    my ($self, $funnel_job, $fan_jobs, $emitting_job, $no_leeching) = @_;
 
    my $emitting_job_id;
 
    if($emitting_job) {
        if($funnel_job) {
            $funnel_job->prev_job( $emitting_job );
            $funnel_job->controlled_semaphore( $emitting_job->controlled_semaphore );   # propagate parent's semaphore if any
        }
        $emitting_job_id = $emitting_job->dbID;
    }
 
    my $funnel_semaphore;
    my $funnel_semaphore_adaptor    = $self->db->get_SemaphoreAdaptor;  # assuming $self was $funnel_job_adaptor
 
    my ($funnel_job_id)     = $funnel_job ? @{ $self->store_jobs_and_adjust_counters( [ $funnel_job ], 0, $emitting_job_id) } : ();
 
    if($funnel_job && !$funnel_job_id) {    # apparently the funnel_job has been created previously, trying to leech to it:
 
        if($no_leeching) {
            die "The funnel job could not be stored, but leeching was not allowed, so bailing out";
 
        } elsif( $funnel_job = $self->fetch_by_analysis_id_and_input_id( $funnel_job->analysis->dbID, $funnel_job->input_id) ) {
            $funnel_job_id = $funnel_job->dbID;
 
            # If the job hasn't run yet, we can still block it
            if ($funnel_job->status eq 'READY') {
                # Mark the job as SEMAPHORED to make sure it's not taken by any worker
                $self->semaphore_job_by_id($funnel_job_id);
                $self->refresh($funnel_job);
            }
 
            if( $funnel_job->status eq 'SEMAPHORED' ) {
 
                $funnel_semaphore = $funnel_job->fetch_local_blocking_semaphore();
 
                # Create if it was missing
                unless ($funnel_semaphore) {
                    $funnel_semaphore = Bio::EnsEMBL::Hive::Semaphore->new(
                        'hive_pipeline'         => $funnel_job->hive_pipeline,
                        'dependent_job_id'      => $funnel_job_id,
                        'local_jobs_counter'    => 0,   # Will be updated below
                        'remote_jobs_counter'   => 0,   # Will be updated below
                    );
                    $funnel_semaphore_adaptor->store( $funnel_semaphore );
                }
 
                $funnel_semaphore->increase_by( $fan_jobs );  # "pre-increase" the semaphore counts before creating the controlling jobs
 
                $self->db->get_LogMessageAdaptor->store_job_message($emitting_job_id, "Discovered and using an existing funnel ".$funnel_job->toString, 'INFO');
            } 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";
        }
    } else {    # Either the $funnel_job was successfully stored, or there wasn't any $funnel_job to start with:
 
        my $whose_hive_pipeline = $funnel_job || $self->db;
 
        my ($local_count, $remote_count)    = Bio::EnsEMBL::Hive::Cacheable::count_local_and_remote_objects( $whose_hive_pipeline, $fan_jobs );
 
        $funnel_semaphore = Bio::EnsEMBL::Hive::Semaphore->new(
            'hive_pipeline'         => $whose_hive_pipeline->hive_pipeline,
            'dependent_job_id'      => $funnel_job_id,
            'local_jobs_counter'    => $local_count,
            'remote_jobs_counter'   => $remote_count,
        );
        $funnel_semaphore_adaptor->store( $funnel_semaphore );
 
        $funnel_semaphore->release_if_ripe();
    }
 
    foreach my $fan_job (@$fan_jobs) {  # set the funnel in every fan's job:
        $fan_job->controlled_semaphore( $funnel_semaphore );
    }
 
    my (@fan_job_ids) = @{ $self->store_jobs_and_adjust_counters( $fan_jobs, 1, $emitting_job_id) };
 
    return ($funnel_semaphore->dbID, $funnel_job_id, @fan_job_ids);
}

◆ store_jobs_and_adjust_counters()

public Reference Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::store_jobs_and_adjust_counters ( )

  Arg [1]    : arrayref of Bio::EnsEMBL::Hive::AnalysisJob $jobs_to_store
  Arg [2]    : (optional) boolean $push_new_semaphore
  Arg [3]    : (optional) Int $emitting_job_id
  Example    :

my @output_job_ids = @{ $job_adaptor->store_jobs_and_adjust_counters( \@jobs_to_store ) };

  Description: Attempts to store a list of jobs, returns an arrayref of successfully stored job_ids
  Returntype : Reference to list of job_dbIDs

Code:

click to view

sub store_jobs_and_adjust_counters {
    my ($self, $jobs, $push_new_semaphore, $emitting_job_id) = @_;
 
    my @output_job_ids                      = ();
 
        # NB: our use patterns assume all jobs from the same storing batch share the same controlled_semaphore:
    my $controlled_semaphore                = scalar(@$jobs) && $jobs->[0]->controlled_semaphore;
    my @jobs_that_failed_to_store           = ();
 
    if( $controlled_semaphore && !$push_new_semaphore ) {   # only if it has not been done yet
        $controlled_semaphore->increase_by( $jobs );  # "pre-increase" the semaphore counts before creating the controlling jobs
    }
 
    foreach my $job (@$jobs) {
 
        my $analysis    = $job->analysis;
        my $job_adaptor = $analysis ? $analysis->adaptor->db->get_AnalysisJobAdaptor : $self;   # if analysis object is undefined, consider the job local
        my $prev_adaptor= ($job->prev_job && $job->prev_job->adaptor) || '';
        my $job_is_local_to_parent  = $prev_adaptor eq $job_adaptor;
 
        if( $controlled_semaphore ) {
            my $job_hive_pipeline = $job->hive_pipeline;
 
            if( $controlled_semaphore->hive_pipeline ne $job_hive_pipeline ) {      # if $job happens to be remote to $controlled_semaphore,
                                                                                    # introduce another job-local semaphore between $job and $controlled_semaphore:
                my $job_local_semaphore = Bio::EnsEMBL::Hive::Semaphore->new(
                    'hive_pipeline'             => $job_hive_pipeline,
                    'dependent_semaphore_url'   => $controlled_semaphore->relative_url( $job_hive_pipeline ),
                    'local_jobs_counter'        => 1,
                    'remote_jobs_counter'       => 0,
                );
                $job_adaptor->db->get_SemaphoreAdaptor->store( $job_local_semaphore );
 
                $job->controlled_semaphore( $job_local_semaphore );
            }
        }
 
        if( $job_adaptor ne $prev_adaptor ) {
            $job->prev_job_id( undef );             # job_ids are local, so for remote jobs they have to be cleaned up before storing
        }
 
        my ($job, $stored_this_time) = $job_adaptor->store( $job );
 
        if($stored_this_time) {
 
            unless($job_adaptor->db->hive_pipeline->hive_use_triggers()) {
                $job_adaptor->dbc->do(qq{
                        UPDATE analysis_stats
                        SET total_job_count=total_job_count+1
                    }
                    .(($job->status eq 'READY')
                        ? " ,ready_job_count=ready_job_count+1 "
                        : " ,semaphored_job_count=semaphored_job_count+1 "
                    ).(($job_adaptor->dbc->driver eq 'pgsql')
                        ? " ,status = CAST(CASE WHEN status!='BLOCKED' THEN 'LOADING' ELSE 'BLOCKED' END AS analysis_status) "
                        : " ,status =      CASE WHEN status!='BLOCKED' THEN 'LOADING' ELSE 'BLOCKED' END "
                    )." WHERE analysis_id=".$job->analysis_id
                );
            }
 
            push @output_job_ids, $job->dbID();     # FIXME: this ID may not make much cross-db sense
 
        } else {
            push @jobs_that_failed_to_store, $job;
 
            my $msg = "JobAdaptor failed to store the "
                     . ($job_is_local_to_parent ? 'local' : 'remote')
                     . " Job( analysis_id=".$job->analysis_id.', '.$job->input_id." ), possibly due to a collision";
            if ($job_is_local_to_parent && $emitting_job_id) {
                $self->db->get_LogMessageAdaptor->store_job_message($emitting_job_id, $msg, 'PIPELINE_CAUTION');
            } else {
                $self->db->get_LogMessageAdaptor->store_hive_message($msg, 'PIPELINE_CAUTION');
            }
 
        }
    }
 
    if( $controlled_semaphore && scalar(@jobs_that_failed_to_store) ) {
        $controlled_semaphore->decrease_by( \@jobs_that_failed_to_store );
    }
 
    return \@output_job_ids;
}

◆ store_out_files()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::store_out_files ( )

  Arg [1]    : Bio::EnsEMBL::Hive::AnalysisJob $job
  Example    :

  Description: update locations of log files, if present
  Returntype : 
  Exceptions :
  Caller     : Bio::EnsEMBL::Hive::Worker

Code:

click to view

◆ unblock_jobs_for_analysis_id()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::unblock_jobs_for_analysis_id ( )

  Arg [1]    : list-ref of int $analysis_id
  Description: Sets all the SEMAPHORED jobs of the given analyses to READY and also unblocks their upstream semaphores
  Caller     : beekeeper.pl and guiHive

Code:

click to view

◆ unsemaphore_job_by_id()

public Bio::EnsEMBL::Hive::DBSQL::AnalysisJobAdaptor::unsemaphore_job_by_id ( )

Undocumented method

Code:

click to view

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

modules/Bio/EnsEMBL/Hive/DBSQL/AnalysisJobAdaptor.pm

Public Member Functions

Detailed Description

Synopsis

Description

Member Function Documentation

◆ balance_semaphores()

◆ check_in_job()

◆ class_specific_execute()

◆ default_insertion_method()

◆ default_overflow_limit()

◆ default_table_name()

◆ discard_jobs_for_analysis_id()

◆ fetch_all_by_analysis_id_status()

◆ fetch_all_incomplete_jobs_by_role_id()

◆ fetch_all_unfinished_jobs_with_no_roles()

◆ fetch_by_analysis_id_and_input_id()

◆ fetch_by_url_query()

◆ fetch_input_ids_for_job_ids()

◆ fetch_job_counts_hashed_by_status()

◆ fetch_some_by_analysis_id_limit()

◆ gc_dataflow()

◆ grab_jobs_for_role()

◆ job_status_cast()

◆ object_class()

◆ prelock_semaphore_for_update()

◆ release_and_age_job()

◆ release_claimed_jobs_from_role()

◆ release_undone_jobs_from_role()

◆ reset_jobs_for_analysis_id()

◆ reset_or_grab_job_by_dbID()

◆ semaphore_job_by_id()

◆ store_a_semaphored_group_of_jobs()

◆ store_jobs_and_adjust_counters()

◆ store_out_files()

◆ unblock_jobs_for_analysis_id()

◆ unsemaphore_job_by_id()