Perl MCE并行处理二维数组：代码修复及优化方案咨询

问题说明 #

尝试用Perl的MCE模块并行处理二维数组（AoA），完成复制并修改数组的操作，但现有代码因索引错误、共享变量处理不当无法运行，以下是修复方案及优化思路。

原代码错误分析 #

1. 索引占位符无效：代码中的?是无意义占位符，无法对应原数组的行索引，导致结果数组无法正确写入。
2. 共享变量处理错误：$modified_table、$data_type、$max_length是父进程变量，MCE子进程拥有独立内存空间，直接修改这些变量不会同步到父进程；同时多进程并发修改$max_length会引发数据竞争，导致结果不准确。

修复后的代码 #

使用MCE的gather机制收集子进程处理结果，避免共享变量的竞争问题，同时正确映射原数组的行索引：

#!/usr/bin/env perl
use warnings;
use 5.14.0;
use Data::Dumper;
use Scalar::Util qw(looks_like_number);
use MCE;

sub get_string {
    my @chars = ( 'a'..'z', 'A'..'Z', '0'..'9', ' ' );
    return join '', map { @chars[rand @chars] } 1 .. rand(20) + 1;
}

# 生成测试二维数组
my $table = [];
for my $row (0 .. 9) {
    $table->[$row] = [ map { get_string } 0 .. 5 ];
}

my $modified_table = [];
my $data_type = [];
my $max_length = 0;

my $mce = MCE->new(
    chunk_size  => 5,
    input_data  => $table,
    user_func   => sub {
        my ($mce, $chunk_ref, $chunk_id) = @_;
        my $chunk_start = $mce->chunk_begin(); # 获取当前chunk在原数组中的起始行索引
        my $local_max = 0;
        my @local_modified;
        my @local_types;

        for my $row_idx (0 .. $#$chunk_ref) {
            my $original_row = $chunk_start + $row_idx;
            my @mod_row;
            my @type_row;

            for my $col (0 .. $#{$chunk_ref->[$row_idx]}) {
                my $modified_str = $chunk_ref->[$row_idx][$col] =~ s/\s/_/gr;
                push @mod_row, $modified_str;
                push @type_row, looks_like_number($modified_str) ? 1 : 0;

                my $len = length($modified_str);
                $local_max = $len if $len > $local_max;
            }

            # 收集当前行的结果和原索引
            push @local_modified, [ $original_row, \@mod_row ];
            push @local_types, [ $original_row, \@type_row ];
        }

        # 将当前chunk的结果和局部最大值传递给父进程
        MCE->gather(\@local_modified, \@local_types, $local_max);
    },
)->run();

# 汇总子进程的结果
while (my ($mod_data, $type_data, $local_max) = MCE->gather()) {
    # 更新全局最大长度
    $max_length = $local_max if $local_max > $max_length;

    # 填充modified_table
    for my $row_entry (@$mod_data) {
        my ($orig_row, $row_data) = @$row_entry;
        $modified_table->[$orig_row] = $row_data;
    }

    # 填充data_type
    for my $row_entry (@$type_data) {
        my ($orig_row, $row_data) = @$row_entry;
        $data_type->[$orig_row] = $row_data;
    }
}

# 输出结果
say Dumper $modified_table;
say Dumper $data_type;
say $max_length;

关键修改点 #

• 用$mce->chunk_begin()获取当前chunk在原数组中的起始行索引，计算每个元素对应的原行号，解决索引映射问题。
• 子进程内部先收集局部处理结果，再通过MCE->gather()传递给父进程，避免直接修改共享变量的问题。
• 每个子进程计算局部最大长度，父进程汇总后得到全局最大值，避免并发竞争导致的错误。

更优方案：使用MCE::Loop简化代码 #

如果追求更简洁的写法，可以使用MCE::Loop模块，它封装了gather等逻辑，代码更紧凑：

#!/usr/bin/env perl
use warnings;
use 5.14.0;
use Data::Dumper;
use Scalar::Util qw(looks_like_number);
use MCE::Loop;

sub get_string {
    my @chars = ( 'a'..'z', 'A'..'Z', '0'..'9', ' ' );
    return join '', map { @chars[rand @chars] } 1 .. rand(20) + 1;
}

my $table = [];
for my $row (0 .. 9) {
    $table->[$row] = [ map { get_string } 0 .. 5 ];
}

my ($modified_table, $data_type, $max_length) = ( [], [], 0 );

MCE::Loop::init {
    chunk_size => 5,
    max_workers => MCE::Util::get_ncpu(), # 使用CPU核心数作为工作进程数
};

my @results = mce_loop {
    my ($mce, $chunk_ref, $chunk_id) = @_;
    my $chunk_start = $mce->chunk_begin();
    my $local_max = 0;
    my @output;

    for my $row_idx (0 .. $#$chunk_ref) {
        my $orig_row = $chunk_start + $row_idx;
        my @mod_row;
        my @type_row;

        for my $col (0 .. $#{$chunk_ref->[$row_idx]}) {
            my $str = $chunk_ref->[$row_idx][$col] =~ s/\s/_/gr;
            push @mod_row, $str;
            push @type_row, looks_like_number($str) ? 1 : 0;
            $local_max = length($str) if length($str) > $local_max;
        }

        push @output, [ $orig_row, \@mod_row, \@type_row ];
    }

    return [ \@output, $local_max ];
} $table;

# 汇总结果
for my $res (@results) {
    my ($rows_data, $local_max) = @$res;
    $max_length = $local_max if $local_max > $max_length;

    for my $row_entry (@$rows_data) {
        my ($orig_row, $mod_row, $type_row) = @$row_entry;
        $modified_table->[$orig_row] = $mod_row;
        $data_type->[$orig_row] = $type_row;
    }
}

say Dumper $modified_table;
say Dumper $data_type;
say $max_length;

内容的提问来源于stack exchange，提问作者sid_com