parallel\run

Here's a more substantial example of how to use the run functional API.<?php/********************************************* * Sample parallel functional API *  * Scenario * ------------------------------------------- * Given a large number of rows of * data to process, divide the work amongst * a set of workers.  Each worker is responsible * for finishing their assigned task. *  * In the code below, assume we have arbitrary  * start and end IDs (rows) - we will try to * divide the number of IDs (rows) evenly * across 8 workers.  The workers will get the * following batches to process to completion: * * Total number of IDs (rows): 1371129 * Each worker will get 171392 IDs to process * * Worker 1: IDs from 11001 to 182393 * Worker 2: IDs from 182393 to 353785 * Worker 3: IDs from 353785 to 525177 * Worker 4: IDs from 525177 to 696569 * Worker 5: IDs from 696569 to 867961 * Worker 6: IDs from 867961 to 1039353 * Worker 7: IDs from 1039353 to 1210745 * Worker 8: IDs from 1210745 to 1382130 * * Each worker then processes 5000 rows at a time * until they are done with their assigned work * *********************************************/use \parallel\{Runtime, Future, Channel, Events};$minId = 11001;$maxId = 1382130;$workers = 8;$totalIds = $maxId - $minId;// Try to divide IDs evenly across the number of workers$batchSize = ceil($totalIds / $workers);// The last batch gets whatever is left over$lastBatch = $totalIds % $batchSize;// The number of IDs (rows) to divide the overall// task into sub-batches $rowsToFetch = 5000;print "Total IDs: " . $totalIds . "\n";print "Batch Size: " . $batchSize . "\n";print "Last Batch: " . $lastBatch . "\n";$producer = function(int $worker, int $startId, int $endId, int $fetchSize) {    $tempMinId = $startId;    $tempMaxId = $tempMinId + $fetchSize;    $fetchCount = 1;        print "Worker " . $worker . " working on IDs from " . $startId . " to " . $endId . "\n";        while($tempMinId < $endId) {        for($i = $tempMinId; $i < $tempMaxId; $i++) {            $usleep = rand(500000, 1000000);            usleep($usleep);            print "Worker " . $worker .  " finished batch " . $fetchCount . " from ID " . $tempMinId . " to " . $tempMaxId . "\n";            // Need to explicitly break out of the for loop once complete or else it will forever process only the first sub-batch            break;        }                // Now we move on to the next sub-batch for this worker        $tempMinId = $tempMaxId;        $tempMaxId = $tempMinId + $fetchSize;        if($tempMaxId > $endId) {            $tempMaxId = $endId;        }        // Introduce some timing randomness        $sleep = rand(1,5);        sleep($sleep);        $fetchCount++;    }        // This worker has completed their entire batch    print "Worker " . $worker .  " finished\n";    };// Create our workers and have them start working on their task// In this case, it's a set of 171392 IDs to processfor($i = 0; $i < $workers; $i++) {    $startId = $minId + ($i * $batchSize);    $endId = $startId + $batchSize;    if($i == ($workers - 1)) {        $endId = $maxId;    }    \parallel\run($producer, array(($i+1), $startId, $endId, $rowsToFetch));}?>

Although function declaration is not allowed inside thread exec code, include is allowed. So if we want to declare a function, we could write another file that contain the function and include it.# main.php<?php$runtime = new parallel\Runtime ();$future = $runtime->run ( function () {    $future = $runtime->run ( function () {        include "included.php";        return add (1, 3);    }, [ ] );echo $future->value ();# output: 4# included.php<?phpfunction add($a, $b){    return $a + $b;}

<?php/** * Sample parralel functional API * using a generator instead of a static list of items to process *  * Items to process in parallel come from a generator * It could be anything : e.g fetch a mysql array, a DirectoryIterator... * Thus the number of items to process in parallel is NOT known in advance *  * This algorithm attributes items to each parallel thread dynamically * As soon as a thread has finished working * It is assigned a new item to process * until all items are processed (generator closes) *  * In this example we process 50 items in 5 parallel threads * It produces output in this form (output changes at each run) : *  * ThreadId: 1 => Item: 1 (Start) * ThreadId: 2 => Item: 2 (Start) * ThreadId: 3 => Item: 3 (Start) * ThreadId: 4 => Item: 4 (Start) * ThreadId: 5 => Item: 5 (Start) * ThreadId: 5 => Item: 5 Sleep: 3s (End) * ThreadId: 5 => Item: 6 (Start) * ThreadId: 3 => Item: 3 Sleep: 4s (End) * ThreadId: 3 => Item: 7 (Start) * ThreadId: 2 => Item: 2 Sleep: 6s (End) * ThreadId: 2 => Item: 8 (Start) * ... * ThreadId: 4 => Item: 44 Sleep: 6s (End) * ThreadId: 4 => Item: 49 (Start) * ThreadId: 3 => Item: 46 Sleep: 5s (End) * ThreadId: 3 => Item: 50 (Start) * ThreadId: 2 => Item: 43 Sleep: 9s (End) * Destroy ThreadId: 2 * ThreadId: 1 => Item: 47 Sleep: 5s (End) * Destroy ThreadId: 1 * ThreadId: 4 => Item: 49 Sleep: 7s (End) * Destroy ThreadId: 4 * ThreadId: 5 => Item: 48 Sleep: 10s (End) * Destroy ThreadId: 5 * ThreadId: 3 => Item: 50 Sleep: 10s (End) * Destroy ThreadId: 3 */use \parallel\{Runtime, Future, Channel, Events};// Generate list of items to process with a generatorfunction generator(int $item_count) {    for ($i=1; $i <= $item_count; $i++) {        yield $i;    }}function testConcurrency(int $concurrency, int $item_count) {    $generator = generator($item_count);    // Function executing in each thread. Have a snap for a random time for example !    $producer = function (int $item_id) {        $seconds = rand(1, 10);        sleep($seconds);        return ['item_id' => $item_id, 'sleep_seconds' => $seconds];    };    // Fill up threads with initial 'inactive' state    $threads = array_fill(1, $concurrency, ['is_active' => false]);    while (true) {        // Loop through threads until all threads are finished        foreach ($threads as $thread_id => $thread) {            if (!$thread['is_active'] and $generator->valid()) {                // Thread is inactive and generator still has values : run something in the thread                $item_id = $generator->current();                $threads[$thread_id]['run'] = \parallel\run($producer, [$item_id]);                echo "ThreadId: $thread_id => Item: $item_id (Start)\n";                $threads[$thread_id]['is_active'] = true;                $generator->next();            } elseif (!isset($threads[$thread_id]['run'])) {                // Destroy supplementary threads in case generator closes sooner than number of threads                echo "Destroy ThreadId: $thread_id\n";                unset($threads[$thread_id]);            } elseif ($threads[$thread_id]['run']->done()) {                // Thread finished. Get results                $item = $threads[$thread_id]['run']->value();                echo "ThreadId: $thread_id => Item: {$item['item_id']} Sleep: {$item['sleep_seconds']}s (End)\n";                if (!$generator->valid()) {                    // Generator is closed then destroy thread                    echo "Destroy ThreadId: $thread_id\n";                    unset($threads[$thread_id]);                } else {                    // Thread is ready to run again                    $threads[$thread_id]['is_active'] = false;                }            }        }        // Escape loop when all threads are destroyed        if (empty($threads)) break;    }}$concurrency = 5;$item_count = 50;testConcurrency($concurrency, $item_count);?>