如何实现Promise.race效果？

[he426100]

• 需求
  并行多个任务，取最快返回的结果或抛出的异常，剩下的丢弃
  - 实现思路
  \Swow\Selector 可以做到取最快结果这个功能，参考examples下代码写一个promsie_race方法
  - 问题
  可以拿到最快的结果，但是剩下的全部会阻塞，造成死锁。请问应该怎么做？
• 经dixyes和twose指点，代码如下

/**
 * @param array $callbacks
 * @param int $timeout 超时
 * @param bool $throw 是否抛出异常
 * @return mixed
 */
function promise_race(array $callbacks, int $timeout = -1, bool $throw = true): mixed
{
    $coroutines = [];
    defer(static function () use (&$coroutines) {
        foreach ($coroutines as $coroutine) {
            if ($coroutine->isExecuting()) {
                $coroutine->kill();
            }
        }
    });
    $channel = new Channel();
    foreach ($callbacks as $callback) {
        $coroutines[] = Coroutine::run(static function () use ($channel, $callback) {
            try {
                $channel->push($callback());
            } catch (\Throwable $e) {
                $channel->push($e);
            }
        });
    }
    try {
        return $channel->pop($timeout);
    } catch (\Throwable $e) {
        if ($throw) {
            throw $e;
        }
    }
    return false;
}

[dixyes]

defer个Coroutine::run来回收他们就行了

[he426100]

defer个Coroutine::run来回收他们就行了

回收是指杀死协程吗？

[dixyes]

你可以杀死它们 也可以正常回收（pop/push一下channel），取决于你具体的要求

[twose]

use Swow\Channel;
use Swow\Coroutine;

/**
 * @param array $callbacks
 * @param int $timeout 超时
 * @param bool $throw 是否抛出异常
 * @return mixed
 */
function promise_race(array $callbacks, int $timeout = -1, bool $throw = true): mixed
{
    $coroutines = [];
    try {
        $channel = new Channel();
        foreach ($callbacks as $callback) {
            $coroutines[] = Coroutine::run(static function () use ($channel, $callback) {
                $channel->push($callback());
            });
        }
        $result = $channel->pop();
    } catch (\Throwable $e) {
        if ($throw) {
            throw $e;
        }
        $result = false;
    }
    foreach ($coroutines as $coroutine) {
        if ($coroutine->isExecuting()) {
            $coroutine->kill();
        }
    }
    return $result;
}

$r = promise_race([
    function () {
        usleep(3_000);
        echo "Never here 3\n";
        return 3;
    },
    function () {
        usleep(1_000);
        echo "I am here 1\n";
        return 1;
    },
    function () {
        usleep(2_000);
        echo "Never here 2\n";
        return 2;
    },
]);
var_dump($r);

I am here 1
int(1)

好像不需要 Selector

[he426100]

当有协程抛出异常时会出问题，try catch没起作用，会报错，比如改一下测试代码

$r = promise_race([
    function () {
        usleep(3_000);
        echo "Never here 3\n";
        return 3;
    },
    function () {
        usleep(1_000);
        throw new \Exception("I am here 1\n");
    },
    function () {
        usleep(2_000);
        echo "Never here 2\n";
        return 2;
    },
], -1 , false);
var_dump($r);

结果

Warning: [Fatal error in R3] Uncaught Exception: I am here 1
 in /workspaces/git/learn-swow/promise.php:19
Stack trace:
#0 /workspaces/git/learn-swow/promise.php(83): {closure}()
#1 [internal function]: {closure}()
#2 {main}
  thrown in /workspaces/git/learn-swow/promise.php on line 19
Never here 2
int(2)

[he426100]

测试了下，try无法捕获协程内部的异常，这就跟try无法捕获Promise一个道理吧？

[twose]

那是肯定的，都不在一个栈上。你可以在每个协程内部都 try catch，然后用 Channel 来传递异常信息。

[twose]

use Swow\Channel;
use Swow\Coroutine;

/**
 * @param array $callbacks
 * @param int $timeout 超时
 * @param bool $throw 是否抛出异常
 * @return mixed|Throwable|false
 */
function promise_race(array $callbacks, int $timeout = -1, bool $throw = true): mixed
{
    $coroutines = [];
    $channel = new Channel();
    foreach ($callbacks as $callback) {
        $coroutines[] = Coroutine::run(static function () use ($channel, $callback) {
            try {
                $result = $callback();
            } catch (Throwable $result) {
            }
            $channel->push($result);
        });
    }
    $result = $channel->pop();
    if ($result instanceof Throwable) {
        if ($throw) {
            throw $result;
        } else {
            $result = false;
        }
    }
    foreach ($coroutines as $coroutine) {
        if ($coroutine->isExecuting()) {
            $coroutine->kill();
        }
    }
    return $result;
}

$r = promise_race([
    function () {
        usleep(3_000);
        echo "Never here 3\n";
        return 3;
    },
    function () {
        usleep(1_000);
        echo "I am here 1\n";
        return 1;
    },
    function () {
        usleep(2_000);
        echo "Never here 2\n";
        return 2;
    },
]);
var_dump($r);

$r = promise_race([
    function () {
        usleep(3_000);
        echo "Never here 3\n";
        return 3;
    },
    function () {
        usleep(1_000);
        throw new \Exception("I am here 1\n");
    },
    function () {
        usleep(2_000);
        echo "Never here 2\n";
        return 2;
    },
], -1 , false);
var_dump($r);

这样就行了

本来以为你想捕获协程创建失败之类的错误，没想到你是想捕获里面的。