note 126581 added to parallel.run

<?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 generator function 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); ?> ---- Server IP: 45.112.84.4 Probable Submitter: 45.112.85.242 (proxied: 92.184.102.202) ---- Manual Page -- https://php.net/manual/en/parallel.run.php Edit -- https://main.php.net/note/edit/126581 Del: integrated -- https://main.php.net/note/delete/126581/integrated Del: useless -- https://main.php.net/note/delete/126581/useless Del: bad code -- https://main.php.net/note/delete/126581/bad+code Del: spam -- https://main.php.net/note/delete/126581/spam Del: non-english -- https://main.php.net/note/delete/126581/non-english Del: in docs -- https://main.php.net/note/delete/126581/in+docs Del: other reasons-- https://main.php.net/note/delete/126581 Reject -- https://main.php.net/note/reject/126581 Search -- https://main.php.net/manage/user-notes.php