visualize_jobs.pl File Reference
Go to the source code of this file.
Functions | |
| public | main () |
| public | precache_participating_pipelines () |
| public | find_the_top () |
| public | trace_job_up () |
| public | trace_semaphore_up () |
| public | draw_job_node () |
| public | add_job_node () |
| public | draw_semaphore_and_accu () |
| public | add_semaphore_node () |
Function Documentation
◆ add_job_node()
| public add_job_node | ( | ) |
Undocumented method
Code:
click to view
sub add_job_node {
my $job = shift @_;
my $job_id = $job->dbID;
my $job_pipeline_name = $job->hive_pipeline->hive_pipeline_name;
my $job_node_name = 'job_'.$job_id.'__'.$job_pipeline_name;
unless($job_node_hash{$job_node_name}++) {
draw_job_node( $job, $job_node_name);
# recursion via child jobs:
if( !$stop_analysis or ($job->analysis != $stop_analysis) ) {
my $children = $job->adaptor->fetch_all_by_prev_job_id( $job_id );
foreach my $child_job ( @$children ) {
my $child_node_name = add_job_node( $child_job );
my $child_can_be_controlled = $child_job->fetch_local_blocking_semaphore;
unless( $self->{'suppress'} and $child_can_be_controlled ) {
$self->{'graph'}->add_edge( $job_node_name => $child_node_name,
color => 'blue',
);
}
}
# a local semaphore potentially blocked by this job:
if(my $controlled_semaphore = $job->controlled_semaphore) {
my $semaphore_node_name = add_semaphore_node( $controlled_semaphore );
my $job_status = $job->status;
my $parent_is_blocking = ($job_status eq 'DONE' or $job_status eq 'PASSED_ON') ? 0 : 1;
my $parent_controlling_colour = $parent_is_blocking ? 'red' : 'darkgreen';
my $blocking_arrow = $parent_is_blocking ? 'tee' : 'none';
$self->{'graph'}->add_edge( $job_node_name => $semaphore_node_name,
color => $parent_controlling_colour,
style => 'dashed',
arrowhead => $blocking_arrow,
);
}
}
}
return $job_node_name;
}
◆ add_semaphore_node()
| public add_semaphore_node | ( | ) |
Undocumented method
Code:
click to view
sub add_semaphore_node {
my $semaphore = shift @_;
my $semaphore_url = $semaphore->relative_url( 0 ); # request for an absolute URL
my $semaphore_id = $semaphore->dbID;
my $semaphore_pipeline_name = $semaphore->hive_pipeline->hive_pipeline_name;
my $semaphore_node_name = 'semaphore_'.$semaphore_id.'__'.$semaphore_pipeline_name;
unless($semaphore_url_hash{$semaphore_url}++) {
my ($accu_node_name, $target_cluster_name);
if(my $dependent_job = $semaphore->dependent_job) {
my $dependent_job_node_name = add_job_node( $dependent_job );
$accu_node_name = draw_semaphore_and_accu($semaphore, $dependent_job_node_name);
$target_cluster_name = $dependent_job->analysis->relative_display_name($main_pipeline);
if($dependent_job->analysis->hive_pipeline->hive_pipeline_name eq $main_pipeline->hive_pipeline_name) {
$target_cluster_name =~s{^.*/}{}; # workaround since we may have added $main_pipeline into TheApiary
} else {
$target_cluster_name =~s{/}{___};
}
} elsif(my $dependent_semaphore = $semaphore->dependent_semaphore) {
my $dependent_semaphore_node_name = add_semaphore_node( $dependent_semaphore );
$accu_node_name = draw_semaphore_and_accu($semaphore, $dependent_semaphore_node_name);
$target_cluster_name = $semaphore->hive_pipeline->hive_pipeline_name;
# can we trace the local blocking jobs up to their roots?
foreach my $start_job ( @{ find_the_top( $dependent_semaphore->fetch_my_local_controlling_jobs ) } ) {
my $job_node_name = add_job_node( $start_job );
}
} else { # The semaphore is not blocking anything, possibly the end of execution.
$accu_node_name = draw_semaphore_and_accu($semaphore, undef);
$target_cluster_name = $semaphore_pipeline_name;
}
# adding the semaphore node to the cluster of the dependent job's analysis:
push @{$self->{'graph'}->cluster_2_nodes->{ $target_cluster_name }}, $semaphore_node_name;
}
return $semaphore_node_name;
}
◆ draw_job_node()
| public draw_job_node | ( | ) |
Undocumented method
Code:
click to view
sub draw_job_node {
my ($job, $job_node_name) = @_;
my $job_shape = 'box3d';
my $job_status = $job->status;
my $job_status_colour = {'DONE' => 'DeepSkyBlue', 'READY' => 'green', 'SEMAPHORED' => 'grey', 'FAILED' => 'red'}->{$job_status} // 'yellow';
my $analysis_status_colour = {
"EMPTY" => "white",
"BLOCKED" => "grey",
"LOADING" => "green",
"ALL_CLAIMED" => "grey",
"SYNCHING" => "green",
"READY" => "green",
"WORKING" => "yellow",
"DONE" => "DeepSkyBlue",
"FAILED" => "red",
};
my $job_id = $job->dbID;
my $job_params = destringify($job->input_id);
my $job_label = qq{<<table border="0" cellborder="0" cellspacing="0" cellpadding="1">}
.qq{<tr><td><u><i>job_id:</i></u></td><td><i>$job_id</i></td></tr>};
if(my $param_id_stack = $job->param_id_stack) {
$job_label .= qq{<tr><td><u><i>params from:</i></u></td><td><i>$param_id_stack</i></td></tr>};
}
foreach my $param_key (sort keys %$job_params) {
my $param_value = $job_params->{$param_key};
$job_label .= "<tr><td>$param_key:</td><td> $param_value</td></tr>";
}
$job_label .= "</table>>";
$self->{'graph'}->add_node( $job_node_name,
shape => $job_shape,
style => 'filled',
fillcolor => $job_status_colour,
label => $job_label,
);
# adding the job to the corresponding analysis' cluster:
my $analysis_name = $job->analysis->relative_display_name($main_pipeline);
my $cluster_label;
if($analysis_name =~ m{^(\w+)/(\w+)$}) {
$analysis_name = $1 . '___' . $2;
$cluster_label = $2;
} else {
$cluster_label = $analysis_name;
}
my $analysis_status = $job->analysis->status;
push @{$self->{'graph'}->cluster_2_nodes->{ $analysis_name }}, $job_node_name;
$self->{'graph'}->cluster_2_attributes->{ $analysis_name }{ 'cluster_label' } = $cluster_label;
$self->{'graph'}->cluster_2_attributes->{ $analysis_name }{ 'style' } = 'rounded,filled';
$self->{'graph'}->cluster_2_attributes->{ $analysis_name }{ 'fill_colour_pair' } = [ $analysis_status_colour->{$analysis_status} ];
$analysis_name_2_pipeline{ $analysis_name } = $job->hive_pipeline;
}
◆ draw_semaphore_and_accu()
| public draw_semaphore_and_accu | ( | ) |
Undocumented method
Code:
click to view
sub draw_semaphore_and_accu {
my ($semaphore, $dependent_node_name) = @_;
my $semaphore_id = $semaphore->dbID;
my $semaphore_pipeline_name = $semaphore->hive_pipeline->hive_pipeline_name;
my $semaphore_node_name = 'semaphore_'.$semaphore_id.'__'.$semaphore_pipeline_name;
my $semaphore_blockers = $semaphore->local_jobs_counter + $semaphore->remote_jobs_counter;
my $semaphore_is_blocked = $semaphore_blockers > 0;
my $meta_shape = $self->{'show_accu_keys'}
? ['house', 'invhouse' ] # house shape hints that accu data will be shown if present
: ['triangle', 'invtriangle']; # triangle shape hints that no accu data will be shown even if present
my $columns_in_table = $self->{'show_accu_values'} ? 3 : 2;
my ($semaphore_shape, $semaphore_bgcolour, $semaphore_fgcolour, $dependent_blocking_arrow_colour, $dependent_blocking_arrow_shape ) = $semaphore_is_blocked
? ($meta_shape->[0], 'grey', 'brown', 'red', 'tee')
: ($meta_shape->[1], 'darkgreen', 'white', 'darkgreen', 'none');
my @semaphore_label_parts = ();
if($semaphore_is_blocked) {
if(my $local=$semaphore->local_jobs_counter) { push @semaphore_label_parts, "local: $local" }
if(my $remote=$semaphore->remote_jobs_counter) { push @semaphore_label_parts, "remote: $remote" }
} else {
push @semaphore_label_parts, "open";
}
my $semaphore_label = join(', ', @semaphore_label_parts);
my $accusem_label = qq{<<table border="0" cellborder="0" cellspacing="0" cellpadding="1">};
$accusem_label .= qq{<tr><td colspan="$columns_in_table"><font color="$semaphore_fgcolour"><b><i>$semaphore_label</i></b></font></td></tr>};
my %accu_ptrs = ();
if($self->{'show_accu_keys'}) {
my $raw_accu_data = $semaphore->fetch_my_raw_accu_data;
if(@$raw_accu_data) {
$accusem_label .= qq{<tr><td colspan="$columns_in_table"> </td></tr>}; # skip one table row between semaphore attributes and accu data
my %struct_name_2_key_signature_and_value = ();
foreach my $accu_rowhash (@$raw_accu_data) {
push @{ $struct_name_2_key_signature_and_value{ $accu_rowhash->{'struct_name'} } },
[ $accu_rowhash->{'key_signature'}, $accu_rowhash->{'value'}, $accu_rowhash->{'sending_job_id'} ];
}
my $sending_job_pipeline_name = $semaphore->hive_pipeline->hive_pipeline_name; # assuming cross-database links are currently not stored
foreach my $struct_name (sort keys %struct_name_2_key_signature_and_value) {
$accusem_label .= $self->{'show_accu_values'}
? qq{<tr><td></td><td><b><u>$struct_name</u></b></td><td></td></tr>}
: qq{<tr> <td><b><u>$struct_name</u></b></td><td></td></tr>};
my @sorted_values = sort {(($a->[2]//0) <=> ($b->[2]//0)) || ($a->[0] cmp $b->[0]) || ($a->[1] cmp $b->[1])} @{ $struct_name_2_key_signature_and_value{$struct_name} };
foreach my $accu_vector ( @sorted_values ) {
my ($key_signature, $value, $sending_job_id) = @$accu_vector;
$sending_job_id //= 0;
my $protected_value = $self->{'graph'}->protect_string_for_display($value);
my $port_label = "${semaphore_node_name}_${struct_name}_${sending_job_id}";
my $port_attribute = $sending_job_id ? qq{port="$port_label"} : '';
if(my $sending_job_node_name = 'job_'.$sending_job_id.'__'.$sending_job_pipeline_name) {
push @{ $accu_ptrs{$sending_job_node_name} }, $port_label;
}
$accusem_label .= $self->{'show_accu_values'}
? qq{<tr><td $port_attribute>$key_signature</td><td> <b>--></b> </td><td>$protected_value</td></tr>}
: qq{<tr><td $port_attribute></td><td>$key_signature</td></tr>};
}
}
}
}
$accusem_label .= "</table>>";
$self->{'graph'}->add_node( $semaphore_node_name,
shape => $semaphore_shape, # 'note',
margin => '0,0',
style => 'filled',
fillcolor => $semaphore_bgcolour,
label => $accusem_label,
);
if($dependent_node_name) {
$self->{'graph'}->add_edge( $semaphore_node_name => $dependent_node_name,
color => $dependent_blocking_arrow_colour,
style => 'dashed',
arrowhead => $dependent_blocking_arrow_shape,
tailport => 's',
headport => 'n',
);
}
if($self->{'show_accu_pointers'}) {
foreach my $sending_job_node_name (keys %accu_ptrs) {
foreach my $receiving_port (@{ $accu_ptrs{$sending_job_node_name} }) {
$self->{'graph'}->add_edge( $sending_job_node_name => $semaphore_node_name,
headport => $receiving_port.':w',
color => 'black',
style => 'dotted',
);
}
}
}
return $semaphore_node_name;
}
◆ find_the_top()
| public find_the_top | ( | ) |
Undocumented method
Code:
click to view
sub find_the_top {
my ($anchor_jobs) = @_;
my @starters = ();
# first try to find the start_analysis on the way up:
foreach my $anchor_job ( @$anchor_jobs ) {
push @starters, trace_job_up($anchor_job);
my $children = $anchor_job->adaptor->fetch_all_by_prev_job_id( $anchor_job->dbID );
foreach my $child ( @$children ) {
if( my $semaphore = $child->fetch_local_blocking_semaphore ) {
push @starters, trace_semaphore_up( $semaphore );
}
}
}
return \@starters;
}
◆ main()
| public main | ( | ) |
◆ precache_participating_pipelines()
| public precache_participating_pipelines | ( | ) |
Undocumented method
Code:
click to view
my @pipelines_to_check = @_;
my %scanned_pipeline_urls = ();
while( my $current_pipeline = shift @pipelines_to_check ) {
my $current_pipeline_url = $current_pipeline->hive_dba->dbc->url;
unless( $scanned_pipeline_urls{ $current_pipeline_url }++ ) {
foreach my $df_target ( $current_pipeline->collection_of('DataflowTarget')->list ) {
# touching it for the side-effect of loading it to TheApiary:
my $target_object_pipeline = $df_target->to_analysis->hive_pipeline;
my $target_pipeline_url = $target_object_pipeline->hive_dba->dbc->url;
unless(exists $scanned_pipeline_urls{$target_pipeline_url}) {
push @pipelines_to_check, $target_object_pipeline;
}
}
}
}
}
◆ trace_job_up()
| public trace_job_up | ( | ) |
Undocumented method
Code:
click to view
sub trace_job_up {
my $job = shift @_;
my @buffer = ();
if(my $local_blocking_semaphore = $job->fetch_local_blocking_semaphore) {
push @buffer, trace_semaphore_up( $local_blocking_semaphore );
}
if(my $local_parent_job = $job->prev_job) {
push @buffer, trace_job_up( $local_parent_job );
} else {
push @buffer, $job;
}
return @buffer;
}
◆ trace_semaphore_up()
| public trace_semaphore_up | ( | ) |
Undocumented method
Code:
click to view
sub trace_semaphore_up {
my $semaphore = shift @_;
my @buffer = ();
foreach my $local_controlling_job ( @{ $semaphore->fetch_my_local_controlling_jobs } ) {
push @buffer, trace_job_up( $local_controlling_job );
}
foreach my $remote_semaphore ( @{ Bio::EnsEMBL::Hive::TheApiary->fetch_remote_semaphores_controlling_this_one( $semaphore, $main_pipeline ) } ) {
push @buffer, trace_semaphore_up( $remote_semaphore );
}
return @buffer;
}
