Perl多线程脚本重组二进制文件块MD5验证失败问题求助

• 需求：将二进制文件分块用于分段上传，通过Perl多线程加速传输
• 异常现象：
  • 单线程模式：小文本文件、大文件处理后MD5校验均正常
  • 多线程模式：小文本文件处理正常，但9GB的tar.gz文件（块大小设为500MB）重组后比原文件小约1GB，MD5校验不匹配

单线程正常实现代码 #

use strict;
use POSIX;

my $file = 'test.txt';
my $chunkSize = 3;
my $size = -s $file;
my $parts = ceil($size / $chunkSize);

my %output;
open(my $fileHandle, '<', $file) or die("Error reading file, stopped");
binmode($fileHandle);

for (my $i = 0; $i < $parts; $i++) {
    my $chunk;
    my $offset = $i * $chunkSize;

    seek ($fileHandle, $offset, 0);

    if ($chunkSize > $size) {
        $chunkSize = $size;
    }
    elsif (($chunkSize + $offset) > $size) {
        $chunkSize = $size - $offset;
    }

    my $length = read($fileHandle, $chunk, $chunkSize); 
    say STDERR "Chunk size for $i: " . $chunkSize . "(" . ($chunkSize + $offset) . "/$size) starting at " . $offset .": ";# . $chunk;

    $output{ $i } = $chunk;
}

close($fileHandle); 

open(my $fileHandle, '>', $file . '.comp') or die("Error reading file, stopped");
binmode($fileHandle);

for (1...$parts) {
    print $fileHandle $output{ $_ };
}

close ($fileHandle);

存在问题的多线程实现代码 #

use strict;
use POSIX;
use threads;
use threads::shared;
use Thread::Queue;

my $file = '/backup/2022-12-13/accounts/test.tar.gz';

my $chunkSize = 500 * 1024 * 1024;
#Or my $chunkSize = 3; when using it with the smaller text test file.
my $size = -s $file;
my $parts = ceil($size / $chunkSize);

my %output :shared;

my @queueData;
for (0...$parts) { push (@queueData, $_); }

my $overallProgress :shared;
my $queue = Thread::Queue->new(@queueData);

open(my $fileHandle, '<', $file) or die("Error reading file, stopped");
binmode($fileHandle);

my @threads;
for(1..4) {
    push @threads, threads->create( sub {

        while( my $i = $queue->dequeue_nb ) {
            my $chunk;
            my $offset = ($i - 1) * $chunkSize;

            seek ($fileHandle, $offset, 0);

            if ($chunkSize > $size) {
                $chunkSize = $size;
            }
            elsif (($chunkSize + $offset) > $size) {
                $chunkSize = $size - $offset;
            }

            my $length = read($fileHandle, $chunk, $chunkSize); 
            say STDERR "Chunk size for $i: " . $chunkSize . "(" . ($chunkSize + $offset) . "/$size) starting at " . $offset .": ";

            $output{ $i } = $chunk;
        }
    });
}

$_->join for @threads;
close($fileHandle); 

open(my $fileHandle, '>', $file . '.comp') or die("Error reading file, stopped");
binmode($fileHandle);

for (1...$parts) {
    print "Reassembling part $_...";
    print $fileHandle $output{ $_ };
}

close ($fileHandle);

问题根源 #

1. 共享文件句柄的线程竞争：多个线程共用同一个全局文件句柄，seek和read操作非原子性，线程间会互相覆盖文件指针位置，导致部分线程读取到错误偏移的数据，甚至读取失败。
2. 索引逻辑不一致：队列中传入0到$parts的索引，但线程内计算偏移用($i-1)*$chunkSize，重组时循环又从1到$parts，直接丢失了索引0对应的块。
3. 全局变量被意外修改：线程内修改了全局的$chunkSize，后续线程的块大小计算会基于被修改后的值，导致读取的块长度错误，丢失大量数据。

修正后的多线程实现 #

use strict;
use POSIX;
use threads;
use threads::shared;
use Thread::Queue;

my $file = '/backup/2022-12-13/accounts/test.tar.gz';
my $chunkSize = 500 * 1024 * 1024;
my $size = -s $file;
my $parts = ceil($size / $chunkSize);

my %output :shared;

# 队列传入正确的块索引（0到$parts-1，与单线程逻辑对齐）
my $queue = Thread::Queue->new(0..$parts-1);

my @threads;
for(1..4) {
    push @threads, threads->create( sub {
        # 每个线程独立打开文件句柄，避免共享句柄的竞争问题
        open(my $fh, '<', $file) or die("Thread error reading file: $!");
        binmode($fh);

        while( my $i = $queue->dequeue_nb ) {
            my $chunk;
            my $offset = $i * $chunkSize;
            # 使用局部变量计算当前块大小，不修改全局变量
            my $current_chunk_size = $chunkSize;
            
            if ($current_chunk_size > $size) {
                $current_chunk_size = $size;
            }
            elsif (($current_chunk_size + $offset) > $size) {
                $current_chunk_size = $size - $offset;
            }

            # 线程内独立执行seek和read，无竞争
            seek($fh, $offset, 0) or die("Thread seek error: $!");
            my $length = read($fh, $chunk, $current_chunk_size);
            # 验证读取字节数是否符合预期，及时发现错误
            die("Read only $length bytes for chunk $i (expected $current_chunk_size)") unless $length == $current_chunk_size;

            say STDERR "Chunk $i: $current_chunk_size bytes (offset $offset, total $size)";
            $output{ $i } = $chunk;
        }
        close($fh);
    });
}

$_->join for @threads;

# 按0到$parts-1的顺序重组文件，确保所有块都被写入
open(my $out_fh, '>', $file . '.comp') or die("Error writing output file: $!");
binmode($out_fh);

for my $i (0..$parts-1) {
    print "Reassembling part $i...\n";
    print $out_fh $output{ $i } or die("Write error for chunk $i: $!");
}

close($out_fh);

关键修复点 #

• 线程独立打开文件句柄：每个线程管理自己的文件指针，彻底避免seek/read的竞争问题
• 使用局部变量计算块大小：不修改全局变量，确保每个线程的块大小计算不受其他线程影响
• 统一索引逻辑：队列和重组循环都使用从0开始的索引，与单线程逻辑对齐，避免块丢失
• 增加读取验证：检查实际读取字节数是否等于预期，及时发现读取异常

内容的提问来源于stack exchange，提问作者Timothy R. Butler