PHP code example of kode / runtime

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

kode / runtime example snippets

use Kode\Runtime\Runtime;
use Kode\Runtime\RuntimeEnvironment;

echo "当前运行环境: " . Runtime::environment()->label();
// 输出: SWOOLE | SWOW | FIBER | PROCESS | THREAD | CONSOLE | CLI

var_dump(Runtime::supportsConcurrency()); // 是否具备真正的并发能力

// 枚举方式列举可用环境
foreach (RuntimeEnvironment::available() as $env) {
    echo $env->label() . PHP_EOL;
}

use Kode\Runtime\Runtime;

Runtime::async(function () {
    echo "协程开始\n";
    Runtime::sleep(1.5);
    echo "协程结束\n";
});

echo "主流程继续执行\n";
Runtime::wait();

use Kode\Runtime\Runtime;

$results = Runtime::run(function () {
    return Runtime::parallel([
        'a' => fn () => usleep(100_000) ?: 'A',
        'b' => fn () => usleep(200_000) ?: 'B',
        'c' => fn () => usleep(50_000)  ?: 'C',
    ]);
});

print_r($results); // ['a' => 'A', 'b' => 'B', 'c' => 'C'] 并发执行

use Kode\Runtime\Runtime;
use Kode\Runtime\ChannelInterface;

$channel = Runtime::createChannel(1);

Runtime::async(function () use ($channel) {
    $channel->push("Hello from coroutine");
});

Runtime::async(function () use ($channel) {
    Runtime::sleep(0.5);
    $data = $channel->pop();
    echo "收到: $data\n";

    // 带超时的非阻塞尝试
    $miss = $channel->pop(ChannelInterface::TIMEOUT_NONE);
    var_dump($channel->isTimeout()); // true
});

Runtime::wait();

use Kode\Runtime\Runtime;

Runtime::async(function () {
    $fp = fopen('/tmp/test.txt', 'w');

    Runtime::defer(function () use ($fp) {
        fclose($fp);
    });

    fwrite($fp, "Hello");
    Runtime::sleep(1);
});

use Kode\Runtime\Runtime;

Runtime::run(function () {
    $wg = Runtime::waitGroup();

    foreach (range(1, 5) as $i) {
        $wg->run(static function () use ($i): int {
            Runtime::sleep(0.1);
            return $i * $i;
        });
    }

    $wg->wait();

    print_r($wg->results());   // [1, 4, 9, 16, 25]（按派发顺序）
    echo $wg->hasErrors() ? '有任务失败' : '全部成功';
});

use Kode\Runtime\Runtime;

// 单飞：昂贵的初始化只算一次
$once = Runtime::once();
$value = Runtime::run(fn () => $once->do(fn () => expensiveLookup()));

// 竞速：最快者胜出
$fastest = Runtime::run(fn () => Runtime::race(
    fn () => fetchFromCache(),   // 可能更快
    fn () => fetchFromApi(),     // 可能更慢
));

// 选择：谁先有数据用谁
[$channel, $value] = Runtime::run(function () {
    $a = Runtime::createChannel();
    $b = Runtime::createChannel();
    Runtime::async(fn () => $a->push('from-a'));
    $r = Runtime::select($a, $b);
    return [$r['channel'], $r['value']];
});

use Kode\Runtime\Runtime;

// 最多 2 个并发，其余排队等待
$results = Runtime::run(function () {
    $sem = Runtime::semaphore(2);

    return Runtime::parallel(array_map(
        static fn (int $i) => static fn () => $sem->run(static fn () => doWork($i)),
        range(0, 9)
    ));
});

use Kode\Runtime\Runtime;
use Kode\Runtime\RuntimeEnvironment;

Runtime::setEnvironment(RuntimeEnvironment::Process);

$pid = Runtime::fork(function () {
    echo "子进程 PID: " . getmypid() . "\n";
    Runtime::sleep(1);
});

Runtime::wait();

use Kode\Runtime\RuntimeCommand;
use Kode\Console\Input;
use Kode\Console\Output;

class AsyncTaskCommand extends RuntimeCommand
{
    public function __construct()
    {
        parent::__construct('async:task', '异步任务示例');
    }

    public function fire(Input $in, Output $out): int
    {
        $this->setOutput($out);

        $this->async(function () {
            $this->info('任务开始');
            $this->sleep(1);
            $this->success('任务完成');
        });

        $this->wait();
        return 0;
    }
}

use Kode\Runtime\FiberScheduler;

FiberScheduler::instance()->setErrorHandler(
    static fn (Throwable $e) => $logger->error('协程失败：' . $e->getMessage())
);

Runtime::reset();   // = 适配器置空 + FiberScheduler::resetInstance()

