Attività asincrone Perl per "qualsiasi" codice, non importa quale sia?

Question

Ho scritto un sistema di "correttore" che esegue vari "controlli" su vari servizi, sistemi, database, file, ecc. Un "controllo" è di natura generica e può essere qualsiasi cosa. Tutti i controlli sono riportati in un formato comune in cui passano o falliscono, qualunque cosa sia.

È scritto in modo modulare OO in modo che gli sviluppatori possano semplicemente seguire il framework e scrivere controlli indipendentemente dall'uno e dall'altra. Ogni oggetto contiene un oggetto di segnalazione condiviso che dopo aver eseguito un controllo ha semplicemente $ self -> {'reporting'} -> report (params). I param sono definiti e si presume che gli sviluppatori riferiscano in modo appropriato. L'oggetto report quindi indicizza questi report. Il mio script loader principale ha voci come il seguente:

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

Per dare il via ai controlli e finalizzare il rapporto una volta finito ho fatto:

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

$reportingObject->finalize_report(); 

Ora, poiché questi controlli sono totalmente indipendenti (non preoccuparti per l'oggetto report) possono essere eseguiti in parallelo. Come miglioramento ho fatto:

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

Come ho detto prima gli sviluppatori scriveranno i controlli indipendentemente l'uno dall'altro. Alcuni controlli sono semplici e altri no. I semplici controlli non possono avere codice asincrono in loro, ma altri potrebbero essere necessario eseguire in modo asincrono internamente come

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

Come si può vedere il modello di threading mi permette di fare le cose in termini molto vaghi. Ogni "Verifica" indipendentemente da ciò che viene eseguito in modo indipendente nel proprio thread. Se un singolo sviluppatore ha cose asincrone da fare, non importa quale sia, lo fa semplicemente in modo indipendente nella sua stessa discussione. Voglio un modello simile a questo.

Purtroppo i thread sono lenti e inefficienti. Tutte le librerie asincrone hanno osservatori specifici come IO, ecc. Non voglio nulla di specifico. Mi piacerebbe un modello basato su eventi che mi consenta di avviare semplicemente compiti asincroni, indipendentemente da cosa siano, e semplicemente di notificare quando sono tutti fatti, così posso andare avanti.

Speriamo che questo lo spieghi e che tu possa indirizzarmi nella giusta direzione.

Answer 1

Questo mi sembra una buona misura per un modello di capo-operaio:

• Spawn alcuni operai, all'inizio del programma. Assicurati che tutti abbiano accesso a una coda.

• Accoda quanti più controlli desideri. I lavoratori dequistano i controlli, li eseguono e accodano il risultato in una coda di output.

• Il thread principale esamina i risultati del thread di output e fa tutto ciò che desidera.

• registrazione dei lavoratori in un blocco END

Probabilmente si desidera guardare Thread::Queue::Any se c'è la possibilità che si desidera mettere coderefs nella coda.

Ecco un esempio pienamente eseguibile:

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

Questo stampa

1 
2 
hi 
FAILED 

in un ordine leggermente casuale.