Как разветвляться и читать из нескольких дочерних процессов?

Question

Моя главная цель - выполнить некоторую (внешне дорогостоящую) работу для большого списка объектов. Для этого, если я сделаю это прямо, это займет много времени. Поэтому я решил перейти в параллельный режим и разветвить некоторые (4-8, давайте посмотрим) дочерние процессы, каждый из которых выполняет свою работу для меньшего набора объектов. В основном (родительском) процессе я хотел распечатать ту же общую статистическую информацию о прогрессе, которую я получил для однопроцессной версии.

Однако, когда я разветвляюсь на 4 дочерних процесса и выполняю в них какую-то работу, я вижу, что они живы, но только один из них действительно что-то делает и отправляет информацию родителю.

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

#!/usr/bin/env perl
use strict;
use warnings;

use DateTime;
use DateTime::Format::HTTP;
use Time::HiRes;

my @to_be_processed = (1..300000);
my @queues;
my $nprocs = 4;

my $parent_from_child;
my @child_from_parent;
my @child_to_parent;

$SIG{CHLD} = 'IGNORE';
$|=1; # autoflush

my %stat = (
    total           => scalar(@to_be_processed),
    processed       => 0,
    time_started    => [Time::HiRes::gettimeofday],
);

# divide the list into queues for each subprocess
for (my $i = 0; $i < $stat{total}; $i++ ) {
    my $queue = $i % $nprocs;
    push @{$queues[$queue]}, $to_be_processed[$i];
}

# for progress simulation
srand (time ^ $$);

for (my $proc = 0; $proc < $nprocs; $proc++) {

    # set up the pipes
    pipe $parent_from_child, $child_to_parent[$proc]        or die "pipe failed - $!";

    # fork
    defined(my $pid = fork) or die "fork failed - $!";

    if ($pid) {
        # parent
        close $child_to_parent[$proc];
        printf("[%u] parent says: child %u created with pid %u\n", $$, $proc, $pid);
    }
    else {
        # child
        close $parent_from_child;
        open(STDOUT, ">&=" . fileno($child_to_parent[$proc]))   or die "open failed - $!";

        warn(sprintf("[%u] child alive with %u entries\n", $$, scalar(@{$queues[$proc]})));

        foreach my $id (@{$queues[$proc]}) {
            printf("START: %s\n", $id);

            # simulation of progress
            my $random_microseconds = int(rand(3000000))+200000;
            warn(sprintf("[%u] child 'works' for %u microseconds", $$, $random_microseconds));
            Time::HiRes::usleep( $random_microseconds );

            printf("DONE\n")
        }
        exit(0);
    }
}

# parent: receive data from children and print overall statistics
while (<$parent_from_child>) {
    chomp(my $line = $_);

    if ($line =~ m/^START: (\S+)/) {
        my ($id) = @_;

        printf("%6u/%6u", $stat{processed}, $stat{total});
        if ($stat{time_avg}) {
            my $remaining = ($stat{total} - $stat{processed}) * $stat{time_avg};
            my $eta = DateTime->from_epoch( epoch => time + $remaining );
            $eta->set_time_zone('Europe/Berlin');
            printf(" (ETA %s)", DateTime::Format::HTTP->format_isoz($eta));
        }
        printf("\r");
    }
    elsif ($line =~ /^DONE/) {
        $stat{processed}++;
        $stat{time_processed} = Time::HiRes::tv_interval( $stat{time_started} );
        $stat{time_avg}       = $stat{time_processed} / $stat{processed};
    }
    else {
        printf("%s\n", $line);
    }
}

Обычно предупреждение должно быть устранено. Если вы запустите его, вы увидите, что работает только один ребенок. Мой вопрос: почему? Где моя ошибка и как я могу заставить их всех делать свою работу?

Спасибо К.

Answer 1

Вы можете запустить Perl в режиме strace, и вы обнаружите, что жизни ваших детей довольно коротки, и выглядят так:

close(3)                                = 0
ioctl(4, SNDCTL_TMR_TIMEBASE or TCGETS, 0x7fff753b3a10) = -1 EINVAL (Invalid argument)
lseek(4, 0, SEEK_CUR)                   = -1 ESPIPE (Illegal seek)
fstat(4, {st_mode=S_IFIFO|0600, st_size=0, ...}) = 0
dup2(4, 1)                              = 1
dup(4)                                  = 3
fcntl(4, F_GETFD)                       = 0x1 (flags FD_CLOEXEC)
dup2(3, 4)                              = 4
fcntl(4, F_SETFD, FD_CLOEXEC)           = 0
close(3)                                = 0
fcntl(1, F_SETFD, 0)                    = 0
write(2, "[30629] child alive with 75000 e"..., 39) = 39
brk(0x3582000)                          = 0x3582000
write(1, "START: 1\n", 9)               = -1 EPIPE (Broken pipe)
--- SIGPIPE (Broken pipe) @ 0 (0) ---

Вот почему:

pipe $parent_from_child, $child_to_parent[$proc]        or die "pipe failed - $!";

Вы использовали массив с неверным аргументом для конвейера.Вы должны держать все стороны read открытыми в родительском элементе.Вместо этого вы установили массив, чтобы родитель мог оставить все стороны записи открытыми (но затем в родительском блоке вы немедленно закрыли сторону записи).Поэтому в следующий раз в вашем цикле pipe создаст новый дескриптор, назначит его на $parent_from_child.Таким образом, старое значение больше не имеет ссылок, и Perl очищает его, то есть закрывает дескриптор файла.Так что ваши дети, за исключением последнего, все умирают от SIGPIPE.

Я думаю, у вас сложилось впечатление, что вы можете повторно использовать этот дескриптор чтения и просто назначить ему несколько дескрипторов записи.Ты не можешьpipe всегда создает новый дескриптор чтения и новый дескриптор записи.

Если вы действительно хотите использовать один и тот же дескриптор чтения (вероятно, нет, это приведет к повреждению, когда чередуется вывод двух клиентов), просто создайте его один раз, вне цикла.Все дети наследуют один и тот же дескриптор записи через fork.Скорее всего, вам нужно по одному на каждого ребенка, и вам придется использовать цикл select, чтобы увидеть, какие из них имеют выходную информацию, и прочитать их.

В качестве альтернативы, я уверен, что CPAN имеет готовыйрешение (или десять) для вас.