in reply to Re^14: PANIC: underlying join failed threded tcp server
in thread PANIC: underlying join failed threded tcp server

HEY! Got the server working by putting some of Thread::Queue into my code, and now it seems to be running like a champ. If you'd like to see the code let me know, but I'm guessing you have a fair idea of how its working. Ran 200k commands in 2 hours without a hiccup.

Cool. (Yes. I would really like to see the code. You have my email id.)

And that is really interesting that threads got messed up. I'll definitely try downgrading perl and testing it out but as it is no longer a time critical thing, and more a matter of interest, I probably wont get to it until next week.

I took your latest code -- stripped out the defensive stuff out of rxd.pl:

#    if( threads::list( threads::running ) >= 100 ) {
#        sendResponse("Session rejected. Too many sessions currently r
+unning.\n", 101, $client); #just decided 101 should be the retry retu
+rn code, no reason really
#        # tprint( "Session request by " . $client->sockhost() . " rej
+ected" ); #disabled because it was flodding the log
#        $client->shutdown(2);
#        $client->close;
#    }else

[download]

Tweaked the client (rx,pl) to run many commands in loop and do so from multiple threads:

our $T //= 1000;
our $R //= 1000;

my @threads;
for my $t ( 0 .. $T ){
    push @threads, async {
        for my $r ( 1 .. $R ) {
            print "Client:$t sending command: $r\n";
            my $server;
            my $rc;
            my $response;


            while(1){
                $server = IO::Socket::INET->new(
                                    Proto    => "tcp",
                                    PeerAddr => $address,
                                    PeerPort => $port,
                                );
                unless (defined($server)){print "Unable to connect to 
+RXD; [$! / $^E\n";}
                $server->autoflush(1);

                sendCommand( $server );
                ($response, $rc) = receiveResponse( $server );
                if($rc == 101){
                    shutdown($server, 2);
                    close $server;
                    my $random = 1 + rand(10);
#                    select(undef, undef, undef, $random);
                    next;
                }
                else{last;}
            }

            debug("Closing connection");
            shutdown($server, 2);
            close $server;
        };
    };
}

$_->join for @threads;

[download]

With 5.16.1, this falls in a heap with 1 thread after just a few seconds. The same "invalid handle" stuff as you've been seeing.

With 5.10.1, I ran the above with those numbers -- 1000 concurrent clients, each running 1000 commands (the simple dir /s) and it ran to completion (after 1,000,000 commands served :), without errors in a little over 1 1/2 hours. And that's with 3 cores running the server and 1 core running the clients. Peak memory usage for the server was around 1.5 GB.

Again, I really cannot stress enough how thankful I am for all your help debugging. Could not have gotten this far without you.

YW. I learn almost as much from each of these real-world uses as the people I'm helping do.

With the rise and rise of 'Social' network sites: 'Computers are making people easier to use everyday'
Examine what is said, not who speaks -- Silence betokens consent -- Love the truth but pardon error.
"Science is about questioning the status quo. Questioning authority".
In the absence of evidence, opinion is indistinguishable from prejudice.

RIP Neil Armstrong

Replies are listed 'Best First'.
Re^16: PANIC: underlying join failed threded tcp server
by rmahin (Scribe) on Nov 16, 2012 at 00:21 UTC
    To anyone who is interested in this thread, now or in the future, I believe browserUk's diagnosis that the problem lies in perl 5.16.1. I have now seen this error in several other instances on machines running this perl version, on windows 2008, windows 7, and windows 2003 where they are forced to join a large number of threads. My solution, that seems to be working is to use the thread::queue module. My circumstannces required no external modules, so i put the code i need at the end of my code in another package. Not terribly clean, I'm not sure what the general outlook is regarding coding practices for having more than one package in the same file, but...it was a work around that i found useful. This is the code for the main thread now, with the thread queue at the bottom. Made some other minor adjustments but now you can see the overall structure of the program. 
    ######################################################################
+#######
######## MAIN THREAD    
my $lsn = new IO::Socket::INET(
    Listen => SOMAXCONN,
    LocalPort => $port,
    ReuseAddr => 1,
    Blocking => 1,
    
    ) or die "Failed to open listening port: $!\n";

print "RXD+ had been started on port $port\n";
debug("Password is '$password'");


my @threads = ();
$doneQ = MyQ->new();
$clientQ = MyQ->new();
for(1..$_numThreads){
    my $thd = threads->create('processRequest');
    push(@threads, $thd);
}

debug("Thread creation finished");
#making the CWD the directory RXD is located in
use FindBin qw($Bin);
my $currentDir = $Bin;
if ($^O =~ /win/i) {
    $currentDir =~ tr/\//\\/;
} elsif (($^O !~ /win/i) && ($currentDir !~ /\//)) {
    $currentDir .= "/";
}

chdir($currentDir);

# create a cache hash
while(1) {
    my $client;
    my $fd;
    debug("Accepting connection, there are currently " . $clientQ->pen
+ding() . " connections in the queue");
    unless ($client = $lsn->accept) {
        tprint ("Could not connect to socket: " . $!);
        next;
    }
    debug("Connection accepted");
    if(!defined($client)){tprint("Could not connect to client"); next;
+}

    # change filehandle to file descriptor
    $fd = fileno $client;
    unless(defined($fd)){close($client);next;}
    
    # add filehandle to cache hash
    # tprint("Caching file descriptor- $fd");
    $FDcache{$fd} = $client;
        
    debug("Enqueueing");
    $clientQ->enqueue($fd);
    
    while(my $done = $doneQ->dequeue_nb()){
        debug("Deleting from cache");
        delete $FDcache{$done};
    }
    

}

sub shutdownRXD{
    tprint("Sutting down");
    exit 0; 
}
######################################################################
+#######
package MyQ;
use strict;
use warnings;

use threads::shared;
# Create a new queue possibly pre-populated with items
sub new
{
    my $class = shift;
    my @queue :shared = map { shared_clone($_) } @_;
    return bless(\@queue, $class);
}

sub enqueue
{
    my $queue = shift;
    lock(@$queue);
    # print "Q> enqueueing\n";
    push(@$queue, map { shared_clone($_) } @_)
        and cond_signal(@$queue);
}

# Return a count of the number of items on a queue
sub pending
{
    my $queue = shift;
    lock(@$queue);
    return scalar(@$queue);
}


# Return 1 or more items from the head of a queue, blocking if needed
sub dequeue
{
    my $queue = shift;
    lock(@$queue);
    # print "Q> dequeing\n";
    # Wait for requisite number of items
    cond_wait(@$queue) until (@$queue >= 1);
    cond_signal(@$queue) if (@$queue > 1);

    # Return single item
    return shift(@$queue);

}

# Return items from the head of a queue with no blocking
sub dequeue_nb
{
    my $queue = shift;
    lock(@$queue);

    # Return single item
    return shift(@$queue);
}

    [download]
[reply]
[d/l]

In Section Seekers of Perl Wisdom