danog/parallel
1
0
Fork 0
mirror of https://github.com/danog/parallel.git synced 2025-01-22 14:01:14 +01:00
Code Issues Projects Releases Wiki Activity

Improve worker-pool example (#30)

Browse Source
This commit is contained in:
Cyril Panshine 2017-11-20 19:21:53 +01:00 committed by Niklas Keller
parent dbd3403087
commit bcfb30fa35
2 changed files with 29 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
examples/BlockingTask.php
View File

@ -31,4 +31,8 @@ class BlockingTask implements Task {
public function run(Environment $environment) {
return ($this->function)(...$this->args);
}
public function getArgs() {
return $this->args;
}
}

45
examples/worker-pool.php
View File

@ -7,9 +7,20 @@ use Amp\Loop;
use Amp\Parallel\Example\BlockingTask;
use Amp\Parallel\Worker\DefaultPool;
Loop::run(function () {
$timer = Loop::repeat(100, function () {
printf(".\n");
// A variable to store our fetched results
$results = [];
// We can first define tasks and then run them
$tasks = [
new BlockingTask('file_get_contents', 'http://php.net'),
new BlockingTask('file_get_contents', 'https://amphp.org'),
new BlockingTask('file_get_contents', 'https://github.com'),
];
// Event loop for parallel tasks
Loop::run(function () use (&$results, &$tasks) {
$timer = Loop::repeat(200, function () {
printf(".");
});
Loop::unreference($timer);
@ -18,23 +29,14 @@ Loop::run(function () {
$coroutines = [];
$coroutines[] = function () use ($pool) {
$url = 'https://google.com';
$result = yield $pool->enqueue(new BlockingTask('file_get_contents', $url));
printf("Read from %s: %d bytes\n", $url, strlen($result));
};
$coroutines[] = function () use ($pool) {
$url = 'http://amphp.org';
$result = yield $pool->enqueue(new BlockingTask('file_get_contents', $url));
printf("Read from %s: %d bytes\n", $url, strlen($result));
};
$coroutines[] = function () use ($pool) {
$url = 'https://github.com';
$result = yield $pool->enqueue(new BlockingTask('file_get_contents', $url));
printf("Read from %s: %d bytes\n", $url, strlen($result));
};
foreach ($tasks as $task) {
$coroutines[] = function () use ($pool, $task, &$results) {
$result = yield $pool->enqueue($task);
$url = $task->getArgs()[0];
printf("\nRead from %s: %d bytes\n", $url, strlen($result));
$results[$url] = $result;
};
}
$coroutines = array_map(function (callable $coroutine): Coroutine {
return new Coroutine($coroutine());
@ -44,3 +46,6 @@ Loop::run(function () {
return yield $pool->shutdown();
});
echo "\nResult array keys:\n";
echo var_export(array_keys($results), true);