Changeset 24840

Timestamp:

2011-12-01T12:25:06+13:00 (12 years ago)

Author:

jmt12

Message:

Added lots of comments as I was trying to determine the cause of the exhausted socket pool problem

File:

• gs2-extensions/parallel-building/trunk/src/perllib/SocketsSwimmingThreadPoolServer.pm (modified) (5 diffs)

gs2-extensions/parallel-building/trunk/src/perllib/SocketsSwimmingThreadPoolServer.pm

@@ -33 +33 @@
     processor_cb => $param{processor_cb} || \&processor,
     eom_marker => $param{eom_marker} || "\\n\\.\\n",
-    thread_pool => undef
+    thread_pool => undef,
+    listen_queue => 5
   } => $class;
 }
@@ -71 +72 @@
 }
 
-sub main_loop {
-  my $self= shift;
-  my $counter= 1;
-  until($stop) {
+sub main_loop
+{
+  my $self = shift;
+  my $counter = 1;
+  until($stop)
+  {
     $self->{main_cb}->($counter++, sub {$self->stop});
     sleep $self->{main_yield};
@@ -80 +83 @@
 }
 
-sub accept_requests {
-  my $self= shift;
-  my ($csocket, $n, %socket);
-  my $lsocket= new IO::Socket::INET(
-    %{$self->{socket_defaults}},
-    Proto => 'tcp',
-    Listen => 1,
-    Reuse => 1);
+sub accept_requests
+{
+  my $self = shift;
+  my %socket;
+  # lsocket => listerner [sic] socket
+  my $lsocket = new IO::Socket::INET(%{$self->{socket_defaults}}, Proto => 'tcp', Listen => 1, Reuse => 1);
   die "Can't create listerner socket. Server can't start. $!." unless $lsocket;
-  until($stop) {
-    $csocket= $lsocket->accept;
-    $n= fileno $csocket;
-    $socket{$n}= $csocket;
+  until($stop)
+  {
+    # csocket => connected socket (for doing the actual communications)
+    my $csocket = $lsocket->accept;
+    my $n = fileno $csocket;
+    $socket{$n} = $csocket;
+    # add this to the queue that a virtual horde of request_handlers are
+    # waiting upon (much like vultures)
     $accept_queue->enqueue($n . ' ' . inet_ntoa($csocket->peeraddr));
-    while($n= $closed_queue->dequeue_nb) {
+    # we then go ahead and destroy any sockets that we are no longer using.
+    # Note the use on a non-blocking dequeue (so it returns straight away) and
+    # that $n may be undefined (if the queue is empty) skipping the while loop
+    # altogether
+    while($n = $closed_queue->dequeue_nb)
+    {
+      # note: 2 = finished both reading and writing to this socket
       $socket{$n}->shutdown(2);
-      delete $socket{$n}}}
+      delete $socket{$n}
+    }
+  }
+  # note: 2 = finished both reading and writing to this socket
   $lsocket->shutdown(2);
 }
 
+# @function request_handler()
+#
+# There are *thread_count* request_handler threads all sitting, waiting, to
+# process any incoming reqeuests received via the socket. Note that the whole
+# 'race for data' issue is being handled by the Thread::Queue->dequeue command
+# which is, according to the documentation, thread safe. I presume the actual
+# dequeuing is somehow synchronized.
 sub request_handler
 {
@@ -107 +128 @@
   until($stop)
   {
-    ($n, $ip)= split / /, $accept_queue->dequeue;
+    # note: dequeue is blocking so this thread will wait here until something
+    # is enqueued
+    ($n, $ip)= split / /, $accept_queue->dequeue();
     next unless $n;
+    # open socket for reading and writing... reading first of course
     open my $socket, '+<&=' . $n or die $!;
+    # this will read in the payload from the socket until the end of message
+    # sentinel value is detected. Note that this is done in a loop, so the
+    # client side of the socket may send multiple requests. Eventually the
+    # client will close their end of the socket causing the data to be
+    # undefined (maybe - I don't quite understand the defining construct - see
+    # below) and breaking out of the loop.
     if(defined($data = $self->receive_client_request($socket)))
     {
+      # we then pass the payload to the registered processor function (provided
+      # by the caller) to do the actual legwork, the result of which is written
+      # back to the socket.
       my $result = $self->{processor_cb}->($data, $ip, threads->tid, sub {$self->stop});
       print $socket $result, "\n.\n";
     }
+    # this is where the socket is closed... so why am I eventually hitting a
+    # "cannot assign address" problem?
     close $socket;
+    # add this socket number to the list of sockets to be destroyed
     $closed_queue->enqueue($n);
   }
@@ -123 +159 @@
 {
   my ($self, $socket)= @_;
+  # how does this construct work? eom_marker is a scalar (a string) so how can
+  # it be assigned to an anonymous array?
   my ($eom, $buffer, $data)= $self->{eom_marker};
-  while($buffer= <$socket>)
+  while ($buffer = <$socket>)
   {
-    $data.= $buffer;
+    $data .= $buffer;
     last if $data =~ s/$eom$//;
   }