Threaded Application Sequencing/Rendezvous

learnedbyerror has asked for the wisdom of the Perl Monks concerning the following question:

Oh learned monks,

Yet again, I need your assistance. I am looking for something similar to the IBM JCL of my youth. I have a moderately complex application sequence containing 11 steps that I am currently running serially. I have written 4 of these steps to be multi-threaded and have greatly reduced my run time. I am currently looking for an elegant solution to allow me to run some of the other steps in parallel, to further reduce run time, and to allow me to test the steps for success and either restart or abend.

The step sequence is shown in the following ASCII diagram:

==ACTIONS= ===============TIME==============
           0   1   2   3   4   5   6   7   8
Action  1  |-->|
Action  1a     |-->|
Action  2  |-->|
Action  2a     |-->|
Action  3          |-->|
Action  4              |-->|
Action  5                  |-->|
Action  6                      |-->|
Action  7                  |------>|
Action  8                          |-->|
Action  9                              |-->|
Action 10          |-->|
Action 11                              |-->|

I can currently achieve my functional goals through writing a hard coded script to carry this out; however, this is not elegant and would contain quite a bit of code itself.

I have spent many hours over the last couple of months looking through CPAN trying to find a solution that would let me model the dependencies and have the orchestration of the steps handled via a job management system. While I have found some possible solutions such as Gearman and BPM::Engine, the solutions are either overkill and/or more complex to setup and maintain than a hand tooled script.

My ideal solution would contain the following characteristics:

Configured via a markup language file
Provide a way to query current status and runtime statistics from the steps being executed
Be a turnkey solution that does not required a lot of work to install on a new system

Thank You in advance for sharing your wisdom with me

lbe

Update:

After a couple of cups of coffee, a couple of sheets of scratch paper, and an hour in vi, I have an approach which I am going to code up. It is a combination of the recommendations of BrowserUK and robiticus.

I am going to store the configuration using perl reference syntax. This is a little more verbose than other alternatives; however, it will eliminate some coding; I may change this in the future if proves hard to support.

I have roughed out the beginnings of a base object that will contain all of the logic to run the job (i.e. ready, running, complete, abend, ...) I will sub-class this object to handle updating a global refernce with the current state/running information on each step to support the reporting.

And lastly, I will use a main loop to iterate over the list of objects calling ready and done methods for each object. When ready is called, the object will exit immediately if it is in the running or done state. Otherwise, it will check the status of its predecessors in the global reference.and will run the step if the predecessors are complete or set status to cancelled if the predecessors have abended or cancelled. Subsequents calls to the complete method will update its status when the step ends, successfully or unsuccesfully.

I am going to hold off on the reporting side until I get the basic above solid. Then I'll add a small RPC server that will dump the global reference to the caller. I can then build a small separate program to support its monitoring.

Thanks for the ideas!! lbe

Update 2:

After letting the recommendations sink in, I crafted/hacked a solution combining the input from BrowserUK and roboticus. I created an object that contains the minimal information needed to to sequence the job steps. My intent is to sub-class this as needed to address any needs specific to a specific job step - such as incremental run-time statistics for some of the heavily threaded longer running steps. I created a script that uses my object that contains a main loop and and sub-routines to load the sequence data and a do-nothing routine to use in the test.

Please take a look and provide any feedback that you may have.

Thanks, lbe

The Module (Step::Base)

package Step::Base;

use v5.10.1;

use threads;
use threads::shared;

use namespace::autoclean;

#use feature qw( switch );

use Moose;

has 'name' => (
    is       => 'ro',
    isa      => 'Str',
    required => 1,
);

has 'code_to_run' => (
    is       => 'ro',
    isa      => 'Str',
    required => 1,
);

has 'args' => (
    is  => 'ro',
    isa => 'ArrayRef',
);

has 'return_code' => (
    is  => 'rw',
    isa => 'ArrayRef',
);

has 'state' => (
    is      => 'rw',
    isa     => 'Str',
        default => 'Created',
);

has 'success' => (
    is  => 'rw',
    isa => 'Int',
);

has 'error' => (
    is  => 'rw',
    isa => 'Str',
);

has 'time_started' => (
    is  => 'rw',
    isa => 'Int',
);

has 'time_finished' => (
    is  => 'rw',
    isa => 'Int',
);

has 'success' => (
    is  => 'rw',
    isa => 'Int',
);

has 'predecessors' => (
    is  => 'ro',
    isa => 'ArrayRef',
);

has 'thr' => (
    is  => 'rw',
    isa => 'Object',
);

has 'steps' => (
        is => 'ro',
);

sub run {
    my ( $self, ) = @_;

        $self->time_started(time);
    my $thr = threads->create( $self->code_to_run, $self->args );
    if ( $thr->error ) {
                $self->state('Abended');
        $self->error(@$);
                $self->time_finished(time);
                $self->success(0);
        return( 0 );
    }
    else {
        $self->thr($thr);
        $self->success(1);
        $self->state('Running');
        return( 1 );
    }
} ## ---------- end sub run

sub ready {
    my ( $self, ) = @_;

        return(0) unless $self->state eq 'Created';

        $self->_predecessors_complete( $self->predecessors ) ? return(
+1) : return(0);
} ## ---------- end sub ready

