multithreading - Threads.数组中的最大值

Question

实际上，比方说，启动了 2 个线程。而一个包含5个元素的数组，所有的计算都是进行1次交易。 2 出现并因缺乏工作而死亡。或者，如果您指定 9999 个元素，那么他会在最后 30-50 个项目中醒来。在我看来，他并不是真正的多线程。

#! usr/bin/perl -w
use strict;
use  warnings;
use  5.010;
use threads;
use threads::shared;
use Data::Dumper;

my $n=0;

my $max:shared=0;
my @threads;
my $shr:shared=0;
say "Enter the number of threads";
my $threads=<stdin>;
my @pack;

#Заполняем массив
say "Enter the number of array elements";
my $c=<stdin>;
for (my $i=0;$i<$c;$i++){
    my $r= int rand 999;
    @pack=(@pack,$r);
                        }
                         say  @pack;

# Создаём нужное количество потоков
for my $t (1..$threads) {
    push @threads,threads->create(\&find,$t);
                        }
# Дожидаемся окончания работы всех потоков
for my  $t (@threads){ $t->join();}

sub find {
    # Номер текущего потока
    my $num= shift;
    say "+Thread $num started";

    while($n<=$#pack){
        # Распределяем элементы по потокам
        lock($shr);#lock ($max);
        my $stg=1+$shr++;

        # Если элементы закончились, прерываем работу
        if ($stg>=$#pack+2) {
            say "-Thread $num done. \n";
            return;}


        if($max<$pack[$n]){$max=$pack[$n];}
        say "Thread: $num \tStage: $stg  \tMax: $max";
        $n++;
        sleep 1+int rand (3);       
             }
        }

Answer 1

线程的全部工作代码都在一个 block 中，在这个 block 中获得了lock($shr)，防止多个线程做任何工作。

---

听起来您有一个大数组，并且您希望工作线程处理该大数组的部分内容。

线程唯一需要同步的时间是抓取工作和更新结果时。

请注意，您需要在访问数组时锁定数组，但在工作时无法保持锁定状态，因此您需要将工作单元复制到线程局部变量中。

下面求一个数组所有元素的总和。

#!/usr/bin/perl
use strict;
use warnings;
use feature qw( say );

use threads;
use threads::shared;

use List::Util qw( sum );

# This is where the real work is done.
# Notice how it there's no thread-related code in it
# and how it doesn't use any shared variables.
sub work {
    sleep(1+rand(3));
    return sum(@_);
}

{
    @ARGV == 3
        or die("usage\n");

    my ($data_size, $num_threads, $chunk_size) = @ARGV;

    my $next :shared = 0;
    my @data :shared;
    my $sum  :shared;

    for (1..$data_size) {
        push @data, int(rand(9999));
    }

    say "Without threads: ", sum @data;

    for my $t (1..$num_threads) {
        async {
            say sprintf("[%s]: Worker started.", threads->tid);
            LOOP: while (1) {
                # Grab work.
                my @local_data;
                {
                    lock $next;
                    last LOOP if $next >= @data;

                    my $new_next = $next + $chunk_size;
                    $new_next = @data if $new_next > @data;

                    say sprintf("[%s] Processing %s..%s", threads->tid, $next, $new_next-1);
                    @local_data = @data[$next .. $new_next-1];
                    $next = $new_next;
                }

                my $local_sum = work(@local_data);    

                # Update results.
                { lock $sum; $sum += $local_sum; }
            }
            say sprintf("[%s] done.", threads->tid)
        };
    }

    $_->join for threads->list;

    say "With threads: $sum";
}

输出:

$ ./a 125 3 20
Without threads: 577556
[1]: Worker started.
[1] Processing 0..19
[2]: Worker started.
[2] Processing 20..39
[3]: Worker started.
[3] Processing 40..59
[2] Processing 60..79
[1] Processing 80..99
[3] Processing 100..119
[1] Processing 120..124
[2] done.
[1] done.
[3] done.
With threads: 577556