diff --git a/src/Psl/Channel/Internal/ChannelState.php b/src/Psl/Channel/Internal/ChannelState.php index afca5d01..388ae398 100644 --- a/src/Psl/Channel/Internal/ChannelState.php +++ b/src/Psl/Channel/Internal/ChannelState.php @@ -6,8 +6,9 @@ use Psl\Channel\ChannelInterface; use Psl\Channel\Exception; -use Psl\DataStructure\Queue; -use Psl\DataStructure\QueueInterface; + +use function array_shift; +use function count; /** * @template T @@ -19,16 +20,15 @@ final class ChannelState implements ChannelInterface { /** - * @var QueueInterface + * @var array */ - private QueueInterface $messages; + private array $messages = []; private bool $closed = false; public function __construct( private ?int $capacity = null, ) { - $this->messages = new Queue(); } /** @@ -60,7 +60,7 @@ public function isClosed(): bool */ public function count(): int { - return $this->messages->count(); + return count($this->messages); } /** @@ -68,11 +68,7 @@ public function count(): int */ public function isFull(): bool { - if (null === $this->capacity) { - return false; - } - - return $this->capacity === $this->count(); + return $this->capacity && $this->capacity === $this->count(); } /** @@ -80,7 +76,7 @@ public function isFull(): bool */ public function isEmpty(): bool { - return 0 === $this->messages->count(); + return 0 === count($this->messages); } /** @@ -95,8 +91,8 @@ public function send(mixed $message): void throw Exception\ClosedChannelException::forSending(); } - if (null === $this->capacity || $this->capacity > $this->count()) { - $this->messages->enqueue($message); + if (null === $this->capacity || $this->capacity > count($this->messages)) { + $this->messages[] = $message; return; } @@ -112,17 +108,14 @@ public function send(mixed $message): void */ public function receive(): mixed { - $empty = 0 === $this->count(); - $closed = $this->closed; - if ($closed && $empty) { - throw Exception\ClosedChannelException::forReceiving(); - } + if ([] === $this->messages) { + if ($this->closed) { + throw Exception\ClosedChannelException::forReceiving(); + } - if ($empty) { throw Exception\EmptyChannelException::create(); } - /** @psalm-suppress MissingThrowsDocblock */ - return $this->messages->dequeue(); + return array_shift($this->messages); } } diff --git a/src/Psl/Channel/Internal/Receiver.php b/src/Psl/Channel/Internal/Receiver.php index c464a642..8e9ce901 100644 --- a/src/Psl/Channel/Internal/Receiver.php +++ b/src/Psl/Channel/Internal/Receiver.php @@ -34,29 +34,27 @@ public function __construct( public function receive(): mixed { // there's a pending operation? wait for it. - $this->deferred?->getAwaitable()->then(static fn() => null, static fn() => null)->ignore()->await(); + $this->deferred?->getAwaitable()->then(static fn() => null, static fn() => null)->await(); if ($this->state->isEmpty()) { $this->deferred = new Async\Deferred(); $identifier = Async\Scheduler::repeat(0.000000001, function (): void { - if ($this->state->isClosed()) { - /** - * Channel has been closed from the receiving side. - * - * @psalm-suppress PossiblyNullReference - */ - if (!$this->deferred->isComplete()) { - /** @psalm-suppress PossiblyNullReference */ - $this->deferred->error(Exception\ClosedChannelException::forReceiving()); - } + if (!$this->state->isEmpty()) { + /** @psalm-suppress PossiblyNullReference */ + $this->deferred->complete(null); return; } - if (!$this->state->isEmpty()) { + /** + * Channel has been closed from the sender side. + * + * @psalm-suppress PossiblyNullReference + */ + if ($this->state->isClosed() && !$this->deferred->isComplete()) { /** @psalm-suppress PossiblyNullReference */ - $this->deferred->complete(null); + $this->deferred->error(Exception\ClosedChannelException::forReceiving()); } }); @@ -94,7 +92,9 @@ public function getCapacity(): ?int */ public function close(): void { - $this->deferred?->error(Exception\ClosedChannelException::forReceiving()); + if ($this->state->isEmpty()) { + $this->deferred?->error(Exception\ClosedChannelException::forReceiving()); + } $this->state->close(); } diff --git a/src/Psl/Channel/Internal/Sender.php b/src/Psl/Channel/Internal/Sender.php index 31c62d1b..a9a956c3 100644 --- a/src/Psl/Channel/Internal/Sender.php +++ b/src/Psl/Channel/Internal/Sender.php @@ -34,7 +34,7 @@ public function __construct( public function send(mixed $message): void { // there's a pending operation? wait for it. - $this->deferred?->getAwaitable()->then(static fn() => null, static fn() => null)->ignore()->await(); + $this->deferred?->getAwaitable()->then(static fn() => null, static fn() => null)->await(); if ($this->state->isFull()) { $this->deferred = new Async\Deferred(); @@ -42,7 +42,7 @@ public function send(mixed $message): void $identifier = Async\Scheduler::repeat(0.000000001, function (): void { if ($this->state->isClosed()) { /** - * Channel has been closed from the receiving side. + * Channel has been closed from the receiver side. * * @psalm-suppress PossiblyNullReference */ diff --git a/tests/benchmark/Channel/CommunicationBench.php b/tests/benchmark/Channel/CommunicationBench.php new file mode 100644 index 00000000..e686518d --- /dev/null +++ b/tests/benchmark/Channel/CommunicationBench.php @@ -0,0 +1,75 @@ +receive(); + } + } catch (Channel\Exception\ClosedChannelException) { + return; + } + }); + + /** @psalm-suppress MissingThrowsDocblock */ + $file = File\open_read_only(__FILE__); + $reader = new IO\Reader($file); + /** @psalm-suppress MissingThrowsDocblock */ + while (!$reader->isEndOfFile()) { + $byte = $reader->readByte(); + + /** @psalm-suppress InvalidArgument */ + $sender->send($byte); + } + + $sender->close(); + + Async\Scheduler::run(); + } + + public function benchUnboundedCommunication(): void + { + [$receiver, $sender] = Channel\bounded(10); + + Async\Scheduler::defer(static function () use ($receiver) { + try { + while (true) { + $receiver->receive(); + } + } catch (Channel\Exception\ClosedChannelException) { + return; + } + }); + + /** @psalm-suppress MissingThrowsDocblock */ + $file = File\open_read_only(__FILE__); + $reader = new IO\Reader($file); + /** @psalm-suppress MissingThrowsDocblock */ + while (!$reader->isEndOfFile()) { + $byte = $reader->readByte(); + + /** @psalm-suppress InvalidArgument */ + $sender->send($byte); + } + + $sender->close(); + + Async\Scheduler::run(); + } +} diff --git a/tests/benchmark/Type/ArrayKeyTypeBench.php b/tests/benchmark/Type/ArrayKeyTypeBench.php index f5a917b3..5606b927 100644 --- a/tests/benchmark/Type/ArrayKeyTypeBench.php +++ b/tests/benchmark/Type/ArrayKeyTypeBench.php @@ -4,6 +4,7 @@ namespace Psl\Tests\Benchmark\Type; +use PhpBench\Attributes\Groups; use Psl\Tests\Benchmark\Type\Asset\ExplicitStringableObject; use Psl\Tests\Benchmark\Type\Asset\ImplicitStringableObject; use Psl\Type; @@ -11,6 +12,7 @@ /** * @extends GenericTypeBench> */ +#[Groups(['type'])] final class ArrayKeyTypeBench extends GenericTypeBench { /** diff --git a/tests/benchmark/Type/DictTypeBench.php b/tests/benchmark/Type/DictTypeBench.php index 0c925ed7..6a81ce3d 100644 --- a/tests/benchmark/Type/DictTypeBench.php +++ b/tests/benchmark/Type/DictTypeBench.php @@ -5,6 +5,7 @@ namespace Psl\Tests\Benchmark\Type; use ArrayIterator; +use PhpBench\Attributes\Groups; use Psl\Dict; use Psl\Type; use Psl\Vec; @@ -12,6 +13,7 @@ /** * @extends GenericTypeBench> */ +#[Groups(['type'])] final class DictTypeBench extends GenericTypeBench { /** diff --git a/tests/benchmark/Type/IntTypeBench.php b/tests/benchmark/Type/IntTypeBench.php index 2ca2a4d3..35e6d039 100644 --- a/tests/benchmark/Type/IntTypeBench.php +++ b/tests/benchmark/Type/IntTypeBench.php @@ -4,6 +4,7 @@ namespace Psl\Tests\Benchmark\Type; +use PhpBench\Attributes\Groups; use Psl\Tests\Benchmark\Type\Asset\ExplicitStringableObject; use Psl\Tests\Benchmark\Type\Asset\ImplicitStringableObject; use Psl\Type; @@ -11,6 +12,7 @@ /** * @extends GenericTypeBench> */ +#[Groups(['type'])] final class IntTypeBench extends GenericTypeBench { /** diff --git a/tests/benchmark/Type/NonEmptyStringTypeBench.php b/tests/benchmark/Type/NonEmptyStringTypeBench.php index b38cda3f..bc079d50 100644 --- a/tests/benchmark/Type/NonEmptyStringTypeBench.php +++ b/tests/benchmark/Type/NonEmptyStringTypeBench.php @@ -4,6 +4,7 @@ namespace Psl\Tests\Benchmark\Type; +use PhpBench\Attributes\Groups; use Psl\Tests\Benchmark\Type\Asset\ExplicitStringableObject; use Psl\Tests\Benchmark\Type\Asset\ImplicitStringableObject; use Psl\Type; @@ -11,6 +12,7 @@ /** * @extends GenericTypeBench> */ +#[Groups(['type'])] final class NonEmptyStringTypeBench extends GenericTypeBench { /** diff --git a/tests/benchmark/Type/ShapeTypeBench.php b/tests/benchmark/Type/ShapeTypeBench.php index b5b496b4..e7fb3846 100644 --- a/tests/benchmark/Type/ShapeTypeBench.php +++ b/tests/benchmark/Type/ShapeTypeBench.php @@ -5,11 +5,13 @@ namespace Psl\Tests\Benchmark\Type; use ArrayIterator; +use PhpBench\Attributes\Groups; use Psl\Type; /** * @extends GenericTypeBench> */ +#[Groups(['type'])] final class ShapeTypeBench extends GenericTypeBench { /** diff --git a/tests/benchmark/Type/StringTypeBench.php b/tests/benchmark/Type/StringTypeBench.php index 22d74f0a..6587848d 100644 --- a/tests/benchmark/Type/StringTypeBench.php +++ b/tests/benchmark/Type/StringTypeBench.php @@ -4,6 +4,7 @@ namespace Psl\Tests\Benchmark\Type; +use PhpBench\Attributes\Groups; use Psl\Tests\Benchmark\Type\Asset\ExplicitStringableObject; use Psl\Tests\Benchmark\Type\Asset\ImplicitStringableObject; use Psl\Type; @@ -11,6 +12,7 @@ /** * @extends GenericTypeBench> */ +#[Groups(['type'])] final class StringTypeBench extends GenericTypeBench { /** diff --git a/tests/benchmark/Type/VecTypeBench.php b/tests/benchmark/Type/VecTypeBench.php index 3f899cf7..e33687af 100644 --- a/tests/benchmark/Type/VecTypeBench.php +++ b/tests/benchmark/Type/VecTypeBench.php @@ -5,12 +5,14 @@ namespace Psl\Tests\Benchmark\Type; use ArrayIterator; +use PhpBench\Attributes\Groups; use Psl\Type; use Psl\Vec; /** * @extends GenericTypeBench>> */ +#[Groups(['type'])] final class VecTypeBench extends GenericTypeBench { /** @@ -54,7 +56,7 @@ public function provideHappyPathMatches(): array } /** - * @return array>, value: array}> + * @return array>, value: array}> * * @psalm-suppress MissingThrowsDocblock this block should never throw */