multithreaded tcp listener with IO::Socket

[Random_Walk]

I am writing a little TCP/IP server to listen for heartbeats from remote machines and raise an alert if one is missed. To keep things simple rather than spawning child processes to handle connections I start a thread that comunicates through a Thread::Queue. Is this safe to do with IO:Socket ? My code follows with a small script to send the heartbeat at the end. Comments on the code also welcome

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

#get the port to bind to or default to 8000
my $port = $ARGV[0] || 8000;

# a hash to record client machines and thread queue for internal comun
+ication
my %clients;
my $queue = Thread::Queue -> new;
my $monitor = threads->create ("monitor", $queue);

#ignore child processes to prevent zombies
$SIG{CHLD} = 'IGNORE';

#create the listen socket
my $listen_socket = IO::Socket::INET->new(LocalPort => $port,
                                          Listen => 10,
                                          Proto => 'tcp',
                                          Reuse => 1);
#make sure we are bound to the port
die "Cant't create a listening socket: $@" unless $listen_socket;

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

#wait for connections at the accept call
while (my $connection = $listen_socket->accept) {
    # spawn a thread to handle the connection
    my $child = threads->create ("read_data", $queue, $connection);
}

sub read_data {
    # accept data from the socket and put it on the queue
    my ($queue, $socket) = @_;
    while (<$socket>) {
        print "listener got: $_";
        $queue -> enqueue(time." $_");
    }
}

sub monitor {
    my $queue = shift;
    while (1) {
        while ($queue -> pending) {
            my $data = $queue -> dequeue;
            print "monitor got: $data";
            $data =~ /(\d+) Heartbeat from (\S+) next one in (\d+) min
+utes/;
            my $time   = $1;
            my $client = $2;
            my $cycle  = $3;
            if (defined $clients{$client} and $clients{$client} -> [0]
+ eq 'NAK') {
                print "$client sent a beat again\n";
            }
            $clients{$client} = [ 'OK', $time + $cycle * 60 ];
        }
        for my $client (keys %clients) {
            next if $clients{$client}->[0] eq 'NAK';
            next if $clients{$client}->[1] > time;
            print "$client missed a heartbeat, expected at $clients{$c
+lient}->[1], now it is ".time."\n";
            $clients{$client}->[0] = 'NAK';
        }
        sleep 30;
    }
}

And here is a short script to send heartbeats

#!/usr/bin/perl
use strict;
use warnings;
use IO::Socket;

chomp(my $hostname  = `hostname`);
$hostname =~ s/\..*//;
my $me         = $ARGV[0] || $hostname;
my $cycle_time = $ARGV[1] || 5;
my $port       = $ARGV[2] || 8000;
my $ecg        = 'rtmr';

while (1) {
    print "sending heartbeat\n";
    my $socket = IO::Socket::INET->new(PeerAddr => $ecg,
                                PeerPort => $port,
                                Proto    => "tcp",
                                Type     => SOCK_STREAM)
        or die "Couldn't connect to $ecg:$port : $@\n";

    print "Heartbeat from $me next one in $cycle_time minutes\n";
    print $socket "Heartbeat from $me next one in $cycle_time minutes\
+n";
    close($socket);
    print "zzzzz......\n";
    sleep $cycle_time * 60;
}

Thanks in advance,
R.

Pereant, qui ante nos nostra dixerunt!

[BrowserUk]

As written, your server is going to leak memory. The rate of leakage will depend upon the number of clients you are monitoring.

• You are never closing the inbound sockets.

  This means you are never recycling filenos within your server, which will become a problem over time.

• You are never cleaning up (detach or join), the threads that you are spawning to handle the inbound connections.

  Although the threads will become dormant once the client disconnects, the threads memory will never be recycled which will become a problem over time.

• I'm not sure that this is doing anything for your application as it stands?

  #ignore child processes to prevent zombies
  $SIG{CHLD} = 'IGNORE';
  

  It will have no useful affect on a threaded process under Win32, but that may not be true for other OSs.

Both of these are fairly easy to correct.

Detach your client threads (I've ditched the my $client as you never do anything with it):

threads->create ("read_data", $queue, $connection)->detach;

and close the client socket once your done with it

sub read_data {
    # accept data from the socket and put it on the queue
    my ($queue, $socket) = @_;
    while (<$socket>) {
        print "listener got: $_";
        $queue -> enqueue(time." $_");
    }
    close $socket;
}

In my quick test with 100 clients, fileno 4 was re-used for all inbound connections which fixes that problem.

With these measures in place, the server process shows a miniscule growth. After 5 cycles of reconnections from 100 clients, it showed ~ 16k extra memory used. This may well be just the normal process of memory acquisition required by Perl's memory manager. This was true for both AS811(5.8.6) and AS817(5.8.8) on my system (XP).

[Random_Walk]

Very good points. The SIGCHLD was left over from my first child spawning implementation. The failure to detach my threads and close my socket are my own coded last thing on a Friday errors. All now fixed.

Thanks a lot,
R.

[BrowserUk]

In general, I currently favour using a thread pool over creating a new thread for each client. Especially when each connection is so breif. One advantage of this is that it eliminates the major possible source of memory leaks (the thread creation/cloning/destruction cycle).

With a duty cycle of a few milliseconds/300 seconds, unless all your clients are synchronised you would probably only need a pool of 2 or 3 threads/100 clients. Even if all your pool was busy when a new client connect occurs, with each communication being so breif, the new client will only have to wait a few milliseconds at most to be serviced.

Modifying your program to use a thread pool would be slightly more complex, but not grossly so. I will try to post something later today by way of demonstration.

[zentara]

I don't know how many connections you need to handle simultaneously, or how often connects/disconnects are made, but you might want to look at letting inetd handle the socket. See RFC: A Simple Socket Server Using 'inetd' and A Simple Socket Server Using 'inetd'

For a situation like "heartbeat monitors", it would seem prudent to keep each monitor connection a separate process, so if one goes haywire, it won't affect the others. If your parent thread goes down, for whatever reason, you will see all monitors fail. That will cause havoc for the nurses who are trying to determine which patient needs help.

[Random_Walk]

There will probably be less than 50 clients sending heartbeats between 1 and 5 minutes apart. It has to run on Windows as well as *nix so I thought to do it all in Perl. I have just hammered it with a multithreaded client and it takes a few secs to clear 100 beats. I have some retry now in the client. If it gets too much perhaps we will specify the server must be on *nix and use the inet way, should not be too onerous a requirement.

The heartbeat monitor itself will send its own heartbeat using IBM Tivoli events to a Tivoli Enterprise Console (T/EC) so its failure should be detected there. It will also send detected client heartbeat failures as T/EC events, records to a data warehouse and localy log them too.

Cheers,
R.

[izut]

Why don't you use mon for that? It's written in Perl, works and extensible.

[Random_Walk]

Looks interesting, is it GPL licence and will it run OK on windows too ?

Cheers,
R.

[perrin]

If you really want to keep things simple, you could just use HTTP, and run one of the pure Perl HTTP servers on CPAN.