Fix PHP warning "Socket disconnected unexpectedly in EventLoop.php:296 in Worker.php on line 32". Also made some function names a bit clearer.

Author: tony-martin-aste

.gitignore

@@ -1,2 +1,4 @@
 .DS_Store
+composer.lock
 vendor/
+php_errors.log

composer.json

@@ -1,7 +1,7 @@
 {
   "name": "crusse/job-server",
   "description": "A library for computing the results of multiple PHP scripts in parallel",
-  "version": "1.0.2",
+  "version": "1.0.3",
   "require": {
     "php": ">=5.3.3",
     "symfony/process": "^2.6"

src/Crusse/JobServer/EventLoop.php

@@ -39,12 +39,12 @@ function __destruct() {
   function connect() {
 
     if ( ( $socket = socket_create( AF_UNIX, SOCK_STREAM, 0 ) ) === false )
-      throw new \Exception( socket_strerror( socket_last_error() ) );
+      throw new \Exception( 'Could not create socket: '. socket_strerror( socket_last_error() ) );
 
     $this->addClientSocket( $socket );
 
     if ( socket_connect( $socket, $this->serverSocketAddr ) === false )
-      throw new \Exception( socket_strerror( socket_last_error() ) );
+      throw new \Exception( 'Could not connect to socket. Peer is probably not listen()ing. '. socket_strerror( socket_last_error() ) );
 
     return $socket;
   }
@@ -106,7 +106,7 @@ function listen( $acceptTimeout = 60 ) {
   }
 
   function subscribe( $callable ) {
-    
+
     if ( !is_callable( $callable ) )
       throw new \InvalidArgumentException( '$callable is not callable' );
 
@@ -119,9 +119,9 @@ function send( $socket, Message $message ) {
       $this->log( 'Error: called send() after stop()' );
       throw new \LogicException( 'Calling send() after stop() is redundant' );
     }
-    
+
     $socketIndex = array_search( $socket, $this->sockets, true );
-    
+
     if ( $socketIndex === false ) {
       $this->log( 'Error: $socket was not found in list of clients' );
       throw new \InvalidArgumentException( 'No valid socket given' );
@@ -130,24 +130,27 @@ function send( $socket, Message $message ) {
     $this->writeBuffer[ $socketIndex ] .= (string) $message;
   }
 
-  function run() {
+  function receive() {
 
     $this->log( 'Using select() timeout of '. $this->acceptTimeout .' s' );
 
     while ( true ) {
 
       // We have no more sockets to poll, all have disconnected
       if ( !$this->sockets && !$this->serverSocket ) {
-        $this->log( 'No more sockets to poll, exiting run() loop' );
+        $this->log( 'No more sockets to poll, exiting receive() loop' );
         break;
       }
 
       $readables = $this->sockets;
       if ( $this->serverSocket )
         $readables[] = $this->serverSocket;
 
-      $writableSocketKeys = array_keys( array_filter( $this->writeBuffer ) );
-     
+      // We're only interested in the sockets for which we have any buffered
+      // data to send.
+      // Note: array_filter() preserves keys.
+      $writableSocketKeys = array_keys( array_filter( $this->writeBuffer, 'strlen' ) );
+
       if ( $writableSocketKeys ) {
         $writables = array();
         foreach ( $writableSocketKeys as $key )
@@ -176,21 +179,23 @@ function run() {
         $this->handleWritableSockets( $writables );
 
       if ( $this->stop ) {
-        $this->log( 'stop() was called, exiting run() loop' );
+        $this->log( 'stop() was called, exiting receive() loop' );
         break;
       }
     }
 
     foreach ( $this->sockets as $socket )
       $this->disconnect( $socket );
+
     $this->sockets = array();
 
-    if ( $this->serverSocket )
+    if ( $this->serverSocket ) {
       $this->disconnect( $this->serverSocket );
+      $this->serverSocket = null;
+    }
   }
 
   function stop() {
-    
     $this->stop = true;
   }
 
@@ -218,7 +223,7 @@ private function handleReadableSockets( $sockets ) {
       }
 
       if ( $this->stop ) {
-        $this->log( 'stop() was called, so will not read from other sockets' );
+        $this->log( 'stop() was called, so will skip reading from remaining readable sockets' );
         break;
       }
     }
@@ -232,6 +237,7 @@ private function handleWritableSockets( $sockets ) {
       $buffer = $this->writeBuffer[ $socketIndex ];
       $bufferLen = strlen( $buffer );
 
+      // Nothing to write
       if ( !$bufferLen )
         continue;
 
@@ -242,7 +248,7 @@ private function handleWritableSockets( $sockets ) {
         throw new \Exception( 'Could not write to socket' );
       }
 
-      $this->log( 'Sent '. $sentBytes .' b to '. $socketIndex );
+      $this->log( 'Sent '. $sentBytes .' bytes to '. $socketIndex );
       $this->writeBuffer[ $socketIndex ] = substr( $buffer, $sentBytes );
     }
   }
@@ -274,9 +280,13 @@ private function getMessagesFromSocket( $socket ) {
     $socketIndex = array_search( $socket, $this->sockets, true );
     $buffer = $this->readBuffer[ $socketIndex ];
 
+    // --------------------------------------------------------------------
     // Populate the MessageBuffer from the socket
+    // --------------------------------------------------------------------
 
     $data = '';
+
+    // "socket_recv() returns the number of bytes received, or FALSE if there was an error"
     $dataLen = socket_recv( $socket, $data, 64 * 1024, MSG_DONTWAIT );
 
     // There was an error
@@ -285,15 +295,21 @@ private function getMessagesFromSocket( $socket ) {
       throw new \Exception( socket_strerror( socket_last_error() ) );
     }
 
-    // Connection was dropped by peer. We expect peers to be dropped only after
-    // $this->stop() has been called, so this is unexpected.
+    // Connection was dropped by peer. Calling code can decide if this is
+    // an error or not by catching the exception (for the Server this is
+    // probably an error, as Workers shouldn't disconnect before the Server,
+    // but for Workers the Server disconnecting is a signal that there's no
+    // more jobs; maybe do this more cleanly later, by sending a "close" message
+    // to the Workers from the Server like HTTP does...).
     if ( $dataLen === 0 )
-      throw new \Exception( 'Socket disconnected unexpectedly' );
+      throw new SocketDisconnectedException( 'Socket disconnected' );
 
-    $this->log( 'Recvd '. $dataLen .' b from '. $socketIndex );
+    $this->log( 'Recvd '. $dataLen .' bytes from '. $socketIndex );
     $this->populateMessageBuffer( $data, $buffer );
 
+    // --------------------------------------------------------------------
     // Get finished Message objects from the MessageBuffer
+    // --------------------------------------------------------------------
 
     $messages = array();
 
@@ -302,7 +318,7 @@ private function getMessagesFromSocket( $socket ) {
       $messages[] = $buffer->message;
       // Check if we received multiple messages' data from the socket
       $overflowBytes = $buffer->bodyLen - $buffer->message->headers[ 'body-len' ];
-      
+
       // We got more bytes than the message consists of, so we got (possibly
       // partially) other messages' data
       if ( $overflowBytes > 0 ) {
@@ -333,11 +349,11 @@ private function populateMessageBuffer( $data, MessageBuffer &$buffer ) {
 
     // We already have the header. Add further data to body.
     if ( $buffer->headerEnd !== false ) {
-      
+
       $dataLen = strlen( $data );
       $buffer->bodyLen += $dataLen;
       $buffer->message->body .= $data;
-      
+
       if ( $buffer->bodyLen >= $buffer->message->headers[ 'body-len' ] )
         $buffer->hasMessage = true;
     }
@@ -370,11 +386,20 @@ private function populateMessageBuffer( $data, MessageBuffer &$buffer ) {
 
   private function disconnect( $socket ) {
 
+    $clientName = array_search( $socket, $this->sockets, true );
+
+    if ( $clientName === false ) {
+      if ( $socket === $this->serverSocket )
+        $clientName = 'SERVER';
+      else
+        $clientName = '[unknown]';
+    }
+
     // Close the connection until the worker client sends us a new result. We
     // silence any errors so that we don't have to test the connection status
     // before we try to close the socket.
     if ( @socket_shutdown( $socket, 2 ) )
-      $this->log( 'Closed connection' );
+      $this->log( 'Closed connection to socket '. $clientName );
   }
 
   private function log( $msg, $socketIndex = 0 ) {
@@ -387,8 +412,7 @@ private function log( $msg, $socketIndex = 0 ) {
       $id = uniqid();
 
     $prefix = ( $this->serverSocket ) ? '[SERVER] ' : '[worker] ';
-    file_put_contents( '/tmp/crusse-job-server.log',
-      $id .' '. $prefix . $msg . PHP_EOL, FILE_APPEND );
+    file_put_contents( '/tmp/crusse-job-server.log', number_format( microtime( true ), 4, '.', '' ) .' '. $id .' '. $prefix . $msg . PHP_EOL, FILE_APPEND );
   }
 }
 

src/Crusse/JobServer/Server.php

@@ -20,12 +20,13 @@ function __construct( $workerCount ) {
 
     if ( $workerCount < 1 )
       throw new \InvalidArgumentException( '$workerCount must be >= 1' );
-    
+
     $this->workerCount = $workerCount;
 
     $tmpDir = sys_get_temp_dir();
     if ( !$tmpDir )
       throw new \Exception( 'Could not find the system temporary files directory' );
+
     $this->serverSocketAddr = $tmpDir .'/php_job_server_'. md5( uniqid( true ) ) .'.sock';
   }
 
@@ -62,12 +63,12 @@ function getOrderedResults() {
 
     $loop = new EventLoop( $this->serverSocketAddr );
     $loop->listen( $this->workerTimeout );
-    $loop->subscribe( array( $this, '_messageCallback' ) );
+    $loop->subscribe( array( $this, '_handleMessageFromWorker' ) );
 
     if ( !$this->workerProcs )
       $this->createWorkerProcs( $this->workerCount );
 
-    $loop->run();
+    $loop->receive();
 
     $results = $this->results;
     ksort( $results );
@@ -85,12 +86,12 @@ function getResults( $jobCallback ) {
 
     $loop = new EventLoop( $this->serverSocketAddr );
     $loop->listen( $this->workerTimeout );
-    $loop->subscribe( array( $this, '_messageCallback' ) );
+    $loop->subscribe( array( $this, '_handleMessageFromWorker' ) );
 
     if ( !$this->workerProcs )
       $this->createWorkerProcs( $this->workerCount );
 
-    $loop->run();
+    $loop->receive();
   }
 
   function setWorkerTimeout( $timeout ) {
@@ -107,16 +108,18 @@ private function createWorkerProcs( $count ) {
       // worker's don't bring down the web server so easily
       $process = new Process( 'exec nice -n 5 php '. dirname( __FILE__ ) .
         '/worker_process.php \''. $this->serverSocketAddr .'\'' );
+
       // We don't need stdout/stderr as we're communicating via sockets
       $process->disableOutput();
       $process->start();
+
       $workers[] = $process;
     }
 
     $this->workerProcs = $workers;
   }
 
-  function _messageCallback( Message $message, EventLoop $loop, $socket ) {
+  function _handleMessageFromWorker( Message $message, EventLoop $loop, $socket ) {
 
     $headers = $message->headers;
 
@@ -134,16 +137,18 @@ function _messageCallback( Message $message, EventLoop $loop, $socket ) {
       $this->results[ $jobNumber ] = ( $this->jobCallback )
         ? true
         : $message->body;
+
       // ...otherwise return the result immediately in the callback
       if ( $this->jobCallback ) {
         call_user_func( $this->jobCallback, $message->body, count( $this->results ),
           count( $this->jobQueue ) );
       }
     }
 
-    // We have all the results; stop the event loop
+    // We have all the results; stop the server (which causes workers to stop
+    // as well)
     if ( count( $this->results ) >= count( $this->jobQueue ) ) {
-      
+
       $loop->stop();
     }
     // Send a job to the worker
@@ -159,7 +164,7 @@ function _messageCallback( Message $message, EventLoop $loop, $socket ) {
       $message->headers[ 'function' ] = $job[ 0 ];
       $message->body = $job[ 1 ];
 
-      // Job was sent to worker, free memory
+      // Job was sent to worker. Free memory.
       $this->jobQueue[ $this->sentJobCount ] = '';
       $this->sentJobCount++;
 

src/Crusse/JobServer/SocketDisconnectedException.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace Crusse\JobServer;
+
+class SocketDisconnectedException extends \Exception {
+
+}
+

src/Crusse/JobServer/Worker.php

@@ -7,6 +7,7 @@
 require_once dirname( __FILE__ ) .'/EventLoop.php';
 require_once dirname( __FILE__ ) .'/Message.php';
 require_once dirname( __FILE__ ) .'/MessageBuffer.php';
+require_once dirname( __FILE__ ) .'/SocketDisconnectedException.php';
 
 class Worker {
 
@@ -26,7 +27,13 @@ function run() {
       $loop->subscribe( array( $this, '_messageCallback' ) );
       $socket = $loop->connect();
       $this->sendMessage( $loop, $socket, 'new-worker' );
-      $loop->run();
+      $loop->receive();
+    }
+    catch ( SocketDisconnectedException $e ) {
+      // This is expected. Currently, if the Server disconnects the socket that
+      // the Worker is communicating on, it's a signal to the Worker that there
+      // are not more jobs to handle. Maybe do this more cleanly later, by
+      // sending a "close" message to the Workers from the Server like HTTP does...
     }
     catch ( \Exception $e ) {
       trigger_error( $e->getMessage() .' in '. $e->getFile() .':'. $e->getLine(), E_USER_WARNING );