I am working on a mulithreaded TCP server. In the main thread, I listen on a socket and create a new thread for new incoming connections. I want to save all incoming connections in a hash so that I can access them from yet another thread.
From the monitor thread, I can not read any newly added connections. It seems a new clients hash is created when creating the monitor thread.
How do i keep list of all sockets and loop them from my monitor thread?
Current code:
#!/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;
You need to share $clients
via share
from threads::shared
:
my $clients = &share({});
The old-fashioned syntax is due to a documented issue with Perl’s prototypes. If you have at least Perl 5.8.9, use the nicer
my $clients = shared_clone({});
instead.
You also want to protect $clients
with a lock, e.g.,
my $clients_lock : shared;
{
lock $clients_lock;
$clients->{time()} = fileno $connection;
}
Finally, because IO::Socket::INET
instances are Perl typeglobs, you can’t share them, so instead add their socket descriptors (from fileno
) to $clients
and then fdopen
the socket when necessary with
open my $fh, ">&=", $sockdesc or warn ...
The program below repeats inbound data to the other connected sockets:
#!/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;
Don't have too much experience using threads in Perl, but I think you just want to share your client list:
use threads::shared; my $clients : shared = {};
Update:
Perl complains about:
my $hash : shared = {};
but it seems to be ok with:
my $hash = {};
share($hash);
Also, this code:
my $hash = { key1 => "value1" };
share($hash);
seems to clear the hashtable, but
my $hash = {};
share($hash);
$hash->{key1} = "value1";
works like I'd expect.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With