Как я могу обработать планирование потоков с зависимостями в Perl?

Question

У меня есть следующий сценарий:

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 finishes
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

Я бы хотел, чтобы каждый сабвуфер запускался как можно скорее, а не ждал, пока все они закончат.

Буду очень признателен за помощь в создании чистого решения для этого простого сценария - я новичок в многопоточности.

Я не уверен, имеет ли это значение, но все эти субмарины находятся в объекте.

Answer 1

Я бы предложил модель «Босс / Рабочий», в которой один поток управляет подпрограммами, выполняемыми в рабочих потоках, которые, в свою очередь, сообщают о своем статусе боссу после завершения.

В этой модели босс является единственной нитью, которая должна знать, как следует заказывать задачи.Это может выглядеть примерно так:

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
  ...
}

Конечно, абстракция может сделать это аккуратнее - вы можете спрятать Пул и Очередь за одним объектом, для которого не нужно знать sub_1()о статусной очереди вообще:

$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) {
  ...
}

Answer 2

Вот возможное решение, использующее потоки и разделение потоков.Большая часть кода просто макетирует тест и эмулирует потоки, которые должны выполнить «работу» до завершения.В этом примере основной поток порождает семь потоков, каждый из которых имеет случайное количество времени, которое они должны выполнять «работу».Потоки не могут начать работать, пока другие потоки, от которых они зависят (установленные в массиве зависимостей), не закончили.Вы можете изменить зависимости потоков и запустить пример несколько раз, чтобы проиллюстрировать, что он работает правильно.

Кроме того, каждый поток можно завершить после его запуска, а основной поток - после завершения всех подпотоковпроверка статуса хэша.

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--;
        }
    }
}

Answer 3

Форкируйте и создайте 2 процесса:

В процессе 1:

sub_1; sub_3

В процессе 2:

sub_2; wait for sub_1 end; sub_4