From 9e72ea6689c18dbce1f6bbe6980ed3afda045095 Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Sun, 23 Nov 2014 00:55:01 +0100
Subject: [PATCH 01/10] Initial implementation

---
 src/main/php/util/data/Distribution.class.php | 51 +++++++++++++++++++
 src/main/php/util/data/Sequence.class.php     | 10 ++++
 .../data/unittest/DistributionTest.class.php  | 39 ++++++++++++++
 3 files changed, 100 insertions(+)
 create mode 100755 src/main/php/util/data/Distribution.class.php
 create mode 100755 src/test/php/util/data/unittest/DistributionTest.class.php

diff --git a/src/main/php/util/data/Distribution.class.php b/src/main/php/util/data/Distribution.class.php
new file mode 100755
index 0000000..959b9b6
--- /dev/null
+++ b/src/main/php/util/data/Distribution.class.php
@@ -0,0 +1,51 @@
+<?php namespace util\data;
+
+/**
+ * 
+ */
+class Distribution extends \lang\Object implements \Iterator {
+  protected $it;
+  protected $workers;
+  protected $worker;
+
+  /**
+   * Creates a new Generator instance
+   *
+   * @param  php.Iterator $it
+   * @param  var[] $workers
+   */
+  public function __construct(\Iterator $it, array $workers) {
+    $this->it= $it;
+    $this->workers= $workers;
+  }
+
+  /** @return void */
+  public function rewind() {
+    $this->worker= 0;
+    $this->it->rewind();
+  }
+
+  /** @return var */
+  public function current() {
+    return $this->workers[$this->worker]->__invoke($this->it->current());
+  }
+
+  /** @return int */
+  public function key() {
+    return $this->it->key();
+  }
+
+  /** @return void */
+  public function next() {
+    if (++$this->worker >= sizeof($this->workers)) {
+      $this->worker= 0;
+    }
+
+    $this->it->next();
+  }
+
+  /** @return bool */
+  public function valid() {
+    return $this->it->valid();
+  }
+}
\ No newline at end of file
diff --git a/src/main/php/util/data/Sequence.class.php b/src/main/php/util/data/Sequence.class.php
index 1398641..51aa473 100755
--- a/src/main/php/util/data/Sequence.class.php
+++ b/src/main/php/util/data/Sequence.class.php
@@ -391,6 +391,16 @@ public function peek($action) {
     }));
   }
 
