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

[markseger]

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

[BrowserUk]

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.

[markseger]

Clearly the code you posted is more economic and I suspect it will grow as more error handling is added. For example it looks like it opens a socket and keeps writing to it until the socket closes and that's not the behavior I'm looking for. Rather I want my worker thread to manage open/closing the socket while the mainline does the I/O to it. If multiple sockets are opened, the mainline will write the same strings to all of them, dealing with each connections independently closing/reconnecting. The main line should not care how many (if any) connections are active.

That's probably not all that obvious from looking at the code itself.

-mark

[BrowserUk]

All of which is completely irrelevant as I'm only referring to the mechanism of transferring handles from one thread to another. Still, I've noticed that there is a propensity for some people to prefer complicated to simple.