Beefy Boxes and Bandwidth Generously Provided by pair Networks
Think about Loose Coupling
 
PerlMonks  

Multithreaded Server Crashes under heavy load

by rmahin (Beadle)
on Aug 28, 2012 at 23:55 UTC ( #990358=perlquestion: print w/ replies, xml ) Need Help??
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."; }

Comment on Multithreaded Server Crashes under heavy load
Download Code
Re: Multithreaded Server Crashes under heavy load
by zentara (Archbishop) on Aug 29, 2012 at 10:39 UTC
    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,

    Are you running into a system resources limit? Have you checked your memory usage of the script when it hits 500 threads?

    Google for ulimit windows for some tips.


    I'm not really a human, but I play one on earth.
    Old Perl Programmer Haiku ................... flash japh
Re: Multithreaded Server Crashes under heavy load
by BrowserUk (Pope) on Aug 29, 2012 at 12:13 UTC

    First pass: Change use threads; to use threads stack_size => 4096; and see if things get better.

    Next, use whatever tools are available on your system to instrument memory, cpu and socket activity for the process as it is running and capture it to a file. Make that log (sanitised as appropriate) available on line and post a link to it.


    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

    div
      Made the change to use threads but it produced the same result. Looking for a tool to use to capture the information requested now. Is there a tool that you could recommend that can capture all of that information on a Windows environment? Thanks.
        Is there a tool that you could recommend that can capture all of that information on a Windows environment?

        Yes. Type perfmon.exe into a command line.

        There is a bit of a learning curve involved in using it, but it is gui with pretty extensive help built in.


        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

Re: Multithreaded Server Crashes under heavy load
by bulk88 (Priest) on Aug 30, 2012 at 06:00 UTC
    Have you considered attaching a C debugger to the crashed process (the "Debug" button in the Error Reporting popup) and getting a C callstack?

    Or atleast posting "To see what data... click here"->"Error Signature" area, or slightly better, "To see what data... click here"->"technical information about the error report, click here"->upper text box (but you can't right-click copy that information AFAIK, I'm most interested in "Code:" number, if its not 0xC0000005, you are one really lucky person and the crash is much more interesting).

    What version number of Perl are using, and ActivePerl or Strawberry Perl?
      Hi thanks for the idea, will install visual studio and try the C debugger soon, and im using ActivePerl version 5.14.2

Log In?
Username:
Password:

What's my password?
Create A New User
Node Status?
node history
Node Type: perlquestion [id://990358]
Approved by marto
help
Chatterbox?
and the web crawler heard nothing...

How do I use this? | Other CB clients
Other Users?
Others wandering the Monastery: (5)
As of 2014-10-22 02:53 GMT
Sections?
Information?
Find Nodes?
Leftovers?
    Voting Booth?

    For retirement, I am banking on:










    Results (112 votes), past polls