+  /**
+   * Returns a new stream which maps the given function to each element
+   *
+   * @param  var[] $workers
+   * @return self
+   */
+  public function distribute(array $workers) {
+    return new self(new Distribution($this->getIterator(), $workers));
+  }
+
   /**
    * Returns a new stream which counts the number of elements as iteration
    * proceeeds. A short form of `peek()` with a function incrementing a local
diff --git a/src/test/php/util/data/unittest/DistributionTest.class.php b/src/test/php/util/data/unittest/DistributionTest.class.php
new file mode 100755
index 0000000..488f361
--- /dev/null
+++ b/src/test/php/util/data/unittest/DistributionTest.class.php
@@ -0,0 +1,39 @@
+<?php namespace util\data\unittest;
+
+use util\data\Sequence;
+
+class DistributionTest extends AbstractSequenceTest {
+
+  #[@test]
+  public function one() {
+    $processed= [0 => []];
+    $workers= [
+      function($e) use(&$processed) { $processed[0][]= $e; }
+    ];
+    Sequence::of([1, 2, 3, 4, 5, 6, 7, 8])->distribute($workers)->count();
+    $this->assertEquals([0 => [1, 2, 3, 4, 5, 6, 7, 8]], $processed);
+  }
+
+  #[@test]
+  public function two() {
+    $processed= [0 => [], 1 => []];
+    $workers= [
+      function($e) use(&$processed) { $processed[0][]= $e; },
+      function($e) use(&$processed) { $processed[1][]= $e; }
+    ];
+    Sequence::of([1, 2, 3, 4, 5, 6, 7, 8])->distribute($workers)->count();
+    $this->assertEquals([0 => [1, 3, 5, 7], 1 => [2, 4, 6, 8]], $processed);
+  }
+
+  #[@test]
+  public function three() {
+    $processed= [0 => [], 1 => [], 2 => []];
+    $workers= [
+      function($e) use(&$processed) { $processed[0][]= $e; },
+      function($e) use(&$processed) { $processed[1][]= $e; },
+      function($e) use(&$processed) { $processed[2][]= $e; }
+    ];
+    Sequence::of([1, 2, 3, 4, 5, 6, 7, 8])->distribute($workers)->count();
+    $this->assertEquals([0 => [1, 4, 7], 1 => [2, 5, 8], 2 => [3, 6]], $processed);
+  }
+}
\ No newline at end of file

From 0ba216d4aedbb9aaad904744e57815f0bf984728 Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Sun, 23 Nov 2014 00:55:46 +0100
Subject: [PATCH 02/10] Initial implementation

---
 ...ibutionTest.class.php => SequenceDistributionTest.class.php} | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)
 rename src/test/php/util/data/unittest/{DistributionTest.class.php => SequenceDistributionTest.class.php} (95%)

diff --git a/src/test/php/util/data/unittest/DistributionTest.class.php b/src/test/php/util/data/unittest/SequenceDistributionTest.class.php
similarity index 95%
rename from src/test/php/util/data/unittest/DistributionTest.class.php
rename to src/test/php/util/data/unittest/SequenceDistributionTest.class.php
index 488f361..b170ee4 100755
--- a/src/test/php/util/data/unittest/DistributionTest.class.php
+++ b/src/test/php/util/data/unittest/SequenceDistributionTest.class.php
@@ -2,7 +2,7 @@
 
 use util\data\Sequence;
 
-class DistributionTest extends AbstractSequenceTest {
+class SequenceDistributionTest extends AbstractSequenceTest {
 
   #[@test]
   public function one() {

From d899f5e3045f31b32b852a0ca8ce294b6418e059 Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Sun, 23 Nov 2014 14:40:48 +0100
Subject: [PATCH 03/10] First implementation with local worker processes

---
 src/main/php/util/data/Distribution.class.php | 22 +++----
 .../util/data/LocalWorkerProcess.class.php    | 53 ++++++++++++++++
 src/main/php/util/data/Sequence.class.php     |  4 +-
 .../php/util/data/WorkerFunctions.class.php   | 29 +++++++++
 .../php/util/data/WorkerProcesses.class.php   | 60 +++++++++++++++++++
 src/main/php/util/data/Workers.class.php      | 11 ++++
 .../SequenceDistributionTest.class.php        | 38 +++++++++---
 .../php/util/data/unittest/Worker.class.php   | 52 ++++++++++++++++
 8 files changed, 248 insertions(+), 21 deletions(-)
 create mode 100755 src/main/php/util/data/LocalWorkerProcess.class.php
 create mode 100755 src/main/php/util/data/WorkerFunctions.class.php
 create mode 100755 src/main/php/util/data/WorkerProcesses.class.php
 create mode 100755 src/main/php/util/data/Workers.class.php
 create mode 100755 src/test/php/util/data/unittest/Worker.class.php

diff --git a/src/main/php/util/data/Distribution.class.php b/src/main/php/util/data/Distribution.class.php
index 959b9b6..c715423 100755
--- a/src/main/php/util/data/Distribution.class.php
+++ b/src/main/php/util/data/Distribution.class.php
@@ -6,42 +6,44 @@
 class Distribution extends \lang\Object implements \Iterator {
   protected $it;
   protected $workers;
-  protected $worker;
+  protected $inv;
 
   /**
    * Creates a new Generator instance
    *
    * @param  php.Iterator $it
-   * @param  var[] $workers
+   * @param  util.data.Workers $workers
    */
