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

Re^2: Sharing sockets between the main script and thread

by markseger (Beadle)
on Nov 28, 2008 at 21:57 UTC ( #726680=note: print w/ replies, xml ) Need Help??


in reply to Re: Sharing sockets between the main script and thread
in thread Sharing sockets between the main script and thread

Looks like I finally got this working, sort of, and so am posting what I have. However, I significantly cut back my testing code and it has a very odd quirk in it. There's a print statement at the very top of the loop in the server that simply print 'top'. If I comment it out, the script hangs after the first connection - it eventually wakes up but things get very ugly. If I remove the comment it works fine. Also, if while hanging I do a 'netstat -a', it also wakes up. very bizarre

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...

#!/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"; }
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 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"; }
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


Comment on Re^2: Sharing sockets between the main script and thread
Select or Download Code
Re^3: Sharing sockets between the main script and thread
by BrowserUk (Pope) on Nov 29, 2008 at 13:06 UTC
    If anyone has any clue why it only works correctly with that "print top" statement

    Quite frankly, if your code does indeed work correctly with that "print top" statement, then I'd just leave that statement there. Then, I'd contact the Magic Circle and show it to them because they'd probably pay you a big lump of cash once they work out how it "works", because they'll be able to use for as the basis of some damn good illusions.

    If you add use strict: to the top of your program and fix all these errors:

    C:\test>perl -c junk8.pl Global symbol "$q1" requires explicit package name at junk8.pl line 16 +. Global symbol "$q2" requires explicit package name at junk8.pl line 17 +. Global symbol "$q1" requires explicit package name at junk8.pl line 18 +. Global symbol "$q2" requires explicit package name at junk8.pl line 18 +. Global symbol "%sockOpened" requires explicit package name at junk8.pl + line 31. Global symbol "%sockOpened" requires explicit package name at junk8.pl + line 31. Global symbol "%sockOpened" requires explicit package name at junk8.pl + line 32. Global symbol "$q1" requires explicit package name at junk8.pl line 34 +. Global symbol "$q2" requires explicit package name at junk8.pl line 35 +. Global symbol "%sockOpened" requires explicit package name at junk8.pl + line 40. Global symbol "%sockOpened" requires explicit package name at junk8.pl + line 40. Global symbol "$bytes" requires explicit package name at junk8.pl line + 46. Global symbol "%sockOpened" requires explicit package name at junk8.pl + line 46. Global symbol "$bytes" requires explicit package name at junk8.pl line + 49. Global symbol "$bytes" requires explicit package name at junk8.pl line + 53. Global symbol "$port" requires explicit package name at junk8.pl line +63. Global symbol "$port" requires explicit package name at junk8.pl line +68. Global symbol "$port" requires explicit package name at junk8.pl line +68. Global symbol "$port" requires explicit package name at junk8.pl line +69. Global symbol "$sockNumConn" requires explicit package name at junk8.p +l line 87. Global symbol "$sockNumConn" requires explicit package name at junk8.p +l line 114. Global symbol "$time" requires explicit package name at junk8.pl line +133. Global symbol "$time" requires explicit package name at junk8.pl line +134. junk8.pl had compilation errors.

    You might get close to understanding some of your problems. The entire logic of your code is dependant upon this hash %sockOpened. You use it to direct the flow of your program all over the place:

    $sockOpened{$fn}->close() if defined($sockOpened{$fn}); ... if (!defined($sockOpened{$fn}) && !open($sockOpened{$fn}, ">&$ +fn")) { ... $bytes = syswrite($sockOpened{$fn}, "$count/n", length($count) ++1, 0);

    But you never declare that variable, and you never write to it, so what is it that you are testing?

    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...

    Sorry, but it doesn't seem to be working for you.


    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.
      First and foremost I do tend to get lazy with test scripts and indeed was so with respect to this one and 'strict'. Sorry about that. After I posted this note and continued to plug away, that was one of the next things I did. It didn't help...

      As for $sockOpened not being used, it IS being used. If you look at then 'open' statement you'll see that array contains the file descriptors of each opened socket.

      In any event I finally found the problem! The bottom line is that although I do my locking at the top of the main while loop, there is virtually no time for the CPU to yield any time to the thread, which itself is hanging on a lock(). The 'print' statement causes it to yield and for that matter replacing it with anything that will yield the cpu will do so. I tried threads->yield which also worked as well as sleeping for 0.01 with select. In any event, if you want to take a look at one that does work AND includes strict:

      #!/usr/bin/perl -w use Time::HiRes; use IO::Socket; use IO::Select; use threads; use threads::shared; use Thread::Queue; use strict; $SIG{"INT"}=\&sigInt; # for ^C $SIG{"PIPE"}=\&sigPipe; # socket comm errors my %sockConns; share(%sockConns); my $q1 = new Thread::Queue; my $q2 = new Thread::Queue; my $thread=threads->create('manageSock', $q1, $q2)->detach; $|=1; my $done=0; my $count=0; my %sockOpened; while(!$done) { select(undef, undef, undef, .01); $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"); my $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 } sub manageSock { my $q1=shift; my $q2=shift; my $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); my $sockNumConn; 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(); my $time=sprintf("$intSeconds.%06d", $intUsecs); print "$time $text\n"; }
      There actually is a bonus question to all this which I suspect is buried in the guts of TCP. If you run the server with the 'select' at the top of the loop commented out and then run the client scrip from the base note with the format 'client address 1', it will connect the server, read a record, close the connection and try to reestablish the connection. Everything will hang. If you immediately do a netstat -a, for some unknown reason the server immediately wakes up and I have no idea why...

      In any event if you DO leave in the select it runs just fine.

      -mark

        There actually is a bonus question .... If you run the server .... Everything will hang.

        Maybe it's a platform difference, but I commented out the select at the top, and then the sleep at the bottom, and nothing changed. No hang. Everything just ran a little faster.

        As for the code. You are using a shared hash, locking, and two queues to achieve what this code achieves with a single queue and no locking. And for no benefit that I can perceive.


        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.

Log In?
Username:
Password:

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

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

    For retirement, I am banking on:










    Results (129 votes), past polls