mirror of
https://github.com/appwrite/appwrite.git
synced 2026-05-26 13:51:13 +00:00
Rewrite SwoolePromise with callback-based resolution
Replace busy-waiting in then() with proper callback queuing: - Store waiting callbacks in array when promise is pending - When promise resolves/rejects, process all waiting callbacks - Run callbacks in coroutines for proper async execution - This eliminates deadlocks from busy-wait polling Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Jake Barnby
co-authored by
Claude Opus 4.5
Claude Opus 4.5
parent
cd75f17815
commit
80643faacc
@@ -2,13 +2,30 @@
|
||||
|
||||
namespace Appwrite\Promises;
|
||||
|
||||
use Swoole\Coroutine\Channel;
|
||||
|
||||
class Swoole extends Promise
|
||||
{
|
||||
/**
|
||||
* Callbacks waiting for this promise to settle
|
||||
* Each entry is [Promise, callable|null, callable|null]
|
||||
*
|
||||
* @var array<array{self, callable|null, callable|null}>
|
||||
*/
|
||||
protected array $waiting = [];
|
||||
|
||||
public function __construct(?callable $executor = null)
|
||||
{
|
||||
parent::__construct($executor);
|
||||
if ($executor === null) {
|
||||
return;
|
||||
}
|
||||
|
||||
$resolve = function ($value) {
|
||||
$this->doResolve($value);
|
||||
};
|
||||
$reject = function ($reason) {
|
||||
$this->doReject($reason);
|
||||
};
|
||||
|
||||
$this->execute($executor, $resolve, $reject);
|
||||
}
|
||||
|
||||
protected function execute(
|
||||
@@ -25,76 +42,184 @@ class Swoole extends Promise
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal resolve that triggers waiting callbacks
|
||||
*/
|
||||
protected function doResolve(mixed $value): void
|
||||
{
|
||||
if ($this->state !== self::STATE_PENDING) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Handle thenable values
|
||||
if (\is_object($value) && \method_exists($value, 'then')) {
|
||||
$value->then(
|
||||
fn($v) => $this->doResolve($v),
|
||||
fn($r) => $this->doReject($r)
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
$this->result = $value;
|
||||
$this->state = self::STATE_FULFILLED;
|
||||
$this->processWaiting();
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal reject that triggers waiting callbacks
|
||||
*/
|
||||
protected function doReject(mixed $reason): void
|
||||
{
|
||||
if ($this->state !== self::STATE_PENDING) {
|
||||
return;
|
||||
}
|
||||
|
||||
$this->result = $reason;
|
||||
$this->state = self::STATE_REJECTED;
|
||||
$this->processWaiting();
|
||||
}
|
||||
|
||||
/**
|
||||
* Process all waiting callbacks
|
||||
*/
|
||||
protected function processWaiting(): void
|
||||
{
|
||||
foreach ($this->waiting as [$promise, $onFulfilled, $onRejected]) {
|
||||
$callback = $this->state === self::STATE_FULFILLED ? $onFulfilled : $onRejected;
|
||||
|
||||
if ($callback === null) {
|
||||
// Pass through the value/reason
|
||||
if ($this->state === self::STATE_FULFILLED) {
|
||||
$promise->doResolve($this->result);
|
||||
} else {
|
||||
$promise->doReject($this->result);
|
||||
}
|
||||
} else {
|
||||
// Run callback in a coroutine
|
||||
\go(function () use ($promise, $callback) {
|
||||
try {
|
||||
$result = $callback($this->result);
|
||||
$promise->doResolve($result);
|
||||
} catch (\Throwable $e) {
|
||||
$promise->doReject($e);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
$this->waiting = [];
|
||||
}
|
||||
|
||||
/**
|
||||
* Override then to use callback-based approach instead of busy-waiting
|
||||
*/
|
||||
public function then(
|
||||
?callable $onFulfilled = null,
|
||||
?callable $onRejected = null
|
||||
): self {
|
||||
$promise = new self();
|
||||
|
||||
if ($this->state === self::STATE_PENDING) {
|
||||
// Queue the callbacks for later
|
||||
$this->waiting[] = [$promise, $onFulfilled, $onRejected];
|
||||
} else {
|
||||
// Already settled, process immediately
|
||||
$callback = $this->state === self::STATE_FULFILLED ? $onFulfilled : $onRejected;
|
||||
|
||||
if ($callback === null) {
|
||||
if ($this->state === self::STATE_FULFILLED) {
|
||||
$promise->doResolve($this->result);
|
||||
} else {
|
||||
$promise->doReject($this->result);
|
||||
}
|
||||
} else {
|
||||
\go(function () use ($promise, $callback) {
|
||||
try {
|
||||
$result = $callback($this->result);
|
||||
$promise->doResolve($result);
|
||||
} catch (\Throwable $e) {
|
||||
$promise->doReject($e);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return $promise;
|
||||
}
|
||||
|
||||
/**
|
||||
* Override resolve to use internal method
|
||||
*/
|
||||
public function resolve(mixed $value): self
|
||||
{
|
||||
$this->doResolve($value);
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Override reject to use internal method
|
||||
*/
|
||||
public function reject(mixed $reason): self
|
||||
{
|
||||
$this->doReject($reason);
|
||||
return $this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a promise that completes when all passed in promises complete.
|
||||
*
|
||||
* @param iterable $promisesOrValues Array of promises and/or plain values
|
||||
* @return Promise
|
||||
* @return self
|
||||
*/
|
||||
public static function all(iterable $promisesOrValues): Promise
|
||||
public static function all(iterable $promisesOrValues): self
|
||||
{
|
||||
return self::create(function (callable $resolve, callable $reject) use ($promisesOrValues) {
|
||||
$result = [];
|
||||
$error = null;
|
||||
$promiseCount = 0;
|
||||
$key = 0;
|
||||
$promisesOrValues = \is_array($promisesOrValues)
|
||||
? $promisesOrValues
|
||||
: \iterator_to_array($promisesOrValues);
|
||||
|
||||
// First pass: count promises and store plain values
|
||||
$promiseKeys = [];
|
||||
foreach ($promisesOrValues as $promiseOrValue) {
|
||||
if ($promiseOrValue instanceof Promise) {
|
||||
$promiseKeys[] = $key;
|
||||
$promiseCount++;
|
||||
} else {
|
||||
$result[$key] = $promiseOrValue;
|
||||
}
|
||||
$key++;
|
||||
}
|
||||
$total = \count($promisesOrValues);
|
||||
$promise = new self();
|
||||
|
||||
// If no promises, resolve immediately
|
||||
if ($promiseCount === 0) {
|
||||
if ($total === 0) {
|
||||
$promise->doResolve([]);
|
||||
return $promise;
|
||||
}
|
||||
|
||||
$count = 0;
|
||||
$result = [];
|
||||
$rejected = false;
|
||||
|
||||
$resolveIfDone = static function () use (&$count, $total, &$result, &$rejected, $promise): void {
|
||||
if (!$rejected && $count === $total) {
|
||||
\ksort($result);
|
||||
$resolve($result);
|
||||
return;
|
||||
$promise->doResolve($result);
|
||||
}
|
||||
};
|
||||
|
||||
$channel = new Channel($promiseCount);
|
||||
$key = 0;
|
||||
|
||||
foreach ($promisesOrValues as $promiseOrValue) {
|
||||
if ($promiseOrValue instanceof Promise) {
|
||||
$currentKey = $key;
|
||||
$promiseOrValue->then(
|
||||
function ($value) use ($currentKey, &$result, $channel) {
|
||||
$result[$currentKey] = $value;
|
||||
$channel->push(true);
|
||||
return $value;
|
||||
},
|
||||
function ($err) use ($channel, &$error) {
|
||||
$channel->push(true);
|
||||
if ($error === null) {
|
||||
$error = $err;
|
||||
}
|
||||
foreach ($promisesOrValues as $index => $promiseOrValue) {
|
||||
if ($promiseOrValue instanceof Promise) {
|
||||
$result[$index] = null;
|
||||
$promiseOrValue->then(
|
||||
static function ($value) use (&$result, $index, &$count, $resolveIfDone) {
|
||||
$result[$index] = $value;
|
||||
++$count;
|
||||
$resolveIfDone();
|
||||
return $value;
|
||||
},
|
||||
static function ($error) use (&$rejected, $promise) {
|
||||
if (!$rejected) {
|
||||
$rejected = true;
|
||||
$promise->doReject($error);
|
||||
}
|
||||
);
|
||||
}
|
||||
$key++;
|
||||
}
|
||||
);
|
||||
} else {
|
||||
$result[$index] = $promiseOrValue;
|
||||
++$count;
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for all promises
|
||||
$remaining = $promiseCount;
|
||||
while ($remaining-- > 0) {
|
||||
$channel->pop();
|
||||
}
|
||||
$channel->close();
|
||||
$resolveIfDone();
|
||||
|
||||
if ($error !== null) {
|
||||
$reject($error);
|
||||
return;
|
||||
}
|
||||
|
||||
\ksort($result);
|
||||
$resolve($result);
|
||||
});
|
||||
return $promise;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user