Multithreaded Server Crashes under heavy load

rmahin has asked for the wisdom of the Perl Monks concerning the following question:

Hello! I am having a problem with a threaded tcp server, using absolutely no modules not included with all standard distributions of perl, many of our computers do not have internet access, and installing modules is not something users want to have to deal with. The server listens for connection, after receiving each one, creates a thread with the sockets file descriptor, and then detaches the thread. The will send various information, and wraps information in tags like ***PASSWORD BEGIN***password***PASSWORD END***, as well as commands to be issued. The main purpose of this server is to be able to execute system commands on the machine running the server, so for example in this code we have the command EXEC, which runs the command passed, so if the argument was 'DIR c:\ /s' it runs that command. This works very well, however when its under heavy stress (around like 500 connections being made at approximately the same time), and runs commands that take a while to process, the server sometimes does one of three things:

1. Simply hangs -- will not accept any more connection, and will give no sign of failure.

2. Crashes and executes the end block

3. Crashes and does not execute the end block.

I have been completely unable to pin point why this happens for any of them, so I seek the perl monk wisdom.

I took out many of the other possible commands in the code, but left ones that we use most, and that seem to be causing this behavior. If needed I'll post the client code as well, if you need to test it out, but that does not seem to be the source of the problem. The client essentially just sends commands like "***RX BEGIN******PASSWORD BEGIN***pw***PASSWORD END******COMMAND BEGIN***EXEC***COMMAND END******ARGUMENTS BEGIN***dir c:\test2***ARGUMENTS END******RX END***"

Thanks for the help!

use strict;
use warnings;
use threads;
use threads::shared;
use IO::Handle;
use IO::Socket::INET;
use File::Find;
use File::Path;
use Digest::SHA;

######################################################################
+##
######## ARGUMENT CHECKING
######################################################################
+#
my $debug = 0;
my $port  = 1600;

while ( defined($ARGV[0]) and substr( $ARGV[0],0,1 ) eq "-" ) {
    if ( $ARGV[0] eq "-d" ) { $debug = 1; shift; next; }
    if ( $ARGV[0] eq "-p" ) { $port = $ARGV[1]; shift; shift; next; }
    if ( $ARGV[0] eq "-h" ) { usage(); exit 1; }
    print "Warning: Invalid argument '".$ARGV[0]."' ignored.\n";
    shift;
}

my $password = $ARGV[0] || "hot34tot";
######################################################################
+####
######## Thread-safe printing functions
######################################################################
+#
$|++;

my $semSTDOUT :shared;
sub tprint{
    lock $semSTDOUT;
    my $msg = shift;
    my $tid = shift || threads->tid();
    if($tid ne ""){$tid = "($tid)"};
    print "RXD+ $tid > $msg \n";
}
sub debug{
    if($debug){
        lock $semSTDOUT;
        my $msg = shift;
        my $tid = shift || threads->tid();
        if($tid == 0){ $tid = "MAIN";}
        if($tid ne ""){$tid = "($tid) "};

        print "DEBUG -- " . $tid . $msg . "\n";;
    }
}

sub usage
{
    print "Usage: rxd [-d][-p Port] Password\n";
}

######################################################################
+####


