Send errors and exceptions from worker processes to the server process, so that it is easier to notice worker errors

Author: Crusse

src/Crusse/JobServer/Server.php

@@ -128,11 +128,15 @@ function _messageCallback( Message $message, EventLoop $loop, $socket ) {
     if ( $headers[ 'cmd' ] === 'job-result' ) {
 
       $jobNumber = $headers[ 'job-num' ];
-      // Only store the results if the client called getOrderedResults()
+
+      if ( strtolower( $headers[ 'job-status' ] ) != 'ok' )
+        throw new \RuntimeException( 'Worker error: ['. $headers[ 'job-status' ] .'] '. $message->body );
+
+      // Only store the result if the client called getOrderedResults()
       $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 ) );

src/Crusse/JobServer/Worker.php

@@ -29,34 +29,56 @@ function run() {
       $loop->run();
     }
     catch ( \Exception $e ) {
-      trigger_error( $e->getMessage() );
+      trigger_error( $e->getMessage() .' in '. $e->getFile() .':'. $e->getLine(), E_USER_WARNING );
     }
   }
 
   function _messageCallback( Message $message, EventLoop $loop, $socket ) {
 
     $headers = $message->headers;
 
-    if ( isset( $headers[ 'includes' ] ) ) {
-      foreach ( array_filter( explode( ',', $headers[ 'includes' ] ) ) as $path )
-        require_once $path;
+    try {
+
+      if ( isset( $headers[ 'includes' ] ) ) {
+        foreach ( array_filter( explode( ',', $headers[ 'includes' ] ) ) as $path )
+          require_once $path;
+      }
+
+      if ( !isset( $headers[ 'function' ] ) ) {
+        $result = 'Request has no \'function\' header';
+        $status = 'invalid_request';
+      }
+      else if ( !is_callable( $headers[ 'function' ] ) ) {
+        $result = '\''. $headers[ 'function' ] .'\' is not callable';
+        $status = 'invalid_request';
+      }
+      else {
+        $result = call_user_func( $headers[ 'function' ], $message->body );
+        $status = 'ok';
+      }
     }
+    catch ( \Exception $e ) {
 
-    if ( !isset( $headers[ 'function' ] ) )
-      throw new \Exception( 'Request has no \'function\' header' );
-    if ( !is_callable( $headers[ 'function' ] ) )
-      throw new \Exception( '\''. $headers[ 'function' ] .'\' is not callable' );
+      $result = $e->getMessage() .' in '. $e->getFile() .':'. $e->getLine();
+      $status = 'exception';
+    }
 
-    $result = call_user_func( $headers[ 'function' ], $message->body );
-    $this->sendMessage( $loop, $socket, 'job-result', $headers[ 'job-num' ], $result );
+    $this->sendMessage( $loop, $socket, 'job-result', $headers[ 'job-num' ], $status, $result );
   }
 
-  private function sendMessage( EventLoop $loop, $socket, $cmd, $jobNumber = null, $body = '' ) {
+  private function sendMessage( EventLoop $loop, $socket, $cmd, $jobNumber = null, $status = 'ok', $body = '' ) {
+
+    if ( !is_string( $body ) )
+      throw new \InvalidArgumentException( 'Worker result must be a string' );
 
     $message = new Message();
     $message->headers[ 'cmd' ] = $cmd;
-    if ( $jobNumber !== null )
+
+    if ( $jobNumber !== null ) {
+      $message->headers[ 'job-status' ] = $status;
       $message->headers[ 'job-num' ] = $jobNumber;
+    }
+
     $message->body = $body;
 
     $loop->send( $socket, $message );

tests/test.php

@@ -27,7 +27,7 @@ function generateString($length) {
 $server = new Crusse\JobServer\Server( 4 );
 $server->addWorkerInclude( __DIR__ .'/functions.php' );
 $server->setWorkerTimeout( 2 );
-for ( $i = 0; $i < 100; $i++ )
+for ( $i = 0; $i < 50; $i++ )
   $server->addJob( 'job_test', 'Job '. $i .': '. generateString( 100 * 250 ) );
 
 echo 'Results with callback:'. PHP_EOL . PHP_EOL;
@@ -41,7 +41,7 @@ function generateString($length) {
 $server = new Crusse\JobServer\Server( 4 );
 $server->addWorkerInclude( __DIR__ .'/functions.php' );
 $server->setWorkerTimeout( 2 );
-for ( $i = 0; $i < 100; $i++ )
+for ( $i = 0; $i < 50; $i++ )
   $server->addJob( 'job_test', 'Job '. $i .': '. generateString( 100 * 250 ) );
 
 $time = microtime( true );
@@ -52,6 +52,19 @@ function generateString($length) {
 $elapsedTotal = ( microtime( true ) - $timeTotal ) * 1000;
 
 echo implode( PHP_EOL, array_keys( $res ) ) . PHP_EOL . PHP_EOL;
+
+echo PHP_EOL .'Worker exception test:'. PHP_EOL . PHP_EOL;
+
+$server = new Crusse\JobServer\Server( 4 );
+$server->setWorkerTimeout( 2 );
+$server->addJob( 'nonexistent_function_123', 'Job '. $i .': '. generateString( 100 * 250 ) );
+try {
+  $res = $server->getOrderedResults();
+}
+catch ( \RuntimeException $e ) {
+  echo 'Successfully caught exception: '. $e->getMessage() . PHP_EOL;
+}
+
 echo 'Finished in '. $elapsed .' ms'. PHP_EOL;
 echo 'Total '. $elapsedTotal .' ms'. PHP_EOL;