parallel\Channel 類別

(0.9.0)

無緩衝通道

無緩衝通道將在呼叫 parallel\Channel::send() 時阻塞，直到有接收器為止；並在呼叫 parallel\Channel::recv() 時阻塞，直到有發送器為止。這表示無緩衝通道不僅是在任務之間共享資料的一種方式，也是一種簡單的同步方法。

無緩衝通道是在任務之間共享資料的最快方法，所需的複製最少。

緩衝通道

使用 parallel\Channel::send() 呼叫緩衝通道時，在達到容量上限之前不會阻塞；而呼叫 parallel\Channel::recv() 時，會阻塞直到緩衝區中有數據為止。

通道的閉包

parallel 通道一個強大的特性是它們允許在任務（和運行時）之間交換閉包。

當一個閉包通過通道發送時，該閉包會被緩衝。它不會改變傳輸閉包的通道的緩衝行為，但它會影響閉包內的靜態作用域：發送到不同運行時或相同運行時的相同閉包不會共享它們的靜態作用域。

這意味著，每當執行由通道傳輸的閉包時，靜態狀態將與閉包被緩衝時的狀態相同。

匿名通道

匿名通道構造函數允許程式設計師避免為每個通道命名：parallel 會為匿名通道生成唯一的名稱。

類別概要

final class parallel\Channel {

/* 匿名建構子 */

public __construct()

public __construct(int $capacity)

/* 存取 */

public make(string $name): Channel

public make(string $name, int $capacity): Channel

public open(string $name): Channel

/* 共享 */

public recv(): mixed

public send(mixed $value): void

/* 關閉 */

public close(): void

/* 無限緩衝常數 */

const Infinite;

}

這是一個使用通道為消費者產生數據的範例。在這個範例中，生產者 Runtime 實例將發送消費者應休眠的秒數。

<?php

use parallel\{Runtime, Channel};

main($argv);

function main(array $argv)
{
 if (count($argv) !== 3) {
 echo "Type: hello-parallel.php <number-of-tasks> <maximum-time-of-sleep (in seconds)>" . PHP_EOL;
 echo "Example: hello-parallel.php 5 3" . PHP_EOL;
 die;
 } else {
 $numberOfTasks = intval($argv[1]);
 $maximumTimeOfSleep = intval($argv[2]);
 $t1 = microtime(true);
 parallelize($numberOfTasks, $maximumTimeOfSleep);
 $endTime = microtime(true) - $t1;
 echo PHP_EOL."Finished $numberOfTasks task(s) in {$endTime}s".PHP_EOL;
 }
}

function parallelize(int $numberOfTasks, int $maximumTimeOfSleep)
{
 $channel = new Channel();

 $taskIds = array_map(function () use ($maximumTimeOfSleep) {
 return $id = uniqid("task::");
 }, range(0, $numberOfTasks - 1));

 $timesToSleep = array_map(function () use ($maximumTimeOfSleep) {
 return rand(1, $maximumTimeOfSleep);
 }, $taskIds);

 $producer = new Runtime();
 $producerFuture = $producer->run(function (Channel $channel, array $timesToSleep) {
 foreach ($timesToSleep as $timeToSleep) {
 $channel->send($timeToSleep);
 }
 }, [$channel, $timesToSleep]);

 $consumerFutures = array_map(function (string $id) use ($channel) {
 $runtime = new Runtime();
 return $runtime->run(function (string $id, Channel $channel) {
 $timeToSleep = $channel->recv();
 echo "Hello from $id. I will sleep for $timeToSleep second(s).".PHP_EOL;
 sleep($timeToSleep);
 echo "$id slept for $timeToSleep second(s).".PHP_EOL;
 return $timeToSleep;
 }, [$id, $channel]);
 }, $taskIds);

 wait($consumerFutures);
 wait([$producerFuture]);
}

function wait(array $futures)
{
 return array_map(function ($future) {
 return $future->value();
 }, $futures);
}

往上

往下

rustysun ¶

5 年前

一個使用無緩衝通道的例子。
<?php

use parallel\{Channel,Runtime};

$sum=function(array $a, Channel $ch) {
 $sum=0;
 foreach ($a as $v) {
 $sum+=$v;
 }
 $ch->send($sum);
};
try {
 $a=[7, 2, 8, 1, 4, 0, 9, 10];
 //無緩衝通道
 $runtime=new Runtime;
 $ch2=new Channel;
 $runtime->run($sum, [array_slice($a, 0, $num), $ch2]);
 $runtime->run($sum, [array_slice($a, $num), $ch2]);
 //從通道接收
 $x=$ch2->recv();
 $y=$ch2->recv();
 $ch2->close();
 echo "\nch2:", $x, "\t", $y, "\t", $x + $y, "\n";
} catch(Error $err) {
 echo "\n錯誤:", $err->getMessage();
} catch(Exception $e) {
 echo "\n例外:", $e->getMessage();
}

