danog/parallel
1
0
Fork 0
mirror of https://github.com/danog/parallel.git synced 2024-11-26 20:34:40 +01:00
Code Issues Projects Releases Wiki Activity

Remove $this magic from Thread callback

Browse Source 
Channel is now passed as the first argument.
This commit is contained in:
Aaron Piotrowski 2017-12-10 16:43:19 -06:00
parent 4d4841f449
commit 852f580915
No known key found for this signature in database
GPG Key ID: ADD1EF783EDE9EEB
6 changed files with 18 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
examples/parcel.php
View File

@ -5,13 +5,14 @@ require dirname(__DIR__).'/vendor/autoload.php';
use Amp\Delayed;
use Amp\Loop;
use Amp\Parallel\Context\Thread;
use Amp\Parallel\Sync\Channel;
use Amp\Parallel\Sync\Parcel;
use Amp\Parallel\Sync\ThreadedParcel;
Loop::run(function () {
$parcel = new ThreadedParcel(1);
$context = Thread::spawn(function (Parcel $parcel) {
$context = Thread::spawn(function (Channel $channel, Parcel $parcel) {
$value = yield $parcel->synchronized(function (int $value) {
return $value + 1;
});

9
examples/thread.php
View File

@ -5,6 +5,7 @@ require dirname(__DIR__).'/vendor/autoload.php';
use Amp\Delayed;
use Amp\Loop;
use Amp\Parallel\Context\Thread;
use Amp\Parallel\Sync\Channel;
Loop::run(function () {
$timer = Loop::repeat(1000, function () {
@ -15,15 +16,13 @@ Loop::run(function () {
try {
// Create a new child thread that does some blocking stuff.
$context = Thread::spawn(function () {
printf("\$this: %s\n", get_class($this));
printf("Received the following from parent: %s\n", yield $this->receive());
$context = Thread::spawn(function (Channel $channel): \Generator {
printf("Received the following from parent: %s\n", yield $channel->receive());
print "Sleeping for 3 seconds...\n";
sleep(3); // Blocking call in thread.
yield $this->send("Data sent from child.");
yield $channel->send("Data sent from child.");
print "Sleeping for 2 seconds...\n";
sleep(2); // Blocking call in thread.

8
lib/Context/Internal/Thread.php
View File

@ -118,13 +118,7 @@ class Thread extends \Thread {
*/
private function execute(Channel $channel): \Generator {
try {
if ($this->function instanceof \Closure) {
$result = call($this->function->bindTo($channel, null), ...$this->args);
} else {
$result = call($this->function, ...$this->args);
}
$result = new ExitSuccess(yield $result);
$result = new ExitSuccess(yield call($this->function, $channel, ...$this->args));
} catch (\Throwable $exception) {
$result = new ExitFailure($exception);
}

8
lib/Context/Thread.php
View File

@ -53,7 +53,9 @@ class Thread implements Context {
/**
* Spawns a new thread and runs it.
*
* @param callable $function The callable to invoke in the thread.
* @param callable $function The callable to invoke in the thread. First argument is an instance of
* \Amp\Parallel\Sync\Channel.
* @param mixed ...$args Additional arguments to pass to the given callable.
*
* @return Thread The thread object that was spawned.
*/
@ -66,7 +68,9 @@ class Thread implements Context {
/**
* Creates a new thread.
*
* @param callable $function The callable to invoke in the thread when run.
* @param callable $function The callable to invoke in the thread. First argument is an instance of
* \Amp\Parallel\Sync\Channel.
* @param mixed ...$args Additional arguments to pass to the given callable.
*
* @throws \Error Thrown if the pthreads extension is not available.
*/

5
lib/Worker/WorkerThread.php
View File

@ -3,6 +3,7 @@
namespace Amp\Parallel\Worker;
use Amp\Parallel\Context\Thread;
use Amp\Parallel\Sync\Channel;
use Amp\Promise;
/**
@ -14,7 +15,7 @@ class WorkerThread extends AbstractWorker {
* Defaults to \Amp\Parallel\Worker\BasicEnvironment.
*/
public function __construct(string $envClassName = BasicEnvironment::class) {
parent::__construct(new Thread(function (string $className): Promise {
parent::__construct(new Thread(function (Channel $channel, string $className): Promise {
if (!\class_exists($className)) {
throw new \Error(\sprintf("Invalid environment class name '%s'", $className));
}
@ -29,7 +30,7 @@ class WorkerThread extends AbstractWorker {
\define("AMP_WORKER", "amp-worker");
}
$runner = new TaskRunner($this, $environment);
$runner = new TaskRunner($channel, $environment);
return $runner->run();
}, $envClassName));
}

3
test/Sync/ThreadedParcelTest.php
View File

@ -4,6 +4,7 @@ namespace Amp\Parallel\Test\Sync;
use Amp\Loop;
use Amp\Parallel\Context\Thread;
use Amp\Parallel\Sync\Channel;
use Amp\Parallel\Sync\ThreadedParcel;
/**
@ -19,7 +20,7 @@ class ThreadedParcelTest extends AbstractParcelTest {
$value = 1;
$parcel = new ThreadedParcel($value);
$thread = Thread::spawn(function (ThreadedParcel $parcel) {
$thread = Thread::spawn(function (Channel $channel, ThreadedParcel $parcel) {
$parcel->synchronized(function (int $value) {
return $value + 1;
});