Les threads Perl lisant les lignes d'entrée

Question

Je suis tellement confus avec les threads! J'espère que quelqu'un ici peut aider.

Ce qui suit est mon code:

$LOGFILE = "log.txt"; 
open(LOGFILE) or die("Could not open log file."); 
foreach $line (<LOGFILE>) { 
    chomp($line);  



my $number = shift; 


system("echo $number total lines > count"); 
    print "This thread is printing the number $number\n"; 
    sleep(1); 

log.txt:

Answer 1

use strict; 
use warnings qw(all); 
use threads; 

use Thread::Queue qw(); # 3.01+ 

use constant NUM_WORKERS => 10; 

sub worker { 
    my ($job) = @_; 
    print("This is log $job\n"); 
} 

{ 
    my $log_qfn = "log.txt"; 
    open(my $log_fh, '<', $log_qfn) 
     or die("Can't open \"$log_qfn\": $!\n"); 

    my $q = Thread::Queue->new(); 
    for (1..NUM_WORKERS) { 
     async { 
     while (defined(my $job = $q->dequeue())) { 
      worker($job); 
     } 
     }; 
    } 

    while (my $line = <$log_fh>) { 
     chomp($line); 
     $q->enqueue($line); 
    } 

    $q->end(); 

    $_->join() for threads->list(); 
} 

Bien sûr, la sortie ne sera pas nécessairement dans le même ordre. Si vous en avez besoin, il faudra beaucoup de travail supplémentaire.

Answer 2

Tout d'abord, il y a un tas de problèmes avec votre code avant même que nous n'ayons à enfiler. Parce que le code parallèle introduit un monde entier de nouveaux bogues, il est vraiment important d'avoir un code décent pour commencer.

La vôtre ne compile pas, ou ne fonctionne pas avec un seul thread.

• LOGFILE et $LOGFILE ne sont pas la même chose.
• ne pas utiliser foreach lire un descripteur de fichier - utiliser while.
• Vous ne passent pas réellement de données de fichiers partout
• Vous maintenir un tableau @threads que vous n'utilisez pas - la plupart du temps ce n'est pas nécessaire, parce que vous avez threads->list.
• Activer use strict et use warnings; - Les threads sont en désordre et créent de nouvelles classes de bogues et de défauts intermittents. Votre code doit être à l'épreuve des balles avant d'introduire la complexité supplémentaire.

Il y a un point fondamental qui vous manque ici - les threads sont non déterministes. Le point entier est - vous n'avez plus une séquence d'exécution définie, et donc ne peut pas s'appuyer sur n'importe quel ordre de sortie particulier. De plus - le filetage est officiellement déconseillé parce que il est mal compris - les threads de perl ne sont pas légers, donc beaucoup d'entre eux sont horriblement inefficaces. Si vous voulez aller dans cette direction, alors Parallel::ForkManager est probablement l'outil pour le travail.

Et même en laissant de côté - ils parallélisent les ressources de calcul, pas les E/S du disque. Si vous "lisez" simplement un fichier, vous ne le ferez pas plus vite que le contrôleur de disque, ce qui est votre facteur limitant.

Quoi qu'il en soit, pour faire votre chose:

#!/usr/bin/env perl 
use strict; 
use warnings; 
use threads; 
use Thread::Queue qw (); 

my $num_threads = 10; 

my $work_q = Thread::Queue -> new; 
my $output_q = Thread::Queue -> new; 

sub worker { 
    my $count = 0; 
    print "Worker: ", threads -> self -> tid, " started and waiting for input\n"; 
    while (my $item = $work_q -> dequeue) { 
     $output_q -> enqueue (threads -> self -> tid. ": ". $count++. " items: $item"); 
     print "Processing: $item"; 
    } 
    print "Worker: ", threads -> self -> tid, " exiting\n"; 
} 

sub serialise_output { 

    print "Serialiser waiting for things to print: ", threads -> self -> tid,"\n"; 
    while (my $output_item = $output_q -> dequeue) { 
    print $output_item,"\n"; 
    } 
} 

my @workers = map { threads -> create (\&worker) } 1..$num_threads; 
my $serialiser = threads -> create (\&serialise_output); 


open (my $input, '<', 'log.txt') or die $!; 
while (my $line = <$input>) { 
    chomp ($line) ; 
    $work_q -> enqueue ($line); 
} 
close ($input); 

$work_q -> end; #tell the threads that's all of it, so they exit. 
$_ -> join for @workers; #wait for all the workers to exit. 

$output_q -> end; #need to wait for workers to finish, so we don't create race condition. 
$serialiser -> join; 


print "All threads done\n"; 

Note - le sérialiseur obtient est entrée dans un ordre assez aléatoire, sur la base duquel fil traité quelle ligne. Vous pouvez soit diffuser directement ce fichier dans un descripteur de fichier ouvert (n'essayez pas d'ouvrir le même fichier dans plusieurs threads - cela deviendra moche) ou faire une sorte de tri et de coalescence avant d'imprimer.

Vous ne pouvez absolument pas compter sur une commande de votre fichier io, parce que vous divisez le travail - c'est plutôt le point. Mais vous n'avez pas besoin d'un thread séparé pour sérialiser la sortie si vous attendez que vous l'ayez tous (par exemple si vous allez le trier) - vous pouvez simplement le faire dans le thread 'principal' à la place.