Tác vụ không đồng bộ của Perl cho mã "bất kỳ" nào, cho dù đó là gì?

Question

Tôi đã viết một hệ thống "kiểm tra" thực hiện các "kiểm tra" khác nhau trên các dịch vụ, hệ thống, cơ sở dữ liệu, tệp, v.v ... "Kiểm tra" là bản chất chung và có thể là bất kỳ thứ gì. Tất cả các kiểm tra được báo cáo trong một thời tiết định dạng phổ biến mà họ vượt qua hoặc thất bại, bất cứ điều gì có thể được.

Nó được viết theo kiểu OO mô-đun để các nhà phát triển có thể chỉ cần làm theo khuôn khổ và viết séc độc lập với một và khác. Mỗi đối tượng có chứa một đối tượng báo cáo được chia sẻ mà sau khi họ chạy một kiểm tra họ chỉ đơn giản là $ self -> {'reporting'} -> report (params). Các tham số được xác định và các nhà phát triển được giả định báo cáo một cách thích hợp. Đối tượng báo cáo sau đó lập chỉ mục các báo cáo này. kịch bản bộ nạp chính của tôi có mục như sau:

my $reportingObject = new Checks::Reporting(params); 
my @checks; 

push @checks, new Checks::Check_One($reportingObject, params)); 
push @checks, new Checks::Check_One($reportingObject, params)); 
. 
. 
push @checks, new Checks::Check_N($reportingObject, params)); 

Để khởi động kiểm tra và hoàn thiện báo cáo khi chúng được thực hiện tôi đã và đang làm:

foreach my $check (@checks) { 
    $check->run_stuff(); 
} 

$reportingObject->finalize_report(); 

Bây giờ kể từ khi những kiểm tra là hoàn toàn độc lập (đừng lo lắng về đối tượng báo cáo) chúng có thể chạy song song. Như một cải tiến tôi đã thực hiện:

my @threads; 
foreach my $check (@checks) { 
    push @threads, async { $check->run_stuff(); } 
} 

foreach my $thread (@threads) { 
    $thread->join; 
} 

#All threads are complete, and thus all checks are done 
$reportingObject->finalize_report(); 

Như tôi đã nói trước đó các nhà phát triển sẽ viết Kiểm tra độc lập với nhau. Một số kiểm tra là đơn giản và những người khác thì không. Các kiểm tra đơn giản có thể không có mã không đồng bộ trong đó, nhưng những người khác có thể cần phải chạy không đồng bộ trong nội bộ như

sub do_check { 
    my @threads; 
    my @list = @{$self->{'list'}}; 

    foreach my $item (@list) { 
     push @threads, async { 
        #do_work_on_$item 
        #return 1 or 0 for success or fail 
       }; 
     foreach my $thread (@threads) { 
      my $res = $thread->join; 
      if($res == 1) { 
       $self->{'reporting'}->report(params_here); 
      } 
     } 
    } 
} 

Như bạn có thể thấy mô hình luồng cho phép tôi làm những việc trong điều kiện rất mơ hồ. Mỗi "Kiểm tra" không có vấn đề gì nó chạy độc lập trong chủ đề riêng của mình. Nếu một nhà phát triển cá nhân có các công cụ không đồng bộ để làm, bất kể nó là gì, anh ta chỉ đơn giản thực hiện nó một cách độc lập trong chủ đề riêng của nó. Tôi muốn một mô hình tương tự như thế này.

Thật không may, các chủ đề chậm và không hiệu quả. Tất cả các thư viện async đều có các trình theo dõi cụ thể như IO, v.v. Tôi không muốn bất cứ điều gì cụ thể. Tôi muốn một mô hình dựa trên sự kiện cho phép tôi chỉ đơn giản là khởi động các nhiệm vụ không đồng bộ, bất kể chúng là gì và chỉ cần thông báo khi chúng được thực hiện để tôi có thể tiếp tục.

Hy vọng điều đó giải thích và bạn có thể chỉ cho tôi đi đúng hướng.

Answer 1

Điều này có vẻ như một sự phù hợp tốt cho một mô hình sếp-nhân viên:

• đẻ trứng một vài công nhân ở phần đầu của chương trình. Hãy chắc chắn rằng tất cả họ đều có quyền truy cập vào hàng đợi.

• Làm bao nhiêu lần kiểm tra tùy thích. Các công nhân dequeue kiểm tra, thực hiện chúng, và enqueue kết quả trong một hàng đợi đầu ra.

• Chuỗi chính của bạn xem xét kết quả từ chuỗi đầu ra và thực hiện bất kỳ thứ gì nó muốn.

• Tham gia vào lao động trong một END khối

Bạn có thể muốn nhìn vào Thread::Queue::Any nếu có một cơ hội mà bạn muốn đưa coderefs vào hàng đợi.

Dưới đây là một ví dụ hoàn toàn Runnable:

use strict; use feature 'say'; 
use threads; use threads::shared; use Thread::Queue::Any; 
use constant NUM_THREADS => 5; 
local $Storable::Deparse = 1; local $Storable::Eval = 1; # needed to serialize code 

my $check_q = Thread::Queue::Any->new; 
my $result_q = Thread::Queue::Any->new; 

# start the workers 
{ 
    my $running :shared = NUM_THREADS; 
    my @threads = map threads->new(\&worker, $check_q, $result_q, \$running), 1..NUM_THREADS; 

    END { $_->join for @threads } 
} 

# enqueue the checks 
$check_q->enqueue($_) for sub {1}, sub{2}, sub{"hi"}, sub{ die }; 
$check_q->enqueue(undef) for 1..NUM_THREADS; # end the queue 

while(defined(my $result = $result_q->dequeue)) { 
    report($$result); 
} 

sub report { 
    say shift // "FAILED"; 
} 

sub worker { 
    my ($in, $out, $running_ref) = @_; 
    while (defined(my $check = $in->dequeue)) { 
    my $result = eval { $check->() }; 
    $out->enqueue(\$result); 
    } 

    # last thread closes the door 
    lock $$running_ref; 
    --$$running_ref || $out->enqueue(undef); 
} 

này in

1 
2 
hi 
FAILED 

theo một thứ tự ngẫu nhiên một chút.