Runtime::setEnvironment(RuntimeEnvironment::Swoole);

Runtime::async(static fn () => Co\sleep(0.02) && saveResult());
Runtime::wait();          // v3.5.0 起：返回时协程确已结束

use Kode\Runtime\Runtime;

Runtime::version();         // '3.6.0'
Runtime::VERSION;           // 同一常量的字面量

final class Runtime
{
    // 运行环境名称（大写标签）
    public static function getName(): string;

    // 运行环境枚举
    public static function environment(): RuntimeEnvironment;

    // 是否具备真正的并发能力
    public static function supportsConcurrency(): bool;

    // 异步执行（协程/进程/线程，取决于当前运行时）
    public static function async(callable $callback): mixed;

    // 以入口函数方式运行并等待全部异步任务结束
    public static function run(callable $main): mixed;

    // 并发执行多个任务并按原始键收集结果
    public static function parallel(iterable $tasks): array;

    // 休眠（协程环境让出执行权）
    public static function sleep(float $seconds): void;

    // 创建通道
    public static function createChannel(int $capacity = 0): ChannelInterface;

    // 注册当前作用域退出时执行的回调（LIFO）
    public static function defer(callable $callback): void;

    // 等待所有异步任务完成
    public static function wait(): void;

    // 创建等待组（WaitGroup）
    public static function waitGroup(?RuntimeInterface $runtime = null): WaitGroup;

    // 创建单飞原语（Once）
    public static function once(?RuntimeInterface $runtime = null): Once;

    // 竞速：并发执行多个任务，返回第一个完成（成功/失败）的结果
    public static function race(callable ...$tasks): mixed;

    // 选择：等待多个通道中第一个就绪者，返回 ['channel' => ChannelInterface, 'value' => mixed]
    public static function select(ChannelInterface ...$channels): array;

    // 创建信号量（Semaphore），最多 $permits 个并发
    public static function semaphore(int $permits, ?RuntimeInterface $runtime = null): Semaphore;

    // 创建子进程（仅 PCNTL 环境）；回收由调用方 pcntl_waitpid() 负责，
    // 子进程侧会自动丢弃继承自父进程的 defer 作用域与调度器状态
    public static function fork(callable $callback): int;

    // 设置特定运行环境（接受枚举或字符串）
    public static function setEnvironment(RuntimeEnvironment|string $environment): void;

    // 直接注入适配器（便于测试/DI），null 恢复自动探测
    public static function setAdapter(?RuntimeInterface $adapter): void;

    // 获取当前适配器
    public static function adapter(): RuntimeInterface;

    // 重置适配器（测试用）
    public static function reset(): void;
}

enum RuntimeEnvironment: string
{
    case Swoole  = 'swoole';
    case Swow    = 'swow';
    case Fiber   = 'fiber';
    case Process = 'process';
    case Thread  = 'thread';
    case Console = 'console';
    case Cli     = 'cli';

    public function label(): string;            // 大写显示名
    public function isAvailable(): bool;        // 当前环境是否可用
    public function supportsConcurrency(): bool;// 是否真正并发
    public function unavailableMessage(): string;

    public static function resolve(string|self $environment): self;
    public static function detect(): self;      // Swoole→Swow→Fiber→Cli
    public static function available(): array;  // 所有可用环境
}

final class RuntimeAdapterFactory
{
    public const string ENV_SWOOLE  = 'swoole';
    public const string ENV_SWOW    = 'swow';
    public const string ENV_FIBER   = 'fiber';
    public const string ENV_PROCESS = 'process';
    public const string ENV_THREAD  = 'thread';
    public const string ENV_CLI     = 'cli';
    public const string ENV_CONSOLE = 'console';

    public static function create(RuntimeEnvironment|string|null $environment = null): RuntimeInterface;
    public static function createConcurrent(): RuntimeInterface;     // 自动探测最强并发
    public static function createForEnvironment(RuntimeEnvironment|string $environment): RuntimeInterface;
    public static function availableEnvironments(): array;
    public static function isAvailable(RuntimeEnvironment|string $environment): bool;
    public static function isSwooleAvailable(): bool;
    public static function isSwowAvailable(): bool;
    public static function isFiberSupported(): bool;
    public static function isConsoleAvailable(): bool;
}

final class FiberScheduler
{
    public static function instance(): self;       // 全局单例
    public static function resetInstance(): void;  // 重置（测试用）；Runtime::reset() 会一并调用