######################################################################
+####
######## SEND\RECEIVE parsing functions
######################################################################
+#
sub receiveCommand{
    my $client = shift;
    debug("Waiting for command input");
    
    my $command = "";
    my $args = "";
    my $buffer = '';
    my $i = 0;
    while( my $num = sysread($client, $buffer, 100000, length($buffer)
+))  { #attempt to receive 100000 bytes at a time
        $i++;
        my $len = length($buffer);
        debug($num . " bytes received, bringing the total to " . $len)
+;

        $command = flagUnwrap("RX", $buffer);
        if( $command ne ""){
            last;
        }
        if($i > 999999999){ #infinite loop stopper
            debug("Terminating infinite loop, closing connection to $c
+lient"); # just in case, but sysread would probably block anyways
            last;
        }
    }
    if(length($buffer) < 300){
        debug("Raw text received = $buffer");
    }else{
        debug("Received " . length($buffer) . " bytes from RX");
    }
    return ($command);
    
}


sub sendResponse{
    my $msg = shift;
    my $rc = shift;
    my $client = shift;
    $msg = flagWrap("RESPONSE", $msg); 
    my $rcMSG = flagWrap("RETURN CODE", $rc);
    
    my $response = flagWrap("RXD", $msg . $rcMSG);
    my $len = length($response);
    if($len < 200){
        debug("Command complete, sending response '$response'");
    }else{
        debug("Command complete, sending $len bytes in response.");
    }
    tprint("Command completed with a return code of $rc", threads->tid
+());
    syswrite($client, $response, length($response)) or debug(">>ERROR<
+< Couldnt write to client");
    debug("sent");
}
######################################################################
+####




######################################################################
+####
######## Worker Thread subroutine
######################################################################
+#
sub processRequest {
    my $client;
    my $tid = threads->tid;
    my $fno = shift;

    open $client, "+<&", $fno or threads->exit;

    $client->autoflush(1);
    debug("Connected to client machine");
    my $response = "";
    my $returnCode = "";

    my $fullCMD = receiveCommand($client);
    my $pswd = flagUnwrap("PASSWORD", $fullCMD);
    my $command = uc(flagUnwrap("COMMAND", $fullCMD)); #Command ie EXE
+C
    my $args = flagUnwrap("ARGUMENTS", $fullCMD); #arguments for comma
+nd
    tprint("$command $args");
    my $additionalData = flagUnwrap("ADDITIONAL DATA", $fullCMD); #lik
+e a file
    my $shaValue = flagUnwrap("SHA VALUE", $fullCMD);
    my $readMoreSize = flagUnwrap("READ SIZE", $fullCMD); #indicating 
+a long file
    if($pswd eq $password){
        #debug($readMoreSize);
        if($readMoreSize ne ""){ #RX is going to try to send more stuf
+f!
            debug("Now waiting for the client to send $readMoreSize mo
+re bytes");
            my $i = 0;
            while(length($additionalData) < int($readMoreSize) and $i 
+< 9999999){
                $i++;
                my $num = sysread($client, $additionalData, $readMoreS
+ize, length($additionalData));
                my $len = length($additionalData);
                if(defined($num)){
                        debug($num . " bytes received, bringing the to
+tal to " . $len);
                }
            }
        }
        
        my $len = length($additionalData);
        if($len > 0){
            if($len < 200 and $additionalData !~ /\n/ms){
                tprint("Additional data received: $additionalData");
            }else{
                tprint("Received $len bytes of additional data");
            }
        }
        ($response, $returnCode) = executeCommand($command, $args, $ad
+ditionalData, $client, $shaValue);
    }else{
        $response = "Incorrect password specified";
        $returnCode = 1;
    }
     
    #since execprint needs to be able to send while simultaneously exe
+cuting the command
    #cannot store output in a string to send at once
    if ($command ne 'EXECPRINT' and $command ne 'GET' and $command ne 
+'GETB') {
        sendResponse($response, $returnCode, $client);
    }

    debug("Closing connection");
    shutdown($client, 2);
    close $client;
    threads->exit;
}


sub executeCommand{
    my $rxdCMD = uc(shift);
    my $rxdArgs = shift;
    my $rxdData = shift;
    my $client = shift;
    my $shaValue = shift || "";
    my $resp = "";
    my $rc = 0;

    if($rxdCMD ne ""){
        debug("Processing command '$rxdCMD'");
        if ( $rxdCMD eq 'EXEC' && $rxdArgs ne '' )   # EXEC(Command)
        {

            if ( $^O eq "MSWin32" ) {
                #$resp = `$rxdArgs 2>>&1`;
                open(my $fs, $rxdArgs . " 2>&1|") or threads->exit;
                select((select($fs), $| = 1)[0]);
                while(my $line = <$fs>){
                    $resp .= $line;
                }
                close($fs);
            }else {
                $rxdArgs =~ s/ \*/ \\\*/g;
                #$resp = `$rxdArgs 2>&1`;
                open(my $fs, $rxdArgs . " 2>&1|") or threads->exit;
                select((select($fs), $| = 1)[0]);
                $resp = <$fs>;
                close($fs);
            }
            $rc = int($?>>8);
        }
        elsif ( $rxdCMD eq 'MSG' && $rxdArgs ne '')   # EXEC(Command)
        {
            print "MESSAGE FROM RX: $rxdArgs\n";
        }

        elsif ( $rxdCMD eq 'PUT' or $rxdCMD eq 'PUTB' && $rxdArgs ne '
+' )   # PUT/PUTB(Filespec)
        {
            my $outfile = "";
            (my $unused, $outfile) = getDirs($rxdArgs);

            open( PUTFILE, ">$outfile" ) or threads->exit;
            if($rxdCMD eq 'PUTB'){
                debug("setting binmode for PUTFILE");
                binmode (PUTFILE);
            }
                
            if ( $rxdData eq "" ){
                $rc = 1;
                $resp = "File was not received properly";
                close PUTFILE;
                debug("The file was empty!");
            }else{
                print PUTFILE $rxdData;
                $resp = "File was transferred successfully.";
                close PUTFILE;
                debug("File written!");
            }
                                  
            my $rxdSHA = $shaValue;
            debug("Calculating SHA value of file");
            
            my $val = calculateSHA($outfile);
                    
            if($val ne $rxdSHA){
                debug("$val did not equal $rxdSHA");
                $resp = "The file was transferred successfully, howeve
+r there are data integrity errors";
                $rc = 1;
            }else{
                debug("SHA values were a match");
            }
        }
        #########################  NEW CMDS ##########################
+####################
        elsif ( $rxdCMD eq 'EXECPRINT' && $rxdArgs ne '' )   # EXEC(Co
+mmand)
        {
            my $rxdBegin = "***RXD BEGIN***";
            my $h= "***RESPONSE BEGIN***";
            my $f= "***RESPONSE END***";
            my $r= "***RETURN CODE BEGIN***";
            my $r2= "***RETURN CODE END***";
            my $rxdEnd = "***RXD END***";
            open(my $fs, $rxdArgs . " 2>&1|") or threads->exit;
            select((select($fs), $| = 1)[0]);
            syswrite($client, $rxdBegin.$h, length($rxdBegin.$h));
            if(defined($fs)){
                while(my $line = <$fs>){
                    syswrite($client, $line, length($line));
                }
                close ($fs);
                $rc = int($?>>8);
            }else{
                $rc = 1;
            }
            my $end = $f . $r . $rc . $r2 . $rxdEnd;
            syswrite($client, $end, length($end));
            tprint("Command completed with a return code of $rc");
        }
        elsif ( $rxdCMD eq 'GETFILESYSINFO' && $rxdArgs ne '' )  # GET
+FILESYSINFO(Command)
        {
            # verify the directory exist
            if (-d $rxdArgs) {
                # check if trailing slash was added... if not add it
                if ( $^O eq "MSWin32" ) {
                    unless ($rxdArgs =~ /\\$/) {
                            $rxdArgs .= "\\";
                    }
                } else {
                    unless ($rxdArgs =~ /\/$/) {
                        $rxdArgs .= "/";
                    }
                }
                
                # set the SHA object
                my $sha = Digest::SHA->new(256);
                # use File::Find to step the dir structure
                my $string;
                my $error;
                find({wanted => sub {wanted(\$string, \$sha, \$error);
+}, no_chdir => 1}, $rxdArgs);
                           
                sub wanted {
                    my $string = shift;
                    my $sha = shift;
                    my $error = shift;
                    # check if current object is a file
                    if (-f  $File::Find::name) {
                        # Calculate the SHA value
                        eval {
                            $$sha->addfile($File::Find::name, "p");

                            my $value = $$sha->hexdigest;
                            
                            # calculate the size of the file in bytes
                            my $size = -s $File::Find::name;
                            
                            # checking for any XML special chars in th
+e name
                            my $name = $File::Find::name;
                            $name =~ s/&/&amp;/g;
                            $name =~ s/</&lt;/g;
                            $name =~ s/>/&gt;/g;
                            $name =~ s/"/&quot;/g;
                                                       
                            # set results to $string and send $string 
+to rx.pl
                            $$string .= "<file><name>" . $name . "</na
+me><sha>" . $value . "</sha><size>" . $size . "</size></file>";
                        };
                        
                        if ($@) {
                            $$error .= $File::Find::name . "\n";
                        }
                    }
                }
                
                if (defined $error) {
                    $resp = "Erorrs occured on the following files:\n"
+ . $error;
                    $rc = 1;
                } else {
                    $resp = "<sha_values>" . $string . "</sha_values>"
+;
                    $rc = 0;
                }
            } else {
                $resp = "Directory does not exist.";
                $rc = 1;
            }
        }


        elsif ($rxdCMD eq 'MKDIR' && $rxdArgs ne '') {
             
            if (mkpath($rxdArgs)) {
                $resp = "Made directory \"" . $rxdArgs . "\"";
                $rc = 0;
            } else {
                $resp = "Failed to make directory \"" . $rxdArgs . "\"
+";
                $rc = 1;
            }
        }

        elsif ($rxdCMD eq 'RMDIR' && $rxdArgs ne '') {
            if (rmtree($rxdArgs)) {
                $resp = "Removed directory \"" . $rxdArgs . "\"";
                $rc = 0;
            } else {
                $resp = "Failed to remove directory \"" . $rxdArgs . "
+\"";
                $rc = 1;
            }
        }

        elsif ($rxdCMD eq 'RMFILES' && $rxdArgs ne '') {

            if (rmtree($rxdArgs, {keep_root => 1})) {
                $resp = "Removed files in directory \"" . $rxdArgs . "
+\"";
                $rc = 0;
            } else {
                $resp = "Failed to remove files in directory \"" . $rx
+dArgs . "\"";
                $rc = 1;
            }
        }

        else{
            $resp = "Unrecognized command. Pleas refer to the document
+ation for a list of possible commands.";
            $rc = 1;
        }


    }else{
        $rc = 1;
        $resp = "Command was not passed successfully.";
    }
        
    if($resp eq ''){
        if($rc){
            $resp = "Command was not completed successfully.";
        }else{
            $resp = "Command was completed successfully.";
        }
    }
    debug("Command execution completed!");
    return ($resp, $rc);
}

sub calculateSHA{
    my $file = shift;
    my $val = 0;
    open(my $fh, $file) or threads->exit;
    eval{
        my $sha1 = Digest::SHA->new(256);
        $sha1->addfile($fh);
        $val = $sha1->hexdigest;
    };
    if($@){
        debug("An error occurred calculating the SHA-Value. Cannot ver
+ify integrity of file\n$@");
    }
    close($fh);
    return $val;
}

sub getDirs{
    my $args = shift;
    
    my $dir1 = "";
    my $dir2 = "";

    $dir2 = $2 if ($args =~ /"(.+)"\s+"(.+)"/i);
    if($dir2 eq ""){
        $dir1 = $args;
        # $dir2 = $dir1;
    }else{
        $dir1 = $1 if ($args =~ /"(.+)"\s+"(.+)"/i);
    }
    
    debug("Source file is: " . $dir1); 
    debug("Destination file is: " . $dir2); 
    
    
    return ($dir1, $dir2);
}


sub flagWrap {
    my $flag = shift;
    my $string = shift;
    my $flagPattern = '***';
    return $flagPattern . $flag . " BEGIN" . $flagPattern . $string . 
+$flagPattern . $flag . " END" . $flagPattern;
}

sub flagUnwrap{
    my $flag = shift;
    my $string = shift;
    my $flagPattern = '\*\*\*';
    my $regexH = $flagPattern . $flag . " BEGIN" . $flagPattern;
    my $regexT = $flagPattern . $flag . " END" . $flagPattern;
    if($string =~ /$regexH(.+)$regexT/ims){
        return $1;
    }else{
        return "";
    }
}
######################################################################
+####


######################################################################
+#######
######## MAIN THREAD    
my $lsn = new IO::Socket::INET(
    Listen => 100,
    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'");

while(1) {
    my $client = $lsn->accept;
    if(!defined($client)){tprint("Could not connect to client"); next;
+}
    debug("Creating a new thread");
    my $thd = threads->create('processRequest', fileno($client)) or ex
+it;
    $thd ->detach();
    debug("New thread created TID: " . $thd->tid());
}
sub shutdownRXD{
  exit 0; 
}
######################################################################
+#######


END{
    print "The RXD server has been shutdown.";
}
[download]

Back to Seekers of Perl Wisdom