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

[markseger]

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