PERL - многопоточность - дети + внуки - предел потоков работает

Question

Я создал скрипт Perl, где я создаю потоки (ограниченные по потокам, выполняемые в то же время), и каждый поток создает свои собственные дети, которые также должны быть ограничены числом ,

Где я размещаю свой скрипт, я не могу запускать больше, чем потоки X на Perl-скрипт. В приведенном ниже примере у меня есть X = 3 x 7 = 21 поток максимум за это время.

• 3 для 1-го задания ($nb_process_first)
• 7 для 2-го задания ($nb_process_second)

Вопросы:

1. Есть ли лучший способ управлять потоками и их детей ? (например, очереди - не могли бы вы привести мне пример кода, потому что я пробовал без успеха)
2. Мой текущий скрипт не заканчивается со всеми связанными потоками, хотя я использую цикл для всех запущенных потоков, чтобы присоединиться к ним (cf в конце скрипта).

#!/usr/bin/perl -s 

use threads; 

my @threads; 
my $nb_process_first = 3; 
my @running   =(); 

print "START" . "\n"; 

$current = 1; 
while ($current <= 10) { 
    @running = threads->list(threads::running); 
    if (scalar @running < $nb_process_first) { 
     print "Launch firstJob=" . scalar @running . "\n"; 
     my $thread = threads->create(\&firstJob); 
     push(@threads, $thread); 
    } else { 
     redo; 
    } 
    $current++; 
} 

my @joinable = threads->list(threads::joinable); 
while (scalar @joinable != 0) { 
    foreach my $thr (threads->list()) { 
     $thr->join(); 
    } 
    @joinable = threads->list(threads::joinable); 
} 

print "END" . "\n"; 

sub secondJob { 
    for ($i = 0; $i <= 15; $i++) { 
     print "secondJob=" . $i . "\n"; 
     sleep 1; 
    } 
    threads->exit(); 
} 

sub firstJob { 
    my $nb_process_second = 7; 
    my @running   =(); 
    $current = 1; 
    while ($current <= 10) { 
     @running = threads->list(threads::running); 
     if (scalar @running < $nb_process_second) { 
      print "firstJob/Launch secondJob=" . scalar @running . "-" . $current . "\n"; 
      my $secondthread = threads->create(\&secondJob); 
      push(@threads, $secondthread); 
      sleep 2; 
     } 
     $current++; 
    } 
    threads->exit(); 
} 

Answer 1

Thread::Queue - удобная модель для базовой резьбы резьбы по резьбе.

Это идет немного как это:

#!/usr/bin/perl 
use strict; 
use warnings; 
use threads; 
use Thread::Queue; 

my $firstworkitem_q = Thread::Queue -> new(); 
my $secondworkitem_q = Thread::Queue -> new(); 
my $nthreads = 10; 

sub first_worker { 
    while (my $item = $firstworkitem_q -> dequeue()) { 
     print "First worker picked up $item, and queues it to second worker\n"; 
     $secondworkitem_q -> enqueue ($item); 
    } 
} 

sub second_worker { 
    while (my $item = $secondworkitem_q -> dequeue()) { 
     print "Second worker got $item"; 
    } 
} 

my @first_workers; 
for (1..$nthreads) { 
    my $thr = threads -> create (\&first_worker); 
    push (@first_workers, $thr); 
} 

for (1..$nthreads) { 
    my $thr = threads -> create (\&second_worker); 
} 

$firstworkitem_q -> enqueue (@things_to_processs); 
$firstworkitem_q -> end; 

foreach my $firstworker (@first_workers) { 
    $firstworker -> join(); 
} 

#here all the first workers have finished, so we know nothing will be queued to second work queue. 
$secondworkitem_q -> end(); 

foreach my $thr (threads -> list()) { 
    $thr -> join(); 
} 

Вы запихнуть вещи в очереди, и перебирать на нем для обработки. Когда вы end очереди, цикл while получает undef и, таким образом, заканчивается - делая вашу нить соединяемой.

Вам не нужно отслеживать @running так, как вы это делаете, потому что threads -> list() сделает это. И что еще более важно - вам нужно будет @running общей переменной и заблокировать ее, потому что в противном случае у вас есть другая копия в каждом потоке.

firstJob spawn secondJob Я бы уклонился, потому что он может создавать всевозможные фруктовые ошибки. Я предлагаю нерест двух классов рабочего потока. Используйте $queue -> end(), чтобы запустить первый набор рабочих.

Answer 2

Что касается вашего второго вопроса, то нить только объединяемая если они закончили работу (см this answer). Поскольку некоторые из потоков не выполняются при выполнении цикла while, он заканчивается без их соединения.

Ваш цикл должен опираться на количество активных потоков, а не количество соединяемых потоков. Что-то вроде этого:

while (threads->list() > 0) 
{ 
    foreach my $joinable (threads->list(threads::joinable)) 
    { 
     $joinable->join(); 
    } 
} 

Что касается первого вопроса, есть, конечно, другие способы управления потоками. Однако невозможно сказать, что вы должны делать, не зная своей задачи.