PHP - Manual: parallel\Channel

2024-11-14

The parallel\Channel class

(0.9.0)

Unbuffered Channels

An unbuffered channel will block on calls to parallel\Channel::send() until there is a receiver, and block on calls to parallel\Channel::recv() until there is a sender. This means an unbuffered channel is not only a way to share data among tasks but also a simple method of synchronization.

An unbuffered channel is the fastest way to share data among tasks, requiring the least copying.

Buffered Channels

A buffered channel will not block on calls to parallel\Channel::send() until capacity is reached, calls to parallel\Channel::recv() will block until there is data in the buffer.

Closures over Channels

A powerful feature of parallel channels is that they allow the exchange of closures between tasks (and runtimes).

When a closure is sent over a channel the closure is buffered, it doesn't change the buffering of the channel transmitting the closure, but it does effect the static scope inside the closure: The same closure sent to different runtimes, or the same runtime, will not share their static scope.

This means that whenever a closure is executed that was transmitted by a channel, static state will be as it was when the closure was buffered.

Anonymous Channels

The anonymous channel constructor allows the programmer to avoid assigning names to every channel: parallel will generate a unique name for anonymous channels.

类摘要

final class parallel\Channel {

/* Anonymous Constructor */

public __construct()

public __construct(int $capacity)

/* Access */

public make(string $name): Channel

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

public open(string $name): Channel

/* Sharing */

public recv(): mixed

public send(mixed $value): void

/* Closing */

public close(): void

/* Constant for Infinitely Buffered */

const Infinite;

}


This is an example of using a channel to produce data for consumers. In this example, the producer Runtime instance will send the time in seconds in which the consumers shall sleep.

<?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);
}

down

rustysun ¶

2 years ago


an example used unbuffered channel.
<?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];
    //unbuffered channel
    $runtime=new Runtime;
    $ch2=new Channel;
    $runtime->run($sum, [array_slice($a, 0, $num), $ch2]);
    $runtime->run($sum, [array_slice($a, $num), $ch2]);
    //receive from channel
    $x=$ch2->recv();
    $y=$ch2->recv();
    $ch2->close();
    echo "\nch2:", $x, "\t", $y, "\t", $x + $y, "\n";
} catch(Error $err) {
    echo "\nError:", $err->getMessage();
} catch(Exception $e) {
    echo "\nException:", $e->getMessage();
}

//output
//ch2:18  23      41

down

gam6itko ¶

9 months ago


<?php

// the very weird way to calculate factorial ^_^
// we create one thread and synching them with buffered channel
// at fact only one thread is executing at the time 

use parallel\{Channel, Future, Runtime};

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

/**
 * Creates $n threads.
 */
function factorial(int $n): int
{
    // buffered channel - using for sync threads ^_^
    $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);

    // waiting until all threads are done
    do {
        $allDone = array_reduce(
            $futureList,
            function (bool $c, Future $future): bool {

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

    return $channel->recv();
}

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

down

rustysun ¶

2 years ago


<?php
use parallel\Channel;

function sum(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];
    $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);

    //receive from channel
    $x=$ch1->recv();
    $y=$ch1->recv();
    $ch1->close();
    echo "\nch1:", $x, "\t", $y, "\t", $x + $y, "\n";
} catch(Error $err) {
    echo "\nError:", $err->getMessage();
} catch(Exception $e) {
    echo "\nException:", $e->getMessage();
}

add a note

官方地址：https://www.php.net/manual/en/class.parallel-channel.php

有任何技术问题请点击这里网站运营推广招聘

IT PHP 编程语言 Linux 开发编程科技 Elasticsearch HTML/CSS/XML 网络 JAVA NoSQL 数据库面试 C/C++ Golang Git 算法操作系统正则表达式 Redis 互联网 MySql JavaScript 运维软件国际架构设计 Mac OS TCP/IP Excel Vim Windows Socket Oracle VR MongoDB Python 运营 MemCache 商业硬件电子娱乐设计 nginx 摄影游戏 WordPress HTTP 团建数码电器

php7 安装fileinfo扩展 preg_split — 通过一个正则表达式分隔字符串 php位值,解决 PHP 中 usort 在值相同时改变原始位置的问题 [PHP] inet_pton/inet_ntop IP地址转换函数 array_multisort() 多字段排序 PHPMailer设置utf8 PHPMailer字符集CharSet ADORecordSet对象 adodb手册 PHP 8.1 [鸟哥]PHP_INT_MIN 和 -9223372036854775808 opcache预加载 Composer的Packagist资源 mysql面试题 [鸟哥]PHP FFI详解 - 一种全新的PHP扩展方式 PHP7添加redis扩展 PHP mkdir()写出来的权限与mode值不符合 Packagist / Composer 中国全量镜像 php中几种注释规范利用php soap实现web service PHP获取指定函数定义在哪个文件中及行号

略微加速

PHP官方手册 - 互联网笔记

The parallel\Channel class

Unbuffered Channels

Buffered Channels

Closures over Channels

Anonymous Channels

类摘要

目录

User Contributed Notes 4 notes