File Coverage

lib/NBI/Pipeline.pm

Criterion	Covered	Total	%
statement	45	47	95.7
branch	10	16	62.5
condition	2	5	40.0
subroutine	8	8	100.0
pod	4	4	100.0
total	69	80	86.2

package NBI::Pipeline;
#ABSTRACT: Ordered list of NBI::Job objects with dependency wiring
#
# NBI::Pipeline - Minimal multi-job orchestration for nbilaunch.
#
# DESCRIPTION:
#   Holds an ordered list of NBI::Job objects.  When run() is called, jobs are
#   submitted in order.  If a job has _nbi_depends_on set to a previous job
#   object, the afterok dependency is injected into its NBI::Opts before
#   submission.
#
#   This is intentionally minimal in v1.  It does not resolve filenames between
#   jobs - complex launchers handle that via NBI::Manifest->load().
#
# USAGE IN A LAUNCHER:
#   sub build {
#       my ($self, %args) = @_;
#       my ($job1, $m1) = $self->_build_step1(%args);
#       my ($job2, $m2) = $self->_build_step2(%args);
#       $job2->{_nbi_depends_on} = $job1;
#       return (NBI::Pipeline->new(jobs => [$job1, $job2]), $m1);
#   }
#
# nbilaunch checks ref($result): NBI::Job → single submit, NBI::Pipeline → run().
#

use 5.012;
use strict;
use warnings;
use Carp qw(confess);

$NBI::Pipeline::VERSION = $NBI::Slurm::VERSION;

sub new {
    my ($class, %args) = @_;

    my $jobs = $args{jobs} // [];
    ref $jobs eq 'ARRAY'
        or confess "ERROR NBI::Pipeline: 'jobs' must be an arrayref\n";

    for my $j (@$jobs) {
        $j->isa('NBI::Job')
            or confess "ERROR NBI::Pipeline: each job must be an NBI::Job instance\n";
    }

    return bless { jobs => [@$jobs] }, $class;
}

# ── add_job($job) ─────────────────────────────────────────────────────────────
sub add_job {
    my ($self, $job) = @_;
    $job->isa('NBI::Job')
        or confess "ERROR NBI::Pipeline: job must be an NBI::Job instance\n";
    push @{ $self->{jobs} }, $job;
    return $self;
}

# ── run() ─────────────────────────────────────────────────────────────────────
# Submit all jobs in order.  Wires afterok dependencies automatically.
# Returns a list of Slurm job IDs (integers).
sub run {
    my ($self) = @_;

    my @job_ids;
    my %submitted;   # NBI::Job object → job_id (for dependency lookup)

    for my $job (@{ $self->{jobs} }) {
        # Inject afterok dependency if this job depends on a previous one
        if (my $dep_job = $job->{_nbi_depends_on}) {
            my $dep_id = $submitted{$dep_job}
                or confess "ERROR NBI::Pipeline: dependency job was not yet submitted\n";
            $job->opts->add_option("--dependency=afterok:$dep_id");
        }

        # Ensure the provenance directory exists before NBI::Job->run() writes there
        my $tmpdir = $job->opts->tmpdir;
        if ($tmpdir && !-d $tmpdir) {
            require File::Path;
            File::Path::make_path($tmpdir)
                or confess "ERROR NBI::Pipeline: cannot create tmpdir '$tmpdir': $!\n";
        }

        my $job_id = $job->run();
        $submitted{$job} = $job_id;
        push @job_ids, $job_id;
    }

    return @job_ids;
}

# ── print_summary() ───────────────────────────────────────────────────────────
# Human-readable dependency graph.
sub print_summary {
    my ($self) = @_;
    my @lines = ("Pipeline: " . scalar(@{ $self->{jobs} }) . " job(s)");
    for my $job (@{ $self->{jobs} }) {
        my $line = "  [" . $job->name . "]";
        if (my $dep = $job->{_nbi_depends_on}) {
            $line .= "  afterok:[" . $dep->name . "]";
        }
        push @lines, $line;
    }
    print join("\n", @lines) . "\n";
}

1;

__END__