Set error_reporting to E_ALL in the worker script

Author: tony-martin-aste

src/Crusse/JobServer/EventLoop.php

@@ -27,11 +27,11 @@ function __construct( $serverSocketAddr ) {
   function __destruct() {
 
     foreach ( $this->sockets as $socket )
-      $this->closeConnection( $socket );
+      $this->disconnect( $socket );
 
     // If we were listening on a socket, remove the socket file
     if ( $this->serverSocket ) {
-      $this->closeConnection( $this->serverSocket );
+      $this->disconnect( $this->serverSocket );
       unlink( $this->serverSocketAddr );
     }
   }
@@ -69,7 +69,7 @@ private function addClientSocket( $socket ) {
     // array_unshift when removing sockets, as it's slow, so we use unset() 
     // instead.
     for ( $i = 0; $i < $socketCount; $i++ ) {
-      if ( !isset( $this->sockets[ $i ] ) ) {
+      if ( empty( $this->sockets[ $i ] ) ) {
         $foundSpot = true;
         $this->sockets[ $i ] = $socket;
         break;
@@ -137,8 +137,10 @@ function run() {
     while ( true ) {
 
       // We have no more sockets to poll, all have disconnected
-      if ( !$this->sockets && !$this->serverSocket )
+      if ( !$this->sockets && !$this->serverSocket ) {
+        $this->log( 'No more sockets to poll, exiting run() loop' );
         break;
+      }
 
       $readables = $this->sockets;
       if ( $this->serverSocket )
@@ -173,16 +175,18 @@ function run() {
       if ( $writables )
         $this->handleWritableSockets( $writables );
 
-      if ( $this->stop )
+      if ( $this->stop ) {
+        $this->log( 'stop() was called, exiting run() loop' );
         break;
+      }
     }
 
     foreach ( $this->sockets as $socket )
-      $this->closeConnection( $socket );
+      $this->disconnect( $socket );
     $this->sockets = array();
 
     if ( $this->serverSocket )
-      $this->closeConnection( $this->serverSocket );
+      $this->disconnect( $this->serverSocket );
   }
 
   function stop() {
@@ -205,16 +209,18 @@ private function handleReadableSockets( $sockets ) {
       if ( !$messages )
         continue;
 
-      //$this->log( 'Buffer had '. count( $messages ) .' messages' );
+      $this->log( 'Buffer had '. count( $messages ) .' messages' );
 
       foreach ( $messages as $message ) {
         foreach ( $this->callbacks as $callback ) {
           call_user_func( $callback, $message, $this, $socket );
         }
       }
 
-      if ( $this->stop )
+      if ( $this->stop ) {
+        $this->log( 'stop() was called, so will not read from other sockets' );
         break;
+      }
     }
   }
 
@@ -236,7 +242,7 @@ private function handleWritableSockets( $sockets ) {
         throw new \Exception( 'Could not write to socket' );
       }
 
-      //$this->log( 'Sent '. $sentBytes .' b to '. $socketIndex );
+      $this->log( 'Sent '. $sentBytes .' b to '. $socketIndex );
       $this->writeBuffer[ $socketIndex ] = substr( $buffer, $sentBytes );
     }
   }
@@ -251,7 +257,7 @@ private function acceptClient() {
     }
 
     $this->addClientSocket( $socket );
-    //$this->log( 'Accepted client '. ( count( $this->sockets ) - 1 ) );
+    $this->log( 'Accepted client '. ( count( $this->sockets ) - 1 ) );
 
     return $socket;
   }
@@ -282,12 +288,12 @@ private function getMessagesFromSocket( $socket ) {
     // Connection was dropped by peer
     if ( $dataLen === 0 ) {
       // Don't read/write from/to this socket in the future
-      $this->closeConnection( $this->sockets[ $socketIndex ] );
+      $this->disconnect( $this->sockets[ $socketIndex ] );
       unset( $this->sockets[ $socketIndex ] );
       return array();
     }
 
-    //$this->log( 'Recvd '. $dataLen .' b from '. $socketIndex );
+    $this->log( 'Recvd '. $dataLen .' b from '. $socketIndex );
     $this->populateMessageBuffer( $data, $buffer );
 
     // Get finished Message objects from the MessageBuffer
@@ -304,7 +310,7 @@ private function getMessagesFromSocket( $socket ) {
       // partially) other messages' data
       if ( $overflowBytes > 0 ) {
 
-        //$this->log( 'Recvd multiple messages from socket (overflow: '. $overflowBytes .' b)' );
+        $this->log( 'Recvd multiple messages from socket (overflow: '. $overflowBytes .' b)' );
 
         $overflow = substr( $buffer->message->body, -$overflowBytes );
         $buffer->message->body .= substr( $buffer->message->body, 0, -$overflowBytes );
@@ -365,7 +371,7 @@ private function populateMessageBuffer( $data, MessageBuffer &$buffer ) {
     }
   }
 
-  private function closeConnection( $socket ) {
+  private function disconnect( $socket ) {
 
     // 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

src/Crusse/JobServer/worker_process.php

@@ -3,6 +3,8 @@
 if ( php_sapi_name() !== 'cli' )
   exit( 1 );
 
+error_reporting( E_ALL );
+
 require_once dirname( __FILE__ ) .'/Worker.php';
 
 $worker = new \Crusse\JobServer\Worker( $argv[ 1 ] );

tests/functions.php

@@ -1,6 +1,9 @@
 <?php
 
 function job_test( $message ) {
+
+  usleep( rand( 50000, 500000 ) );
+
   return strtoupper( $message );
 }
 

tests/test.php

@@ -1,5 +1,10 @@
 <?php
 
+ini_set('display_errors', 1);
+ini_set('display_startup_errors', 1);
+ini_set('error_log', dirname(__FILE__) .'/php_errors.log');
+error_reporting(E_ALL);
+
 require_once __DIR__ .'/../vendor/autoload.php';
 
 function generateString($length) {
@@ -19,10 +24,10 @@ function generateString($length) {
 
 $timeTotal = microtime( true );
 
-$server = new Crusse\JobServer\Server( 4 );
+$server = new Crusse\JobServer\Server( 6 );
 $server->addWorkerInclude( __DIR__ .'/functions.php' );
 $server->setWorkerTimeout( 2 );
-for ( $i = 0; $i < 50; $i++ )
+for ( $i = 0; $i < 1000; $i++ )
   $server->addJob( 'job_test', 'Job '. $i .': '. generateString( 100 * 250 ) );
 
 echo 'Results with callback:'. PHP_EOL . PHP_EOL;
@@ -33,10 +38,10 @@ function generateString($length) {
 
 echo PHP_EOL .'Ordered results:'. PHP_EOL . PHP_EOL;
 
-$server = new Crusse\JobServer\Server( 4 );
+$server = new Crusse\JobServer\Server( 6 );
 $server->addWorkerInclude( __DIR__ .'/functions.php' );
 $server->setWorkerTimeout( 2 );
-for ( $i = 0; $i < 50; $i++ )
+for ( $i = 0; $i < 1000; $i++ )
   $server->addJob( 'job_test', 'Job '. $i .': '. generateString( 100 * 250 ) );
 
 $time = microtime( true );