When getting results via a callback function, dont store the results, so that we use as little memory as possible

Author: tony-martin-aste

src/Crusse/JobServer/Server.php

@@ -52,12 +52,9 @@ function addWorkerInclude( $phpFilePath ) {
     $this->workerIncludes[] = $phpFilePath;
   }
 
-  function getResults( $jobCallback = null ) {
+  function getOrderedResults() {
 
-    if ( $jobCallback && !is_callable( $jobCallback ) )
-      throw new \InvalidArgumentException( '$jobCallback is not callable' );
-
-    $this->jobCallback = $jobCallback;
+    $this->jobCallback = null;
 
     $loop = new EventLoop( $this->serverSocketAddr );
     $loop->listen( $this->workerTimeout );
@@ -75,6 +72,23 @@ function getResults( $jobCallback = null ) {
     return $results;
   }
 
+  function getResults( $jobCallback ) {
+
+    if ( !is_callable( $jobCallback ) )
+      throw new \InvalidArgumentException( '$jobCallback is not callable' );
+
+    $this->jobCallback = $jobCallback;
+
+    $loop = new EventLoop( $this->serverSocketAddr );
+    $loop->listen( $this->workerTimeout );
+    $loop->subscribe( array( $this, '_messageCallback' ) );
+
+    if ( !$this->workerProcs )
+      $this->createWorkerProcs( $this->workerCount );
+
+    $loop->run();
+  }
+
   // FIXME: workerTimeout isn't working yet. A sleep()ing worker does _not_ 
   // cause a timeout. Check why.
   function setWorkerTimeout( $timeout ) {
@@ -110,7 +124,10 @@ function _messageCallback( Message $message, EventLoop $loop, $socket ) {
     if ( $headers[ 'cmd' ] === 'job-result' ) {
 
       $jobNumber = $headers[ 'job-num' ];
-      $this->results[ $jobNumber ] = $message->body;
+      // Only store the results if the client called getOrderedResults()
+      $this->results[ $jobNumber ] = ( $this->jobCallback )
+        ? true
+        : $message->body;
 
       if ( $this->jobCallback ) {
         call_user_func( $this->jobCallback, $message->body, count( $this->results ),

tests/test.php

@@ -22,18 +22,31 @@ function generateString($length) {
 $server = new Crusse\JobServer\Server( 4 );
 $server->addWorkerInclude( __DIR__ .'/functions.php' );
 $server->setWorkerTimeout( 2 );
-
 for ( $i = 0; $i < 50; $i++ )
   $server->addJob( 'job_test', 'Job '. $i .': '. generateString( 50 ) );
 
-$time = microtime( true );
-$res = $server->getResults( function( $result, $jobNumber, $total ) {
+echo 'Results with callback:'. PHP_EOL . PHP_EOL;
+
+$server->getResults( function( $result, $jobNumber, $total ) {
   echo 'Job '. $jobNumber .'/'. $total .': '. $result . PHP_EOL;
 } );
+
+echo PHP_EOL .'Ordered results:'. PHP_EOL . PHP_EOL;
+
+$server = new Crusse\JobServer\Server( 4 );
+$server->addWorkerInclude( __DIR__ .'/functions.php' );
+$server->setWorkerTimeout( 2 );
+for ( $i = 0; $i < 50; $i++ )
+  $server->addJob( 'job_test', 'Job '. $i .': '. generateString( 50 ) );
+
+$time = microtime( true );
+
+$res = $server->getOrderedResults();
+
 $elapsed = ( microtime( true ) - $time ) * 1000;
 $elapsedTotal = ( microtime( true ) - $timeTotal ) * 1000;
 
-//var_dump( $res );
+echo implode( PHP_EOL, $res ) . PHP_EOL . PHP_EOL;
 echo 'Finished in '. $elapsed .' ms'. PHP_EOL;
 echo 'Total '. $elapsedTotal .' ms'. PHP_EOL;