Perl async Aufgaben für "Any" -Code, egal was es ist?

Question

Ich habe ein "Checker" -System geschrieben, das verschiedene "Checks" für verschiedene Dienste, Systeme, Datenbanken, Dateien usw. durchführt. Ein "Check" ist generischer Natur und kann alles Mögliche sein. Alle Überprüfungen werden in einem gemeinsamen Format gemeldet, unabhängig davon, ob sie erfolgreich sind oder nicht.

Es ist in einer modularen OO-Mode geschrieben, so dass Entwickler das Framework einfach verfolgen und unabhängig voneinander Prüfungen schreiben können. Jedes Objekt enthält ein gemeinsames Berichtsobjekt, das nach der Überprüfung einfach $ self -> {'reporting'} -> report (params) anzeigt. Die Parameter sind definiert und es wird angenommen, dass die Entwickler angemessen berichten. Das Berichtsobjekt indiziert diese Berichte dann. Mein Haupt loader Skript enthält Einträge wie die folgenden:

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)); 

Um die Kontrollen zu beginnen und den Bericht abzuschließen, sobald sie fertig sind, ich habe zu tun:

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

$reportingObject->finalize_report(); 

Nun, da diese Kontrollen sind völlig unabhängig (Mach dir keine Sorgen über das Berichtsobjekt) sie können parallel ausgeführt werden. Als eine Verbesserung habe ich getan:

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(); 

Wie ich schon sagte, die Entwickler schreiben Checks unabhängig voneinander. Einige Überprüfungen sind einfach und andere nicht. Die einfachen Überprüfungen können nicht asynchronen Code in ihnen, aber andere könnten asynchron intern wie

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); 
      } 
     } 
    } 
} 

ausführen müssen Wie Sie das Threading-Modell sehen können mir Dinge in sehr vage zu tun erlaubt. Jedes "Check", egal was es ist, läuft unabhängig in seinem eigenen Thread. Wenn ein einzelner Entwickler asynchrone Aufgaben zu erledigen hat, egal was es ist, macht er es einfach unabhängig in seinem eigenen Thread. Ich möchte ein ähnliches Modell.

Leider sind die Threads langsam und ineffizient. Alle Async-Bibliotheken haben spezielle Beobachter wie IO usw. Ich möchte nichts Spezifisches. Ich hätte gerne ein ereignisbasiertes Modell, das es mir ermöglicht, einfach asynchrone Aufgaben zu starten, ganz gleich, was sie sind, und einfach zu benachrichtigen, wenn alles erledigt ist, damit ich weitermachen kann.

Hoffentlich erklärt das und Sie können mir in die richtige Richtung zeigen.

Answer 1

Dies scheint eine gute Passform für ein Chef-Arbeiter Modell:

• Spawn ein paar Arbeiter am Anfang des Programms. Stellen Sie sicher, dass alle Benutzer Zugriff auf eine Warteschlange haben.

• Fügen Sie so viele Überprüfungen ein, wie Sie möchten. Die Worker entziehen die Prüfungen, führen sie aus und fügen das Ergebnis in eine Ausgabewarteschlange ein.

• Ihr Haupt-Thread betrachtet die Ergebnisse aus dem Ausgabe-Thread und macht, was immer es will.

• Mitglied werden die Arbeiter in einem END Block

Sie wahrscheinlich bei Thread::Queue::Any aussehen soll, wenn es eine Chance, dass Sie wollen, ist coderefs in die Warteschlange einzureihen.

Hier ist ein vollständig lauffähiges Beispiel:

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); 
} 

Dieser druckt

1 
2 
hi 
FAILED 

in leicht zufälliger Reihenfolge.