That said, here's the server code. The way it works is it receives a connection from one or more clients and then starts printing the value of $count to them. There's also a sleep statement at the bottom of the main loop which you can uncomment to speed up the amount of messages sent to the client. You can start/stop one or more clients while this is running and other than that problem with the print statement it seems very solid to me. You can also stop/start the server and the clients will reconnect when the socket becomes available.
Also note I have a bunch of logging messages that helped me coordinate problems between the client and the server. Easy enough to turn off.
SO here's the main server code with LOTS of error handling...
And here's the client I test it with. To run it the first argument is the address for the server - I've been doing all my testing with both client/server on the same system. If you specify a second argument, the client will read a response, sleep for a second and read another, looping until you ^C and it will exit cleanly so you can restart it. As I said you can run multiple instances, starting/stopping them and they do the right thing. Finally, if you give it a 3rd argument it will skip the sleep 1 and connect/disconnect as fast as possible.#!/usr/bin/perl -w use Time::HiRes; use IO::Socket; use IO::Select; use threads; use threads::shared; use Thread::Queue; $SIG{"INT"}=\&sigInt; # for ^C $SIG{"PIPE"}=\&sigPipe; # socket comm errors my %sockConns; share(%sockConns); $q1 = new Thread::Queue; $q2 = new Thread::Queue; my $thread=threads->create('manageSock', $q1, $q2)->detach; my $done=0; my $count=0; while(!$done) { print "top\n"; $count++; lock(%sockConns); foreach my $fn (keys %sockConns) { logit("FN: $fn=$sockConns{$fn}"); if ($sockConns{$fn}==-1) { logit(">>>Close 1st: $fn"); $sockOpened{$fn}->close() if defined($sockOpened{$fn}); delete $sockOpened{$fn}; delete $sockConns{$fn}; $q1->enqueue($fn); my $wait=$q2->dequeue; logit("Continue..."); last; } if (!defined($sockOpened{$fn}) && !open($sockOpened{$fn}, ">&$fn") +) { print "Couldn't open socket $fn for writing\n"; next; } logit("Write: $count TO: $fn"); $bytes=syswrite($sockOpened{$fn}, "$count/n", length($count)+1, 0) +; # Do nothing as socket will disconnet and normal cleanup will kick + in if (!$bytes) { logit("========================> Comm Failure <================= +==="); last; } logit("Wrote $bytes bytes"); } sleep 1; # uncomment to slow responses down # print "awake\n"; } sub manageSock { my $q1=shift; my $q2=shift; $port=2655; my $sockServer = new IO::Socket::INET( Type=>SOCK_STREAM, Reuse=>1, Listen => 1, LocalPort => $port) || error("Could not create local socket on port $port"); logit("Server socket opened on port $port"); my $select=new IO::Select($sockServer); while(1) { logit("Waiting on socket"); while (my @ready=$select->can_read) { my $saveFnum; my $saveHandle; my $waitForClose=0; foreach my $filehandle (@ready) { lock(%sockConns); logit("Socket 'can read'"); if ($filehandle==$sockServer) { my $new=$sockServer->accept() || logmsg('E', "Couldn't accep +t connection request"); $select->add($new); my $fnum=$new->fileno(); $sockConns{$fnum}=0; $sockNumConn++; logit("Connection on FN: $fnum"); } else { my $message=<$filehandle>; my $fnum=$filehandle->fileno(); if (!defined($message)) { logit("Client Disconnect FN: $fnum"); $saveFnum=$fnum; $saveHandle=$filehandle; $waitForClose=1; $sockConns{$fnum}=-1; last; } else { logit("Ignoring: $message"); } } } if ($waitForClose) { logit("Waiting for 1st socket close"); my $fnum=$q1->dequeue; $select->remove($saveHandle); $saveHandle->close(); $sockNumConn--; $q2->enqueue($fnum); # tell main process OK to release lock } } } } sub sigPipe { #trap but ignore } sub sigInt { print "^C\n"; $done=1; } sub logit { my $text=shift; my ($intSeconds, $intUsecs)=Time::HiRes::gettimeofday(); $time=sprintf("$intSeconds.%06d", $intUsecs); print "$time $text\n"; }
If anyone has any clue why it only works correctly with that "print top" statement I've love to hear an answer. In the case of my collectl script I don't have this problem, but there is also a lot of other activity going on the main processing loop so perhaps that's why. I also suspect my scripts could be somewhat more compresses but I guess I've always been in the habit of being more verbose so both myself and others could better understand what I'm doing... -mark#!/usr/bin/perl -w use IO::Socket; use IO::Select; use Time::HiRes; if (!defined($ARGV[0])) { print "usage: client.pl address[:port] continuous nosleep\n"; exit; } ($address,$port)=split(/:/, $ARGV[0]); $port=2655 if !defined($port); $contFlag= defined($ARGV[1]) ? 1 : 0; $sleepFlag=defined($ARGV[2]) ? 0 : 1; $SIG{"INT"}=\&sigInt; # for ^C select STDOUT; $|=1; while(1) { logit("OPEN"); $socket=new IO::Socket::INET( PeerAddr => $address, PeerPort => $port, Proto => 'tcp', Timeout =>1); if (!defined($socket)) { logit("Couldn't connect to server, retrying"); sleep 1; next; } $select = new IO::Select($socket); logit("Try to read"); while (my @ready=$select->can_read()) { logit("Can_read"); $bytes=sysread($socket, $line, 100); if ($bytes==0) { logit("Socket closed on other end"); $socket=''; last; } @handles=($select->can_read(0)); last if scalar(@handles)==0; } chomp $line; logit($line); logit("client close"); $socket->close if $socket ne ''; $select->remove($socket); last if !$contFlag; sleep 1 if $sleepFlag; } sub sigInt { print "Close Socket\n"; $socket->close(); exit; } sub logit { my $text=shift; my ($intSeconds, $intUsecs)=Time::HiRes::gettimeofday(); $time=sprintf("$intSeconds.%06d", $intUsecs); print "$time $text\n"; }
In reply to Re^2: Sharing sockets between the main script and thread
by markseger
in thread Sharing sockets between the main script and thread
by markseger
| For: | Use: | ||
| & | & | ||
| < | < | ||
| > | > | ||
| [ | [ | ||
| ] | ] |