How can I handle scheduling threads with dependencies in Perl?

I have the following scenario:

sub_1 can run immediately
sub_2 can run immediately
sub_3 can run only after sub_1 finishes
sub_4 can run only after sub_1 fini开发者_运维问答shes
sub_5 can run only after sub_2 finishes
sub_6 can run only after sub_2 finishes
sub_7 can run only after both sub_1 and sub_2 finish
sub_8 can run only after both sub_1 and sub_2 finish

I would like each sub to start run as soon as possible, than wait for all of them to finish.

I would really appreciate you help in creating a clean solution for this simple scenario -- I'm new to multi-threading.

I'm not sure if it makes a difference, but those subs are all in an object.

---

I'd suggest a "Boss/Worker" model, wherein one thread manages the subroutines to be executed in worker threads, who in turn report their status back to the boss upon completion.

In this model the boss is the only thread that needs to know how tasks are to be ordered. It might look something like this:

use threads;
use Thread::Queue;
use Thread::Pool;

our $done_queue = Thread::Queue->new;
our $work_pool  = Thread::Pool->new;

sub sub_1 {
  ... do the work ...
  $done_queue->enqueue('sub_1'); # tell the boss we're all done
}

sub sub_2 {
  ... do the work ...
  $done_queue->enqueue('sub_2'); # tell boss we're done
}

...

# Main loop (boss thread)

$work_pool->enqueue(\&sub_1);
$work_pool->enqueue(\&sub_2);

while (my $sub_name = $done_queue->dequeue) {
  # You, the boss thread, keep track of state and
  # transitions however you like.  You know what's
  # just finished and what's finished in the past
  ...
}

Of course, abstraction can make that neater -- you could hide the Pool and the Queue behind a single object, one which didn't require sub_1() to know about the status queue at all:

$boss->enqueue( 'sub_1' => \&sub_1 ); # Will return 'sub_1' via await_completed()
$boss->enqueue( 'sub_2' => \&sub_2 ); # Will return 'sub_1'

while (my $sub_name = $boss->await_completed) {
  ...
}

---

Here's a possible solution using threads and thread sharing. Most of the code is just mocking up the test and emulating threads that have to do "work" before they finish. In the example the main thread spawns seven threads that each have a random amount of time that they have to do "work". The threads cannot begin working until the other threads they are dependent on (set in the dependencies array) have finished. You can change the thread dependencies and run the example a few times to illustrate that it works correctly.

Additionally you can have each thread terminate after it runs and have the main thread terminate after all of the subthreads have finished by checking the status hash.

use strict;
use warnings;
use threads;
use threads::shared;


my %status : shared;

my $dependencies = [
                    {3 => 1},   #three can only run after one has finished...
                    {4 => 1},   #four can only run after one has finished...
                    {5 => 2},   #five can only run after two has finished...
                    {6 => 1},   #etc...
                    {6 => 2},
                    {7 => 1},
                    {7 => 2}
                   ];

main();

sub main{
    foreach my $thread_number (1..7){
        spawn_thread($thread_number);
    }

    while(1){
        print "I am the main thread\n";
        sleep(1);
    }
}

sub spawn_thread{
    my $thread_number = shift;
    $status{$thread_number} = 'wait';
    my $thr = threads->new(\&thread_routine, $thread_number); 

}

sub thread_routine{
    my $thread_number = shift;

    my $working_time_left =  int(rand(5)) + 1;    #make a random time that this thread needs to "work"

    while(1){
        print "I am thread number $thread_number with status $status{$thread_number}\n";
        {
            lock(%status);

                    #see if this thread is active;  if so, see if it finished running running
            if ($status{$thread_number} eq 'active'){
                if ($working_time_left <= 0){
                    $status{$thread_number} = 'ran';
                }
            }
            else{       
                            #see if we can activate                 
                if ($status{$thread_number} eq 'wait'){
                    my $can_activate = 1;
                    foreach my $index (0..$#$dependencies){
                        if (exists $dependencies->[$index]->{$thread_number}){
                            if ($status{$dependencies->[$index]->{$thread_number}} ne 'ran'){
                                $can_activate = 0;
                                last;
                            }
                        }
                    }
                    if ($can_activate){                 
                        $status{$thread_number} = "active";
                    }
                }

            }
        }

        sleep(1);

        if ($status{$thread_number} eq 'active'){   #do "work"
            $working_time_left--;
        }
    }
}

---

Fork and create 2 processes:

In process 1:

sub_1; sub_3

In process 2:

sub_2; wait for sub_1 end; sub_4