sub running {
    my ( $self, ) = @_;

        $self->state eq "Running" ? return(1) : return(0)
} ## ---------- end sub running

sub done {
    my ( $self, ) = @_;

    # check to see if thread is ready to join and join if it is
        given( $self->state ) {
        when( 'Running' ) {
                my @joinable = threads->list(threads::joinable);
                foreach my $thr (@joinable) {
                next unless ( $thr->tid == $self->thr->tid );
                my $rc = $thr->join;
                if ( $rc->{success} ) {
                        $self->state('Completed');
                }
                else {
                        $self->state('Abended');
                                        my $err_msg;
                                        if ( my $thread_error = $thr->
+error ) {
                                                $err_msg = "$thread_er
+ror\n"
                                        }
                                        $err_msg .= $rc->{err_msg};
                        $self->error($err_msg);
                }
                                $self->time_finished(time);
                return (1);
                }
                return (0);
                }
        when( 'Completed' ) {
                        return(1);
                }
                when( 'Abended' ) {
                return(1);
        }
        when( 'Cancelled' ) {
                return(1);
        }
        default {
                return (0);
        }
        }
} ## ---------- end sub done

sub _predecessors_complete {
    my ( $self, $predecessors ) = @_;

    foreach my $p ( @{ $predecessors } ) {
                #given( $main::steps->{ $p }->state ) {
                given( $self->steps->{ $p }->state ) {
                        when ("Complete") {
                                next;
                        }
                        when ("Created") {
                                return(0);
                        }
                        when ("Running") {
                                return(0);
                        }
                        when ("Cancelled") {
                                $self->state("Cancelled");
                                return(0);
                        }
                        when ("Abended") {
                                $self->state("Cancelled");
                                return(0);
                        }
                }
    }
    return ( 1 );
} ## ---------- end sub _predecessors_status

sub time_elapsed {
    my ( $self, ) = @_;

    given( $self->state ) {
                when( 'Running' ) {
                return ( time - $self->time_started );
        }
        when( 'Completed' ) {
                return ( $self->time_finished - $self->time_started );
        }
        default {
                return (undef);
        }
        }
} ## ---------- end sub time_elapsed

1;

__PACKAGE__->meta->make_immutable;
[download]

The control script

use strict;
use warnings;

use v5.10.1;

use Step::Base;

my ( $steps, $step_names ) = load_steps();

while (1) {
    my $cnt_not_ready = 0;
    my $cnt_running   = 0;
    my $cnt_done      = 0;

        FOR: foreach my $name ( @{$step_names} ) {

        if ( $steps->{$name}->done ) {
            $cnt_done++;
        }
        elsif ( $steps->{$name}->state eq 'Created' ) {
            if ( $steps->{$name}->ready ) {
                $steps->{$name}->run ?  $cnt_running++ : $cnt_done++;
            }
            else {
                $cnt_not_ready++;
            }
        }
        elsif ( $steps->{$name}->running ) {
            $cnt_running++;
        }
                else {
                        die "Something ain't right!\n"
                                . "$name\-\>state = " . $steps->{$name
+}->state . "\n"
                                . "\tRunning: " . $steps->{$name}->run
+ning . "\n"
                                . "\t   Done: " . $steps->{$name}->don
+e . "\n";
                }

        printf( "%-15s %s\n", $steps->{$name}->state, $name );
    }

    printf( "Not Ready: %d  Running: %d  Done: %d\n", $cnt_not_ready, 
+$cnt_running, $cnt_done );

        last if ( $cnt_done >= scalar( @{$step_names} ) );

    sleep(5);
}

sub load_steps {
    my $steps;
    my $step_names;

    my $conf = define_steps_conf();

    foreach my $step ( @{$conf} ) {
        my ( $name, $args ) = each %{$step};

        $args->{name}  = $name;
        $args->{steps} = $steps;

        my $s = Step::Base->new($args);

        $steps->{$name} = $s;
        push( @{$step_names}, $name );

        1;

    }

    return ( $steps, $step_names );
} ## ---------- end sub load_steps

sub define_steps_conf {
    my $steps = [
        {
            'Action01' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [],
            }
        },
        {
            'Action01a',
            => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action01', ],
            }
        },
        {
            'Action02' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [],
            }
        },
        {
            'Action02a' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action02', ],
            }
        },
        {
            'Action03' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action01a', 'Action02a', ],
            }
        },
        {
            'Action04' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action03', ],
            }
        },
        {
            'Action05' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action04', ],
            }
        },
        {
            'Action06' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action05', ],
            }
        },
        {
            'Action07' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action05', ],
            }
        },
        {
            'Action08' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action06', 'Action07' ],
            }
        },
        {
            'Action09' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action08', ],
            }
        },
        {
            'Action10' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action01a', 'Action02a' ],
            }
        },
        {
            'Action11' => {
                code_to_run  => 'main::do_nothing',
                predecessors => [ 'Action08', ],
            }
        },
    ];
    return ($steps);
} ## ---------- end sub define_steps_conf

sub do_nothing {
        sleep int( rand(5) + 5.5 );
    if ( int( rand() + 0.9 ) ) { # fail 10% of the time
        return ( { success => 1 } );
    }
    else {
        return ( { success => 0, err_msg => "I screw up" } );
    }
} ## ---------- end sub do_nothing
[download]

Back to Seekers of Perl Wisdom