//輸出
//ch2:18 23 41

往上

往下

gam6itko ¶

3 年前

<?php

// 以非常奇怪的方式計算階乘 ^_^
// 我們創建一個執行緒，並使用緩衝通道同步它們
// 實際上，在同一時間只有一個執行緒在執行

use parallel\{Channel, Future, Runtime};

for ($n = 0; $n <= 10; $n++) {
 echo "!$n = " . factorial($n) . PHP_EOL;
}

/**
 * 建立 $n 個執行緒。
 */
function factorial(int $n): int
{
 // 緩衝通道 - 用於同步執行緒 ^_^
 $channel = new Channel(1);
 $futureList = [];
 for ($i = 2; $i <= $n; $i++) {
 $runtime = new Runtime();
 $futureList[] = $runtime->run(
 static function (Channel $channel, $multiplier): void {
 $f = $channel->recv();
 $channel->send($f * $multiplier);
 },
 [$channel, $i]
 );
 }

 $channel->send(1);

 // 等待所有執行緒完成
 do {
 $allDone = array_reduce(
 $futureList,
 function (bool $c, Future $future): bool {

 return $c && $future->done();
 },
 true
 );
 } while (false === $allDone);

 return $channel->recv();
}

// 輸出：
// !0 = 1
// !1 = 1
// !2 = 2
// !3 = 6
// !4 = 24
// !5 = 120
// !6 = 720
// !7 = 5040
// !8 = 40320
// !9 = 362880
// !10 = 3628800

往上

往下

rustysun ¶

5 年前

<?php
使用 parallel\Channel;

函式 sum(陣列 $a, Channel $ch) {
 $sum=0;
 foreach ($a as $v) {
 $sum+=$v;
 }
 $ch->send($sum);
}

try {
 $a=[7, 2, 8, 1, 4, 0, 9, 10];
 $ch1=Channel::make('sum', 2);
 $ch2=new Channel;
 $num=count($a) / 2;
 sum(array_slice($a, 0, $num), $ch1);
 sum(array_slice($a, $num), $ch1);

 //從通道接收
 $x=$ch1->recv();
 $y=$ch1->recv();
 $ch1->close();
 echo "\nch1:", $x, "\t", $y, "\t", $x + $y, "\n";
} catch(Error $err) {
 echo "\n錯誤:", $err->getMessage();
} catch(Exception $e) {
 echo "\n例外:", $e->getMessage();
}

往上

往下

thierry at pielo dot net ¶

3 個月前

<?php

/**
 * Bzz reloaded!
 * Run two simple tasks in parallel and synchronize them with a channel
 * 
 * parallel\Channel(int $capacity): Buffered channel
 * Creates a buffered channel for communication between tasks
 * @ref https://php.dev.org.tw/manual/en/class.parallel-channel.php
 */

 echo "zzz... " . PHP_EOL;

// Create new buffered channel
$channel = new \parallel\Channel(2);

\parallel\run(
 function($channel) {
 $snaps_count = rand (8, 12);
 echo "Number of snaps: $snaps_count" . PHP_EOL;
 for ($i=1; $i<=$snaps_count; $i++) {
 $other_sleep_time = rand(3, 5);
 $my_sleep_time = rand(1, 3);
 echo "Send sleep time to buffer" . PHP_EOL;
 $start = microtime(true);
 $channel->send($other_sleep_time);
 $wait_time = microtime(true) - $start;
 if ($wait_time > .1) {
 echo "Buffer was full. I waited " . round($wait_time) . "s" . PHP_EOL;
 }
 echo "I sleep for {$my_sleep_time}s" . PHP_EOL;
 sleep($my_sleep_time);
 }
 echo "I finished sleeping. Closing channel" . PHP_EOL;
 $channel->close();
 },
 [$channel]
);

\parallel\run(
 function($channel) {
 try {
 while(true) {
 $my_sleep_time = $channel->recv();
 echo "Other sleeps for {$my_sleep_time}s" . PHP_EOL;
 sleep($my_sleep_time);
 }
 } catch(\parallel\Channel\Error\Closed $e) {
 echo "Channel is closed. Other die.";
 die;
 }
 },
 [$channel]
);

＋新增註釋