-  public function __construct(\Iterator $it, array $workers) {
+  public function __construct(\Iterator $it, Workers $workers) {
     $this->it= $it;
     $this->workers= $workers;
+    $this->inv= 0;
   }
 
   /** @return void */
   public function rewind() {
-    $this->worker= 0;
     $this->it->rewind();
+    if ($this->valid()) {
+      $this->workers->enqueue($this->it->current());
+    }
   }
 
   /** @return var */
   public function current() {
-    return $this->workers[$this->worker]->__invoke($this->it->current());
+    return $this->workers->dequeue();
   }
 
   /** @return int */
   public function key() {
-    return $this->it->key();
+    return $this->inv++;
   }
 
   /** @return void */
   public function next() {
-    if (++$this->worker >= sizeof($this->workers)) {
-      $this->worker= 0;
-    }
-
     $this->it->next();
+    if ($this->valid()) {
+      $this->workers->enqueue($this->it->current());
+    }
   }
 
   /** @return bool */
diff --git a/src/main/php/util/data/LocalWorkerProcess.class.php b/src/main/php/util/data/LocalWorkerProcess.class.php
new file mode 100755
index 0000000..0f0cb44
--- /dev/null
+++ b/src/main/php/util/data/LocalWorkerProcess.class.php
@@ -0,0 +1,53 @@
+<?php namespace util\data;
+
+use peer\Socket;
+use lang\Runtime;
+use lang\IllegalStateException;
+use io\IOException;
+
+/**
+ * 
+ */
+class LocalWorkerProcess extends \lang\Object {
+
+  public function __construct($class, $args= []) {
+    $this->proc= Runtime::getInstance()->newInstance(null, 'class', $class, $args);
+    $line= $this->proc->out->readLine();
+    if ('+' === $line{0}) {
+      sscanf($line, '+ %s:%d', $host, $port);
+      $this->comm= new Socket($host, $port);
+      $this->comm->connect();
+    } else {
+      $this->proc->close();
+      throw new IllegalStateException('Cannot initiate communication with worker: '.$line);
+    }
+  }
+
+  public function pass($element) {
+    $this->comm->write(serialize($element)."\n");
+  }
+
+  public function result() {
+    return unserialize($this->comm->readLine());
+  }
+
+  public function handle() {
+    return $this->comm->getHandle();
+  }
+
+  public function shutdown() {
+    if (-1 === $this->proc->exitValue()) {
+      try {
+        $this->comm->write("SHUTDOWN\n");
+        $this->comm->close();
+      } catch (IOException $ignored) { }
+
+      $this->proc->close();
+    }
+    return $this->proc->exitValue();
+  }
+
+  public function __destruct() {
+    $this->shutdown();
+  }
+}
diff --git a/src/main/php/util/data/Sequence.class.php b/src/main/php/util/data/Sequence.class.php
index 51aa473..f7a2692 100755
--- a/src/main/php/util/data/Sequence.class.php
+++ b/src/main/php/util/data/Sequence.class.php
@@ -394,10 +394,10 @@ public function peek($action) {
   /**
    * Returns a new stream which maps the given function to each element
    *
-   * @param  var[] $workers
+   * @param  util.data.Workers $workers
    * @return self
    */
-  public function distribute(array $workers) {
+  public function distribute(Workers $workers) {
     return new self(new Distribution($this->getIterator(), $workers));
   }
 
diff --git a/src/main/php/util/data/WorkerFunctions.class.php b/src/main/php/util/data/WorkerFunctions.class.php
new file mode 100755
index 0000000..2f72c29
--- /dev/null
+++ b/src/main/php/util/data/WorkerFunctions.class.php
@@ -0,0 +1,29 @@
+<?php namespace util\data;
+
+/**
+ * 
+ */
+class WorkerFunctions extends \lang\Object implements Workers {
+  protected $functions, $offset;
+
+  /**
+   * Creates a new WorkerFunctions instance
+   *
+   * @param  php.Closure[] $functions
+   */
+  public function __construct(array $functions) {
+    $this->functions= $functions;
+    $this->offset= 0;
+  }
+
+  public function enqueue($element) {
+    $this->result= $this->functions[$this->offset]->__invoke($element);
+    if (++$this->offset >= sizeof($this->functions)) {
+      $this->offset= 0;
+    }
+  }
+
+  public function dequeue() {
+    return $this->result;
+  }
+}
\ No newline at end of file
diff --git a/src/main/php/util/data/WorkerProcesses.class.php b/src/main/php/util/data/WorkerProcesses.class.php
new file mode 100755
index 0000000..c118a03
--- /dev/null
+++ b/src/main/php/util/data/WorkerProcesses.class.php
@@ -0,0 +1,60 @@
+<?php namespace util\data;
+
+use io\IOException;
+
+/**
+ * 
+ */
+class WorkerProcesses extends \lang\Object implements Workers {
+  protected $processes, $timeout, $offset, $waitHandles;
+
+  /**
+   * Creates a new WorkerProcesses instance
+   *
+   * @param  util.data.WorkerProcess[] $processes
+   * @param  double $timeout
+   */
+  public function __construct(array $processes, $timeout= null) {
+    $this->processes= $processes;
+    $this->timeout= $timeout;
+    $this->offset= 0;
+    $this->waitHandles= [];
+    foreach ($this->processes as $i => $process) {
+      $this->waitHandles[$i]= $process->handle();
+    }
+  }
+
+  public function enqueue($element) {
+    $this->processes[$this->offset]->pass($element);
+    if (++$this->offset >= sizeof($this->processes)) {
+      $this->offset= 0;
+    }
+  }
+
+  /**
+   * Wait for the first worker process to become ready, and return its result
+   *
+   * @param  var[] $r
+   * @param  double $timeout
+   * @param  util.data.WorkerProcess
+   */
+  protected function waitFor($r, $timeout) {
+    if (null === $timeout) {
+      $tv_sec= $tv_usec= null;
+    } else {
+      $tv_sec= intval(floor($timeout));
+      $tv_usec= intval(($timeout - floor($timeout)) * 1000000);
+    }
+
+    $w= $e= null;
+    if (false === stream_select($r, $w, $e, $tv_sec, $tv_usec) || !$r) {
+      throw new IOException('No results present'.($this->timeout ? ' after '.$this->timeout.' seconds' : ''));
+    }
+
+    return $this->processes[key($r)];
+  }
+
+  public function dequeue() {
+    return $this->waitFor($this->waitHandles, $this->timeout)->result();
+  }
+}
\ No newline at end of file
diff --git a/src/main/php/util/data/Workers.class.php b/src/main/php/util/data/Workers.class.php
new file mode 100755
index 0000000..4473be6
--- /dev/null
+++ b/src/main/php/util/data/Workers.class.php
@@ -0,0 +1,11 @@
+<?php namespace util\data;
+
+/**
+ * 
+ */
+interface Workers {
+
+  public function enqueue($element);
+
+  public function dequeue();
+}
\ No newline at end of file
diff --git a/src/test/php/util/data/unittest/SequenceDistributionTest.class.php b/src/test/php/util/data/unittest/SequenceDistributionTest.class.php
index b170ee4..914d572 100755
--- a/src/test/php/util/data/unittest/SequenceDistributionTest.class.php
+++ b/src/test/php/util/data/unittest/SequenceDistributionTest.class.php
@@ -1,39 +1,59 @@
 <?php namespace util\data\unittest;
 
 use util\data\Sequence;
+use util\data\WorkerFunctions;
+use util\data\WorkerProcesses;
+use util\data\LocalWorkerProcess;
+use lang\Runtime;
 
 class SequenceDistributionTest extends AbstractSequenceTest {
 
   #[@test]
-  public function one() {
+  public function one_function() {
     $processed= [0 => []];
-    $workers= [
+    $workers= new WorkerFunctions([
       function($e) use(&$processed) { $processed[0][]= $e; }
-    ];
+    ]);
     Sequence::of([1, 2, 3, 4, 5, 6, 7, 8])->distribute($workers)->count();
     $this->assertEquals([0 => [1, 2, 3, 4, 5, 6, 7, 8]], $processed);
   }
 
   #[@test]
-  public function two() {
+  public function two_functions() {
     $processed= [0 => [], 1 => []];
-    $workers= [
+    $workers= new WorkerFunctions([
       function($e) use(&$processed) { $processed[0][]= $e; },
       function($e) use(&$processed) { $processed[1][]= $e; }
-    ];
+    ]);
     Sequence::of([1, 2, 3, 4, 5, 6, 7, 8])->distribute($workers)->count();
     $this->assertEquals([0 => [1, 3, 5, 7], 1 => [2, 4, 6, 8]], $processed);
   }
 
   #[@test]
-  public function three() {
+  public function three_functions() {
     $processed= [0 => [], 1 => [], 2 => []];
-    $workers= [
+    $workers= new WorkerFunctions([
       function($e) use(&$processed) { $processed[0][]= $e; },
       function($e) use(&$processed) { $processed[1][]= $e; },
       function($e) use(&$processed) { $processed[2][]= $e; }
-    ];
+    ]);
     Sequence::of([1, 2, 3, 4, 5, 6, 7, 8])->distribute($workers)->count();
     $this->assertEquals([0 => [1, 4, 7], 1 => [2, 5, 8], 2 => [3, 6]], $processed);
   }
+
+  #[@test]
+  public function processes() {
+    $workers= [
+      new LocalWorkerProcess('util.data.unittest.Worker', [3]),
+      new LocalWorkerProcess('util.data.unittest.Worker', [10])
+    ];
+
+    $results= Sequence::of([1, 2, 3, 4, 5, 6, 7, 8])->distribute(new WorkerProcesses($workers))->toArray();
+
+    foreach ($workers as $worker) {
+      $worker->shutdown();
+    }
+
+    $this->assertEquals([3, 20, 9, 40, 15, 60, 21, 80], $results);
+  }
 }
\ No newline at end of file
diff --git a/src/test/php/util/data/unittest/Worker.class.php b/src/test/php/util/data/unittest/Worker.class.php
new file mode 100755
index 0000000..c54012c
--- /dev/null
+++ b/src/test/php/util/data/unittest/Worker.class.php
@@ -0,0 +1,52 @@
+<?php namespace util\data\unittest;
+
+use util\cmd\Console;
+use peer\server\Server;
+use lang\Throwable;
+
+class Worker extends \lang\Object {
+
+  /**
+   * Multiplies numbers given on STDIN with a factor given as argument.
+   * Empty input ends command.
+   *
+   * @param  string[] $args
+   * @return int
+   */
+  public static function main($args) {
+    $server= new Server('127.0.0.1', 0);
+    $server->setProtocol(newinstance('peer.server.ServerProtocol', [$args ? (int)$args[0] : 2], '{
+      public function __construct($factor) {
+        $this->factor= $factor;
+      }
+
+      public function initialize() { }
+
+      public function handleConnect($socket) { }
+
+      public function handleDisconnect($socket) { }
+
+      public function handleError($socket, $e) { }
+
+      public function handleData($socket) {
+        $in= $socket->readLine();
+        if ("SHUTDOWN" === $in) {
+          $this->server->terminate= true;
+        } else if ($in) {
+          $socket->write(serialize(unserialize($in) * $this->factor)."\n");
+        }
+      }  
+    }'));
+
+    try {
+      $server->init();
+      Console::writeLinef('+ %s:%d', $server->socket->host, $server->socket->port);
+      $server->service();
+      Console::writeLine('+ Shutdown');
+      return 0;
+    } catch (Throwable $e) {
+      Console::writeLine('- ', $e->getMessage());
+      return 1;
+    }
+  }
+}
\ No newline at end of file

From 04357b59b19822db92eeb6c6402a2d89f76f2ae4 Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Sun, 23 Nov 2014 14:43:53 +0100
Subject: [PATCH 04/10] Make test robust against changing order

---
 .../php/util/data/unittest/SequenceDistributionTest.class.php | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/src/test/php/util/data/unittest/SequenceDistributionTest.class.php b/src/test/php/util/data/unittest/SequenceDistributionTest.class.php
index 914d572..eb69b1f 100755
--- a/src/test/php/util/data/unittest/SequenceDistributionTest.class.php
+++ b/src/test/php/util/data/unittest/SequenceDistributionTest.class.php
@@ -54,6 +54,8 @@ public function processes() {
       $worker->shutdown();
     }
 
-    $this->assertEquals([3, 20, 9, 40, 15, 60, 21, 80], $results);
+    // The order in which results are returned cannot be guaranteed!
+    sort($results);
+    $this->assertEquals([3, 9, 15, 20, 21, 40, 60, 80], $results);
   }
 }
\ No newline at end of file

From efbb39a7b73fa68cfa72b0a203fd7e4326d28318 Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Sun, 23 Nov 2014 15:37:46 +0100
Subject: [PATCH 05/10] Actually work in parallel

---
 src/main/php/util/data/Distribution.class.php  | 18 ++++++++++++------
 .../php/util/data/LocalWorkerProcess.class.php |  5 +++++
 .../php/util/data/WorkerFunctions.class.php    |  5 +++++
 .../php/util/data/WorkerProcesses.class.php    | 15 +++++++++++++--
 src/main/php/util/data/Workers.class.php       |  2 ++
 5 files changed, 37 insertions(+), 8 deletions(-)

diff --git a/src/main/php/util/data/Distribution.class.php b/src/main/php/util/data/Distribution.class.php
index c715423..f012cab 100755
--- a/src/main/php/util/data/Distribution.class.php
+++ b/src/main/php/util/data/Distribution.class.php
@@ -20,12 +20,16 @@ public function __construct(\Iterator $it, Workers $workers) {
     $this->inv= 0;
   }
 
+  protected function enqueue() {
+    while ($this->valid() && $this->workers->enqueue($this->it->current())) {
+      $this->it->next();
+    }
+  }
+
   /** @return void */
   public function rewind() {
     $this->it->rewind();
-    if ($this->valid()) {
-      $this->workers->enqueue($this->it->current());
-    }
+    $this->enqueue();
   }
 
   /** @return var */
@@ -40,9 +44,11 @@ public function key() {
 
   /** @return void */
   public function next() {
-    $this->it->next();
-    if ($this->valid()) {
-      $this->workers->enqueue($this->it->current());
+    if ($this->workers->pending()) {
+      // Nothing
+    } else {
+      $this->it->next();
+      $this->enqueue();
     }
   }
 
diff --git a/src/main/php/util/data/LocalWorkerProcess.class.php b/src/main/php/util/data/LocalWorkerProcess.class.php
index 0f0cb44..47e92df 100755
--- a/src/main/php/util/data/LocalWorkerProcess.class.php
+++ b/src/main/php/util/data/LocalWorkerProcess.class.php
@@ -11,6 +11,7 @@
 class LocalWorkerProcess extends \lang\Object {
 
   public function __construct($class, $args= []) {
+    $this->cmd= $class.' ['.implode(', ', $args).']';
     $this->proc= Runtime::getInstance()->newInstance(null, 'class', $class, $args);
     $line= $this->proc->out->readLine();
     if ('+' === $line{0}) {
@@ -50,4 +51,8 @@ public function shutdown() {
   public function __destruct() {
     $this->shutdown();
   }
+
+  public function toString() {
+    return $this->getClassName().'(pid= '.$this->proc->getProcessId().', cmd= '.$this->cmd.')';
+  }
 }
diff --git a/src/main/php/util/data/WorkerFunctions.class.php b/src/main/php/util/data/WorkerFunctions.class.php
index 2f72c29..3fac6de 100755
--- a/src/main/php/util/data/WorkerFunctions.class.php
+++ b/src/main/php/util/data/WorkerFunctions.class.php
@@ -21,6 +21,11 @@ public function enqueue($element) {
     if (++$this->offset >= sizeof($this->functions)) {
       $this->offset= 0;
     }
+    return false;
+  }
+
+  public function pending() {
+    return false;
   }
 
   public function dequeue() {
diff --git a/src/main/php/util/data/WorkerProcesses.class.php b/src/main/php/util/data/WorkerProcesses.class.php
index c118a03..a6348d6 100755
--- a/src/main/php/util/data/WorkerProcesses.class.php
+++ b/src/main/php/util/data/WorkerProcesses.class.php
@@ -18,6 +18,7 @@ public function __construct(array $processes, $timeout= null) {
     $this->processes= $processes;
     $this->timeout= $timeout;
     $this->offset= 0;
+    $this->pending= [];
     $this->waitHandles= [];
     foreach ($this->processes as $i => $process) {
       $this->waitHandles[$i]= $process->handle();
@@ -26,9 +27,17 @@ public function __construct(array $processes, $timeout= null) {
 
   public function enqueue($element) {
     $this->processes[$this->offset]->pass($element);
+    $this->pending[$this->offset]= true;
+
     if (++$this->offset >= sizeof($this->processes)) {
       $this->offset= 0;
+      return false;
     }
+    return true;
+  }
+
+  public function pending() {
+    return !empty($this->pending);
   }
 
   /**
@@ -47,11 +56,13 @@ protected function waitFor($r, $timeout) {
     }
 
     $w= $e= null;
-    if (false === stream_select($r, $w, $e, $tv_sec, $tv_usec) || !$r) {
+    if (false === stream_select($r, $w, $e, $tv_sec, $tv_usec) || empty($r)) {
       throw new IOException('No results present'.($this->timeout ? ' after '.$this->timeout.' seconds' : ''));
     }
 
-    return $this->processes[key($r)];
+    $offset= key($r);
+    unset($this->pending[$offset]);
+    return $this->processes[$offset];
   }
 
   public function dequeue() {
diff --git a/src/main/php/util/data/Workers.class.php b/src/main/php/util/data/Workers.class.php
index 4473be6..57808d7 100755
--- a/src/main/php/util/data/Workers.class.php
+++ b/src/main/php/util/data/Workers.class.php
@@ -7,5 +7,7 @@ interface Workers {
 
   public function enqueue($element);
 
+  public function pending();
+
   public function dequeue();
 }
\ No newline at end of file

From eb14564fcc2af8769b3f7c30b795cf369a4e2898 Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Mon, 24 Nov 2014 00:44:34 +0100
Subject: [PATCH 06/10] Add tests for LocalWorkerProcess

---
 .../util/data/LocalWorkerProcess.class.php    | 65 +++++++++++++++++--
 src/main/php/util/data/Sequence.class.php     |  1 -
 .../unittest/LocalWorkerProcessTest.class.php | 49 ++++++++++++++
 .../php/util/data/unittest/Worker.class.php   |  5 ++
 4 files changed, 113 insertions(+), 7 deletions(-)
 create mode 100755 src/test/php/util/data/unittest/LocalWorkerProcessTest.class.php

diff --git a/src/main/php/util/data/LocalWorkerProcess.class.php b/src/main/php/util/data/LocalWorkerProcess.class.php
index 47e92df..fc3671a 100755
--- a/src/main/php/util/data/LocalWorkerProcess.class.php
+++ b/src/main/php/util/data/LocalWorkerProcess.class.php
@@ -4,15 +4,37 @@
 use lang\Runtime;
 use lang\IllegalStateException;
 use io\IOException;
+use util\collections\Queue;
 
 /**
- * 
+ * A worker process that runs on this machine.
+ *
+ * @test  xyp://util.data.unittest.LocalWorkerProcessTest
  */
 class LocalWorkerProcess extends \lang\Object {
+  protected $queue, $pending, $cmd, $proc, $comm;
 
+  /**
+   * Creates a new locally running worker process
+   *
+   * @param  string $class
+   * @param  string[] $args
+   */
   public function __construct($class, $args= []) {
+    $this->queue= new Queue();
+    $this->pending= false;
     $this->cmd= $class.' ['.implode(', ', $args).']';
     $this->proc= Runtime::getInstance()->newInstance(null, 'class', $class, $args);
+    $this->connect();
+  }
+
+  /**
+   * Initiates communication with worker
+   *
+   * @return void
+   * @throws lang.IllegalStateException
+   */
+  protected function connect() {
     $line= $this->proc->out->readLine();
     if ('+' === $line{0}) {
       sscanf($line, '+ %s:%d', $host, $port);
@@ -24,18 +46,49 @@ public function __construct($class, $args= []) {
     }
   }
 
+  public function handle() { return $this->comm->getHandle(); }
+
+  public function pending() { return $this->pending; }
+
+  /**
+   * Pass in an element for processing
+   *
+   * @param  var $element
+   */
   public function pass($element) {
-    $this->comm->write(serialize($element)."\n");
+    if ($this->pending) {
+      $this->queue->put($element);
+    } else {
+      $this->comm->write(serialize($element)."\n");
+      $this->pending= true;
+    }
   }
 
+  /**
+   * Returns a processing result
+   *
+   * @return var
+   * @throws lang.IllegalStateException
+   */
   public function result() {
-    return unserialize($this->comm->readLine());
-  }
+    if (!$this->pending) {
+      if ($this->queue->isEmpty()) {
+        throw new IllegalStateException('No pending results');
+      }
+
+      $this->comm->write(serialize($this->queue->get())."\n");
+    }
 
-  public function handle() {
-    return $this->comm->getHandle();
+    $element= unserialize($this->comm->readLine());
+    $this->pending= false;
+    return $element;
   }
 
+  /**
+   * Shuts down this worker process
+   *
+   * @return int
+   */
   public function shutdown() {
     if (-1 === $this->proc->exitValue()) {
       try {
diff --git a/src/main/php/util/data/Sequence.class.php b/src/main/php/util/data/Sequence.class.php
index f7a2692..03a77bd 100755
--- a/src/main/php/util/data/Sequence.class.php
+++ b/src/main/php/util/data/Sequence.class.php
@@ -416,7 +416,6 @@ public function counting(&$count) {
     }));
   }
 
-
   /**
    * Returns a stream with distinct elements
    *
diff --git a/src/test/php/util/data/unittest/LocalWorkerProcessTest.class.php b/src/test/php/util/data/unittest/LocalWorkerProcessTest.class.php
new file mode 100755
index 0000000..08df5c5
--- /dev/null
+++ b/src/test/php/util/data/unittest/LocalWorkerProcessTest.class.php
@@ -0,0 +1,49 @@
+<?php namespace util\data\unittest;
+
+use util\data\LocalWorkerProcess;
+
+class LocalWorkerProcessTest extends \unittest\TestCase {
+  private static $worker;
+
+  #[@beforeClass]
+  public static function startWorker() {
+    self::$worker= new LocalWorkerProcess('util.data.unittest.Worker', [3]);
+  }
+
+  #[@afterClass]
+  public static function stopWorker() {
+    self::$worker->shutdown();
+  }
+
+  #[@test]
+  public function pass_and_result_roundtrip() {
+    self::$worker->pass(1);
+    $this->assertEquals(3, self::$worker->result());
+  }
+
+  #[@test]
+  public function elements_are_processed_consecutively() {
+    self::$worker->pass(1);
+    $this->assertEquals(3, self::$worker->result());
+    self::$worker->pass(2);
+    $this->assertEquals(6, self::$worker->result());
+  }
+
+  #[@test, @values([
+  #  [[1, 2]],
+  #  [[1, 2, 3]],
+  #  [[1, 2, 3, 4]]
+  #])]
+  public function elements_can_be_processed_in_batches($values) {
+    foreach ($values as $value) {
+      self::$worker->pass($value);
+    }
+
+    $recv= [];
+    foreach ($values as $value) {
+      $recv[]= self::$worker->result() / 3;
+    }
+
+    $this->assertEquals($values, $recv);
+  }
+}
\ No newline at end of file
diff --git a/src/test/php/util/data/unittest/Worker.class.php b/src/test/php/util/data/unittest/Worker.class.php
index c54012c..25a81a9 100755
--- a/src/test/php/util/data/unittest/Worker.class.php
+++ b/src/test/php/util/data/unittest/Worker.class.php
@@ -29,11 +29,16 @@ public function handleDisconnect($socket) { }
       public function handleError($socket, $e) { }
 
       public function handleData($socket) {
+        file_put_contents("server.log", "HANDLE(".$socket->getHandle().")\n", FILE_APPEND);
         $in= $socket->readLine();
+        file_put_contents("server.log", "LINE `$in`\n", FILE_APPEND);
+
         if ("SHUTDOWN" === $in) {
           $this->server->terminate= true;
         } else if ($in) {
           $socket->write(serialize(unserialize($in) * $this->factor)."\n");
+        } else {
+          $socket->write("-ERR\n");
         }
       }  
     }'));

From 372e63bc4c1749a92d8eccbfe8b2dffac1fc485f Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Mon, 24 Nov 2014 22:29:23 +0100
Subject: [PATCH 07/10] Pass startup options to Runtime::newInstance()

---
 src/main/php/util/data/LocalWorkerProcess.class.php | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/src/main/php/util/data/LocalWorkerProcess.class.php b/src/main/php/util/data/LocalWorkerProcess.class.php
index fc3671a..0c3beb5 100755
--- a/src/main/php/util/data/LocalWorkerProcess.class.php
+++ b/src/main/php/util/data/LocalWorkerProcess.class.php
@@ -24,7 +24,9 @@ public function __construct($class, $args= []) {
     $this->queue= new Queue();
     $this->pending= false;
     $this->cmd= $class.' ['.implode(', ', $args).']';
-    $this->proc= Runtime::getInstance()->newInstance(null, 'class', $class, $args);
+
+    $rt= Runtime::getInstance();
+    $this->proc= $rt->newInstance($rt->startupOptions(), 'class', $class, $args);
     $this->connect();
   }
 

From 825785591ae6b9105c851fa962ea8e0c3e11a342 Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Mon, 24 Nov 2014 22:40:11 +0100
Subject: [PATCH 08/10] Travis-CI debugging: Dump exe.args

---
 src/main/php/util/data/LocalWorkerProcess.class.php | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/src/main/php/util/data/LocalWorkerProcess.class.php b/src/main/php/util/data/LocalWorkerProcess.class.php
index 0c3beb5..be0757c 100755
--- a/src/main/php/util/data/LocalWorkerProcess.class.php
+++ b/src/main/php/util/data/LocalWorkerProcess.class.php
@@ -21,12 +21,13 @@ class LocalWorkerProcess extends \lang\Object {
    * @param  string[] $args
    */
   public function __construct($class, $args= []) {
+
+    var_dump(Runtime::getInstance()->getExecutable()->getArguments());
+
     $this->queue= new Queue();
     $this->pending= false;
     $this->cmd= $class.' ['.implode(', ', $args).']';
-
-    $rt= Runtime::getInstance();
-    $this->proc= $rt->newInstance($rt->startupOptions(), 'class', $class, $args);
+    $this->proc= Runtime::getInstance()->newInstance(null, 'class', $class, $args);
     $this->connect();
   }
 

From 2c40ddc43c9af2514cecd7c30786a67dae188bbe Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Mon, 24 Nov 2014 22:47:44 +0100
Subject: [PATCH 09/10] Travis-CI debugging: Use XP @ `.`

---
 .travis.yml                                         | 1 +
 src/main/php/util/data/LocalWorkerProcess.class.php | 3 ---
 2 files changed, 1 insertion(+), 3 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index 4c5dd1d..cd91bed 100755
--- a/.travis.yml
+++ b/.travis.yml
@@ -11,6 +11,7 @@ before_script:
   - wget 'https://github.com/xp-framework/xp-runners/releases/download/v5.0.0/setup' -O - | php
   - wget 'https://github.com/xp-framework/core/releases/download/v6.0.0alpha7/xp-rt-6.0.0alpha7.xar'
   - ls -1 *.xar > boot.pth
+  - echo "use=." > xp.ini
   - echo "[runtime]" >> xp.ini
   - echo "date.timezone=Europe/Berlin" >> xp.ini
 
diff --git a/src/main/php/util/data/LocalWorkerProcess.class.php b/src/main/php/util/data/LocalWorkerProcess.class.php
index be0757c..fc3671a 100755
--- a/src/main/php/util/data/LocalWorkerProcess.class.php
+++ b/src/main/php/util/data/LocalWorkerProcess.class.php
@@ -21,9 +21,6 @@ class LocalWorkerProcess extends \lang\Object {
    * @param  string[] $args
    */
   public function __construct($class, $args= []) {
-
-    var_dump(Runtime::getInstance()->getExecutable()->getArguments());
-
     $this->queue= new Queue();
     $this->pending= false;
     $this->cmd= $class.' ['.implode(', ', $args).']';

From 4e1d26535434f596b68b063f3a3f1127017bcaf6 Mon Sep 17 00:00:00 2001
From: Timm Friebe <thekid@thekid.de>
Date: Mon, 24 Nov 2014 23:24:04 +0100
Subject: [PATCH 10/10] Remove debugging code

---
 src/test/php/util/data/unittest/Worker.class.php | 2 --
 1 file changed, 2 deletions(-)

diff --git a/src/test/php/util/data/unittest/Worker.class.php b/src/test/php/util/data/unittest/Worker.class.php
index 25a81a9..1de413e 100755
--- a/src/test/php/util/data/unittest/Worker.class.php
+++ b/src/test/php/util/data/unittest/Worker.class.php
@@ -29,9 +29,7 @@ public function handleDisconnect($socket) { }
       public function handleError($socket, $e) { }
 
       public function handleData($socket) {
-        file_put_contents("server.log", "HANDLE(".$socket->getHandle().")\n", FILE_APPEND);
         $in= $socket->readLine();
-        file_put_contents("server.log", "LINE `$in`\n", FILE_APPEND);
 
         if ("SHUTDOWN" === $in) {
           $this->server->terminate= true;