Comment enregistrer des sockets dans un hachage et faire une boucle sur un autre thread?

Question

Je travaille sur un serveur TCP mulithreaded. Dans le thread principal, j'écoute sur un socket et crée un nouveau thread pour les nouvelles connexions entrantes. Je veux enregistrer toutes les connexions entrantes dans un hachage afin que je puisse y accéder à partir d'un autre thread. À partir du fil du moniteur, je ne peux pas lire les nouvelles connexions ajoutées. Il semble qu'un hash de nouveaux clients est créé lors de la création du thread de surveillance.

Comment conserver la liste de toutes les sockets et les boucler à partir de mon thread de moniteur?

Code actuel:

#!/usr/bin/perl 
use strict; 
use IO::Socket; 
use threads; 
use Thread::Queue; 

# init 
my $clients = {}; 
my $queue = Thread::Queue->new; 

# thread that monitors 
threads->create("monitor"); 

# create the listen socket 
my $listenSocket = IO::Socket::INET->new(LocalPort => 12345, 
             Listen => 10, 
             Proto => 'tcp', 
             Reuse => 1); 

# make sure we are bound to the port 
die "Cant't create a listening socket: $@" unless $listenSocket; 

print "Server ready. Waiting for connections on 34567 ... \n"; 

# wait for connections at the accept call 
while (my $connection = $listenSocket->accept) { 
    # set client socket to non blocking 
    my $nonblocking = 1; 
    ioctl($connection, 0x8004667e, \\$nonblocking); 

    # autoflush 
    $connection->autoflush(1); 

    # debug 
    print "Accepted new connection\n"; 

    # add to list 
    $clients->{time()} = $connection; 

    # start new thread and listen on the socket 
    threads->create("readData", $connection); 
} 

sub readData { 
    # socket parameter 
    my ($client) = @_; 

    # read client 
    while (<$client>) { 
     # remove newline 
     chomp $_; 

    # add to queue 
     $queue->enqueue($_); 
    } 

    close $client; 
} 

sub monitor { 
    # endless loop 
    while (1) { 

     # loop while there is something in the queue 
     while ($queue->pending) { 

      # get data from a queue 
      my $data = $queue->dequeue; 

      # loop all sockets 
      while (my ($key, $value) = each(%$clients)) { 

       # send to socket 
       print $value "$data\n"; 

      } 
     } 

     # wait 0,25 seconds 
     select(undef, undef, undef, 0.25); 
    } 
} 

close $listenSocket; 

Answer 1

Vous devez partager $clients via share de threads::shared:

my $clients = &share({}); 

La syntaxe ancienne est due à un problème documenté avec les prototypes de Perl. Si vous avez at least Perl 5.8.9, utilisez le plus joli

my $clients = shared_clone({}); 

à la place.

Vous souhaitez également protéger $clients avec un verrou, par exemple.,

my $clients_lock : shared; 
{ 
    lock $clients_lock; 
    $clients->{time()} = fileno $connection; 
} 

Enfin, parce que IO::Socket::INET instances sont typeglobs Perl, vous ne pouvez pas les partager, donc au lieu d'ajouter leurs descripteurs de socket (de fileno) à $clients puis fdopen la prise si nécessaire avec

open my $fh, ">&=", $sockdesc or warn ... 

Le programme ci-dessous répète les données entrantes pour les autres prises connectées:

#!/usr/bin/perl 

use strict; 
use IO::Socket; 
use threads; 
use threads::shared; 
use Thread::Queue; 

# init 
my $clients = &share({}); 
my $clients_lock : shared; 

my $queue = Thread::Queue->new; 

# thread that monitors 
threads->create("monitor"); 

# create the listen socket 
my $port = 12345; 
my $listenSocket = IO::Socket::INET->new(
    LocalPort => $port, 
    Listen  => 10, 
    Proto  => 'tcp', 
    Reuse  => 1 
); 

# make sure we are bound to the port 
die "Can't create a listening socket: $@" unless $listenSocket; 

print "Server ready. Waiting for connections on $port ... \n"; 

# wait for connections at the accept call 
while (my $connection = $listenSocket->accept) { 
    # set client socket to non blocking 
    my $nonblocking = 1; 
    ioctl($connection, 0x8004667e, \\$nonblocking); 

    # autoflush 
    $connection->autoflush(1); 

    # debug 
    print "Accepted new connection\n"; 

    # add to list 
    { 
    lock $clients_lock; 
    $clients->{time()} = fileno $connection; 
    } 

    # start new thread and listen on the socket 
    threads->create("readData", $connection); 
} 

sub readData { 
    # socket parameter 
    my ($client) = @_; 

    # read client 
    while (<$client>) { 
    chomp; 
    $queue->enqueue($_); 
    } 

    close $client; 
} 

sub monitor { 
    # endless loop 
    while (1) { 
    # loop while there is something in the queue 
    while ($queue->pending) { 
     # get data from a queue 
     my $data = $queue->dequeue; 

     # loop all sockets 
     { 
     lock $clients_lock; 
     while (my ($key, $value) = each(%$clients)) { 
      # send to socket 
      if (open my $fh, ">&=", $value) { 
      print $fh "$data\n"; 
      } 
      else { 
      warn "$0: fdopen $value: $!"; 
      } 
     } 
     } 
    } 

    # wait 0,25 seconds 
    select(undef, undef, undef, 0.25); 
    } 
} 

close $listenSocket; 

Answer 2

Ne pas avoir trop d'expérience en utilisant des threads en Perl, mais je pense que vous voulez simplement partager votre liste de clients:

 
    use threads::shared ; 
    my $clients : shared = {}; 

---

Mise à jour:

Perl se plaint:

my $hash : shared = {}; 

mais il semble être ok avec:

my $hash = {}; 
share($hash); 

De plus, ce code:

my $hash = { key1 => "value1" }; 
share($hash); 

semble effacer la table de hachage, mais

my $hash = {}; 
share($hash); 
$hash->{key1} = "value1"; 

fonctionne comme j'attendrais.