From a0d92e9d0c1acf1d0e00dec9e424b4ce4c1e2a3a Mon Sep 17 00:00:00 2001 From: Valentin Clavreul Date: Mon, 6 Oct 2025 17:19:18 +0200 Subject: [PATCH 1/2] feat: implement an async buffer mechanism --- composer.json | 3 +- src/Buffer/AsyncBuffer.php | 140 ++++++++++++++++++++++++ src/Buffer/AsyncBufferItem.php | 49 +++++++++ src/Buffer/UnresolvedItemsException.php | 18 +++ 4 files changed, 209 insertions(+), 1 deletion(-) create mode 100644 src/Buffer/AsyncBuffer.php create mode 100644 src/Buffer/AsyncBufferItem.php create mode 100644 src/Buffer/UnresolvedItemsException.php diff --git a/composer.json b/composer.json index 516a5f6..870ee0e 100644 --- a/composer.json +++ b/composer.json @@ -21,7 +21,8 @@ }, "require": { "php": "^8.1", - "psr/http-message": "^1.0|^2.0" + "psr/http-message": "^1.0|^2.0", + "psr/log": "^3.0" }, "require-dev": { "phpunit/phpunit": "^10.5", diff --git a/src/Buffer/AsyncBuffer.php b/src/Buffer/AsyncBuffer.php new file mode 100644 index 0000000..8de94bb --- /dev/null +++ b/src/Buffer/AsyncBuffer.php @@ -0,0 +1,140 @@ + */ + private array $buffer = []; + private bool $waitingForIdle = false; + /** @var callable(array): Promise */ + private mixed $flusherGeneratorBuilder; + private ?float $bufferingStartTime = null; + + /** + * @param callable(array): Promise $flusherGeneratorBuilder An awaitable callback that will receive all buffered inputs + * @param int|null $bufferSize A maximum buffer size after which the buffer will get automatically flushed + * @param float|null $bufferingMinWaitSecond Buffering time in second before triggering any flush (on EventLoop idle) + * @param float|null $bufferingMaxWaitSecond Buffering time in second after automatically triggering flush + */ + public function __construct( + callable $flusherGeneratorBuilder, + private readonly EventLoop $eventLoop, + private readonly ?LoggerInterface $logger = null, + private readonly ?int $bufferSize = null, + private readonly ?float $bufferingMinWaitSecond = null, + private readonly ?float $bufferingMaxWaitSecond = null, + ) { + $this->flusherGeneratorBuilder = $flusherGeneratorBuilder; + } + + /** + * Register a set of argument in the Buffer that will be passed on flushed + */ + public function register(mixed ...$args): Promise + { + $asyncBufferItem = new AsyncBufferItem($args, $this->eventLoop->deferred()); + $this->buffer[] = $asyncBufferItem; + + $this->logger?->debug('Registered async request', ['class' => __CLASS__, 'args' => $args]); + + if ($this->bufferingStartTime === null) { + $this->bufferingStartTime = microtime(true); + } + + if (!$this->waitingForIdle) { + $this->waitingForIdle = true; + $this->eventLoop->async($this->awaitFlushing()); + } + + if ($this->shouldFlush()) { + $this->flush(); + } + + return $asyncBufferItem->getPromise(); + } + + /** + * Wait for Event Loop to be idle to trigger a flush + */ + private function awaitFlushing(): \Generator + { + if ($this->bufferingMinWaitSecond !== null && $this->secondsSinceBuffering() < $this->bufferingMinWaitSecond) { + $msToWait = \intval(ceil(($this->bufferingMinWaitSecond - $this->secondsSinceBuffering()) * 1000)); + $waitPromise = $this->eventLoop->delay($msToWait); + $this->logger?->debug('Awaiting event loop with delay', ['class' => __CLASS__, 'delay' => $msToWait]); + } else { + $waitPromise = $this->eventLoop->idle(); + $this->logger?->debug('Awaiting idle event loop', ['class' => __CLASS__]); + } + + yield $waitPromise; + $this->flush(); + $this->waitingForIdle = false; + } + + private function shouldFlush(): bool + { + if ($this->bufferSize !== null && \count($this->buffer) >= $this->bufferSize) { + return true; + } + + if ($this->bufferingMaxWaitSecond !== null && $this->secondsSinceBuffering() >= $this->bufferingMaxWaitSecond) { + return true; + } + + return false; + } + + private function flush(): void + { + $this->logger?->debug('Flushing buffer', ['class' => __CLASS__, 'buffer_size' => \count($this->buffer)]); + + /** + * Capture and flush the current buffer + * @param array $buffer + * @throws UnresolvedItemsException + */ + $wrappedGenerator = function (array $buffer) : \Generator { + try { + yield \call_user_func($this->flusherGeneratorBuilder, $buffer); + + $this->logger?->debug('End flushing buffer', ['class' => __CLASS__]); + } catch (\Throwable $t) { + $this->logger?->warning('Failure to flush buffer', ['class' => __CLASS__, 'throwable' => $t, 'throwable_class' => get_debug_type($t)]); + foreach ($buffer as $item) { + if ($item->isPending()) { + $item->reject($t); + } + } + } + + $pendingItems = []; + foreach ($buffer as $item) { + if ($item->isPending()) { + $pendingItems[] = $item; + } + } + + if (count($pendingItems) > 0) { + throw new UnresolvedItemsException($pendingItems); + } + }; + + $this->eventLoop->async($wrappedGenerator($this->buffer)); + $this->buffer = []; + $this->bufferingStartTime = null; + } + + private function secondsSinceBuffering(): float + { + // $this->bufferingStartTime should never be null at this point + return microtime(true) - $this->bufferingStartTime; + } +} diff --git a/src/Buffer/AsyncBufferItem.php b/src/Buffer/AsyncBufferItem.php new file mode 100644 index 0000000..d2a2d96 --- /dev/null +++ b/src/Buffer/AsyncBufferItem.php @@ -0,0 +1,49 @@ +args; + } + + public function getPromise(): Promise + { + return $this->deferred->getPromise(); + } + + public function isPending(): bool + { + return $this->isPending; + } + + public function resolve(mixed $value): void + { + $this->isPending = false; + $this->deferred->resolve($value); + } + + public function reject(\Throwable $throwable): void + { + $this->isPending = false; + $this->deferred->reject($throwable); + } +} diff --git a/src/Buffer/UnresolvedItemsException.php b/src/Buffer/UnresolvedItemsException.php new file mode 100644 index 0000000..aaa48b4 --- /dev/null +++ b/src/Buffer/UnresolvedItemsException.php @@ -0,0 +1,18 @@ +items; + } +} \ No newline at end of file From 5f03b5a4d2baf69d2319079df3a65808795b7f1a Mon Sep 17 00:00:00 2001 From: Valentin Clavreul Date: Mon, 6 Oct 2025 17:29:02 +0200 Subject: [PATCH 2/2] feat: implement a basic collapsing buffer --- src/Buffer/CollapsingBuffer.php | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) create mode 100644 src/Buffer/CollapsingBuffer.php diff --git a/src/Buffer/CollapsingBuffer.php b/src/Buffer/CollapsingBuffer.php new file mode 100644 index 0000000..92a9b43 --- /dev/null +++ b/src/Buffer/CollapsingBuffer.php @@ -0,0 +1,28 @@ +knownPromises)) { + return $this->knownPromises[$key]; + } + + $promise = $this->buffer->register(...$args); + $this->knownPromises[$key] = $promise; + + return $promise; + } +} \ No newline at end of file