    public function spawn(callable $callback, ?callable $onFinish = null): \Fiber;
    public function yield(): void;
    public function sleep(float $seconds): void;
    public function run(): void;                   // 驱动循环直到全部结束，抛出首个协程异常
    public function setErrorHandler(?callable $handler): void;  // 接收其余无人可抛的协程异常
    public function tick(): bool;
    public function hasPendingWork(): bool;
    public function hasOtherWork(): bool;          // Channel 死锁判定
    public function inFiber(): bool;
    public function count(): int;
}

go(callable $callback): mixed;                 // = Runtime::async()
parallel(iterable $tasks): array;              // = Runtime::parallel()
run(callable $main): mixed;                    // = Runtime::run()
channel(int $capacity = 0): ChannelInterface;  // = Runtime::createChannel()
defer(callable $callback): void;               // = Runtime::defer()
delay(float $seconds): void;                   // = Runtime::sleep()
wait(): void;                                  // = Runtime::wait()
waitGroup(?RuntimeInterface $runtime = null): WaitGroup; // = Runtime::waitGroup()
once(?RuntimeInterface $runtime = null): Once;            // = Runtime::once()
race(callable ...$tasks): mixed;                          // = Runtime::race()
select(ChannelInterface ...$channels): array;             // = Runtime::select()
semaphore(int $permits, ?RuntimeInterface $runtime = null): Semaphore; // = Runtime::semaphore()

abstract class RuntimeCommand extends \Kode\Console\Command
{
    protected function async(callable $callback): mixed;
    protected function run(callable $main): mixed;
    protected function parallel(iterable $tasks): array;
    protected function sleep(float $seconds): void;
    protected function defer(callable $callback): void;
    protected function wait(): void;
    protected function createChannel(int $capacity = 0): ChannelInterface;

    // 输出方法
    protected function info(string $message): void;
    protected function warn(string $message): void;
    protected function error(string $message): void;
    protected function success(string $message): void;
    protected function line(string $text, string $color = ''): void;
    protected function table(array $headers, array $rows): void;
    protected function progress(int $current, int $total, int $width = 50): void;
}

final class WaitGroup
{
    // $runtime 为 null 时使用当前门面运行时
    public function __construct(?RuntimeInterface $runtime = null);

    // 增加待完成任务计数（delta 必须 ≥ 0，否则抛 InvalidArgumentException）
    public function add(int $delta = 1): static;

    // 派生一个异步任务，完成后收集返回值；异常收集到 errors()，不会中断其余任务
    public function run(callable $task): static;

    // 通知一个待完成任务已完成（与 add() 配对，归还计数，避免 wait() 死锁）
    public function done(): static;

    // 阻塞直到所有已派发任务完成（协程/事件循环运行时真正让出执行权）
    public function wait(): void;

    // 返回值，按派发序号排序（与 errors() 的键对应）
    public function results(): array;

    // 失败任务的异常，按派发序号排序
    public function errors(): array;

    public function hasErrors(): bool;
    public function count(): int;   // 尚未完成的任务数量
}

final class Once
{
    // $runtime 为 null 时使用当前门面运行时
    public function __construct(?RuntimeInterface $runtime = null);

    // 执行被包裹的回调（仅一次），返回其结果；并发调用者共享同一结果。
    // 若首次执行抛异常，则后续调用同样抛该异常且不会重新执行（Once 语义）。
    public function do(callable $fn): mixed;

    // 回调是否已经执行过（无论成功或失败）
    public function hasRun(): bool;

    // 重置状态，允许下次调用重新执行回调（适用于长生命周期进程周期性重新初始化）
    public function reset(): void;
}

final class Semaphore
{
    // $permits 为许可数量（必须 ≥ 1，否则抛 InvalidArgumentException）
    // $runtime 为 null 时使用当前门面运行时
    public function __construct(int $permits, ?RuntimeInterface $runtime = null);

    // 获取一个许可（无可用许可则阻塞等待）
    public function acquire(): void;

    // 释放一个许可，使一个等待者继续（释放次数不得超过 capacity()）
    public function release(): void;

    // 在临界区内执行回调（acquire → 执行 → release，异常不泄漏许可）
    public function run(callable $task): mixed;

    public function available(): int;  // 当前可用许可数
    public function capacity(): int;   // 总许可数
}

interface ChannelInterface
{
    public const float TIMEOUT_FOREVER = -1.0;  // 永久等待
    public const float TIMEOUT_NONE    =  0.0;  // 不等待（立即返回）

    public function push(mixed $data, float $timeout = self::TIMEOUT_FOREVER): bool;
    public function pop(float $timeout = self::TIMEOUT_FOREVER): mixed;
    public function getCapacity(): int;
    public function getLength(): int;
    public function isEmpty(): bool;
    public function isFull(): bool;
    public function isTimeout(): bool;
    public function close(): void;
    public function isClosed(): bool;
}