Hotplate.php

<?php

require __DIR__ . '/../vendor/rdlowrey/Amp/autoload.php';

require __DIR__ . '/setup.php';

use Amp\Reactor,
    Amp\ReactorFactory,
    Amp\Async\CallResult,
    Amp\Async\Dispatcher,
    Amp\Async\Processes\FrameParser,
    Amp\Async\Processes\ProcessDispatcher;

class Hotplate {

    /**
     * @var \Amp\Reactor
     */
    private $reactor;

    /**
     * @var \Amp\Async\Dispatcher
     */
    private $dispatcher;

    private $recalculated = 0;

    private $workers = 0;

    private $iterations = 0;

    private $to = [];
    private $from = [];

    function __construct(Reactor $reactor, Dispatcher $dispatcher, $workers) {
        $this->reactor = $reactor;
        $this->dispatcher = $dispatcher;
        $dispatcher->setAutoTimeoutInterval($seconds = 10);
        $this->workers = $workers;

        $this->from = array_fill(0, ROWS, array_fill(0, COLUMNS, 50.0));
        for ($i = 1; $i < LAST_ROW; $i++) {
            $this->from[0][$i] = 0;
            $this->from[$i][0] = 100.0;
            $this->from[$i][LAST_COLUMN] = 100.0;
            $this->from[LAST_ROW][$i] = 100.0;
        }
        $row_cache =& $this->from[400];
        for ($column = 1; $column < 330; $column++) {
            $row_cache[$column] = 100;
        }
        $this->from[200][500] = 100;

        $this->to = $this->from;
    }

    function balance() {
        $this->recalculate();
        $this->reactor->run();
    }

    function afterRecalculate(CallResult $result) {
        $echo =  "afterRecalculate . . .";
        if ($result->isSuccess()) {
            echo "$echo success\n";
            list($row, $length, $recalculatedRow) = $result->getResult();
            $stop = $row + $length;
            echo "[$row:{$stop}]\n";

            for (; $row < $stop; $row++) {
                $this->to[$row] = $recalculatedRow[$row];
            }
            $this->iterations++;
            if (++$this->recalculated == $this->workers) {
                $this->recalculated = 0;
                echo "recalculations completed!\n";
                if (!$this->isSteady()) {
                    list($this->to, $this->from) = [$this->from, $this->to];
                    $this->recalculate();
                } else {
                    $this->reactor->stop();
                }
            }
        } else {
            echo "$echo failure:{$result->getError()}\n";
            $this->reactor->stop();
        }

    }

    private function isSteady() {
        echo "isSteady\n";
        return TRUE;
    }

    protected function recalculate() {
        echo "recalculate\n";

        $chunkLength = ROWS / $this->workers;
        for ($i = 0; $i < $this->workers; $i++) {
            $start = $i * $chunkLength - 1;
            $length = $chunkLength + 1;
            if ($start < 0) {
                $start = 0;
                $length = $chunkLength;
            }

            echo "\tdispatch($start, $length)\n";
            $this->dispatcher->call(
                [$this, 'afterRecalculate'],
                'recalculate',
                $start,
                $length,
                array_slice($this->from, $start, $length, TRUE)
            );
        }
    }

}


$start = microtime(true);

$pids = 1;


$bin = PHP_BINARY . ' -c /Users/levijm/Projects/Hotplate/src/php.ini';
$workerScript = __DIR__ . '/../vendor/rdlowrey/Amp/workers/process_worker.php';
$includes = __DIR__ . '/recalculate.php';

echo "$bin $workerScript $includes\n";

$reactor = (new ReactorFactory)->select();

$dispatcher = new ProcessDispatcher(
    $reactor,
    "$bin $workerScript $includes"
);
$dispatcher->setMaxWorkers($pids);
$dispatcher->start();

$hotplate = new Hotplate($reactor, $dispatcher, $pids);

$hotplate->balance();

recalculate.php

<?php

require __DIR__ . '/setup.php';

function recalculate($start, $length, array $data) {
    $stop = $start + $length;
    $to = $data;

    for ($row = $start + 1; $row < $stop && $row < LAST_ROW; $row++) {
        $row_cache =& $to[$row];
        $from_cache =& $from[$row];
        for ($column = 1; $column < LAST_COLUMN; $column++) {
            $row_cache[$column] = (
                $from[$row - 1][$column]
                    + $from[$row + 1][$column]
                    + $from_cache[$column - 1]
                    + $from_cache[$column + 1]
                    + (4 * $from_cache[$column])
            ) / 8;
        }
    }
    if (isset($data[400])) {
        $row_cache =& $to[400];
        for ($column = 1; $column < 330; $column++) {
            $row_cache[$column] = 100;
        }
    }
    if (isset($data[200])) {
        $to[200][500] = 100;
    }

    return [$start, $length, $to];

}

setup.php

<?php

define('ROWS', 768);
define('COLUMNS', 768);
define('LAST_ROW', 767);
define('LAST_COLUMN', 767);

set_error_handler(function ($errno, $errstr, $errfile, $errline ) {
    throw new ErrorException($errstr, $errno, 0, $errfile, $errline);
});