I have started learning Go recently, and when I got to goroutines -- one of Go's strongest features -- I got interested in concurrent and parallel execution in PHP. I knew that PHP has Fibers, but I have never used them, nor have I seen them in any production code, so I decided to dig a bit deeper and learn more about them.

Fibers

Fibers were introduced in PHP 8.1, and they give PHP a low-level way to pause and resume call stacks, which can be used to build cooperative concurrency.

Fibers represent full-stack, interruptible functions. Fibers may be suspended from anywhere in the call-stack, pausing execution within the fiber until the fiber is resumed at a later time.

What this means is that fibers can execute a piece of code, suspend it and allow the thread to execute other pieces of code, and come back later to continue working on the original piece of code.

The main methods of the Fiber class are start, suspend, and resume. Conceptually:

• start() begins the fiber and runs until the fiber suspends or finishes. If the fiber suspends, start() returns the value passed to Fiber::suspend().
• Fiber::suspend() pauses the currently running fiber and returns a value to the caller.
• resume() continues the paused fiber and sends a value back into Fiber::suspend().

A small example:

<?php

// Create a fiber. The callback runs only when start() is called.
$fiber = new Fiber(function () {
    echo "Starting..." . PHP_EOL;

    // Suspend pauses the fiber and returns "Suspended!" to the caller.
    // When the fiber is resumed, suspend() returns the value passed to resume().
    $value = Fiber::suspend('Suspended!');

    echo "Resumed!" . PHP_EOL;

    return "Resumed with {$value}";
});

// start() begins the fiber and runs until the first suspend().
// It returns the value passed to Fiber::suspend().
$result = $fiber->start();

echo $result . PHP_EOL; // Suspended!

// resume() continues the fiber after Fiber::suspend().
// The value passed to resume() becomes the return value of Fiber::suspend().
$fiber->resume('ok');

// Once the fiber has finished, getReturn() gives the fiber's final return value.
echo $fiber->getReturn() . PHP_EOL; // Resumed with ok

While the fiber is suspended, it is possible to work on something unrelated to it, so we can have something like this:

<?php

$fiber = new Fiber(function () {
    echo "Fiber: starting work" . PHP_EOL;

    $value = Fiber::suspend('waiting for external result');

    echo "Fiber: resumed with {$value}" . PHP_EOL;

    return "Fiber: done";
});

echo "Main: starting fiber" . PHP_EOL;

$result = $fiber->start();

echo "Main: fiber suspended with message: {$result}" . PHP_EOL;

for ($i = 1; $i <= 3; $i++) {
    echo "Main: doing other work {$i}" . PHP_EOL;
    sleep(1);
}

echo "Main: resuming fiber" . PHP_EOL;

$fiber->resume('API response');

echo $fiber->getReturn() . PHP_EOL;

This outputs:

Main: starting fiber
Fiber: starting work
Main: fiber suspended with message: waiting for external result
Main: doing other work 1
Main: doing other work 2
Main: doing other work 3
Main: resuming fiber
Fiber: resumed with API response
Fiber: done

This example nicely shows the concurrent nature of the fibers. They are not executed in parallel with the main code -- rather, the main code starts the fiber, which starts to execute. It is up to the fiber to suspend itself, and the main code can resume it. If Fiber::suspend() is called when no fiber is currently running, PHP throws an exception.

Using fibers with Hobby

The Fiber class API and the example from the official PHP docs have made it a bit hard to understand exactly how fibers work and what their purpose is, so I have decided to create a bit more complex pseudo real-life use case for them. In it, we will use the Hobby experiment that I have written about in the last blog.

So, the scenario is this:

We need to dispatch 10 hobbies, and each one is calling an external API that fetches us some company-related data. The API returns immediately, and fetching data is done asynchronously on their side. We want to send the request, wait for the data, and log the data. We will use hobbies that are using fibers, so while one hobby is waiting for its result to become available, the hobbyist can continue running other hobbies. When the result is ready, the suspended hobby resumes and logs the data.

We will first create an interface for tasks that fibers can wait for:

namespace src\Contracts;

interface Awaitable
{
    public function ready(): bool;

    public function result(): mixed;

    public function nextCheckAt(): float;
}

And we will have an attribute that will be used by hobbies that are executed concurrently:

<?php

declare(strict_types=1);

namespace src\Attributes;

use Attribute;

#[Attribute(Attribute::TARGET_CLASS)]
final readonly class ExecuteConcurrently
{
}

In the Hobbyist::process method, we will check if the hobby runs concurrently and schedule it using CooperativeExecutor. At this point, the hobby has not started yet. It is only stored in the executor. The actual execution happens later when Hobbyist calls advanceConcurrentWork(), which calls CooperativeExecutor::advance().

if ($this->runsInFiber($hobby)) {
    $this->cooperativeExecutor->schedule($hobby, $context);

    return;
}

private function runsInFiber(object $hobby): bool
{
    return (new ReflectionClass($hobby))->getAttributes(ExecuteConcurrently::class) !== [];
}

CooperativeExecutor is a class that handles concurrent jobs. When schedule is called, CooperativeExecutor creates a new task:

public function schedule(Hobby $hobby, mixed $context = null): void
{
    $this->tasks[] = $this->makeTask($hobby, $context);
}

private function makeTask(Hobby $hobby, mixed $context = null): Task
{
    return new Task(
        fiber: new Fiber(static fn() => $hobby->handle()),
        runAt: microtime(true),
        waitingOn: null,
        context: $context,
    );
}

We can handle those tasks in the run method, which is the entry point for the Hobbyist class:

public function run(string $queue = 'default'): void
{
    while ($this->running) {
        $this->prepareWorkerLoop();

        if ($this->advanceConcurrentWork()) {
            continue;
        }

        $payload = $this->pullNextPayload($queue);

        if ($payload !== null) {
            $this->process($payload);
            continue;
        }

        $this->idle();
    }
}


private function advanceConcurrentWork(): bool
{
    $advanceResult = $this->cooperativeExecutor->advance();
    $this->handleConcurrentAdvanceResult($advanceResult);

    return ! $advanceResult->isIdle();
}

advance() is responsible for starting/resuming a hobby, handling completions/failure, and parking suspended hobbies.

• It looks for the next task whose runAt time has arrived.

• If no task is ready to run, it returns: AdvanceResult::idle()

• If the task is already started and still waiting on an awaitable, it reschedules the task and returns: AdvanceResult::progressed()

• If the task has never started, it starts the fiber: $task->fiber->start()

• If the task was previously suspended, it resumes the fiber: $this->resumeTask($task)

• When resuming, it checks the awaitable:

  • If it timed out, it throws an exception into the fiber.
  • If it is ready, it passes the awaitable result back into the fiber.

• If the fiber throws an exception, the task is removed from the executor and it returns: AdvanceResult::failed($task, $exception)

• If the fiber finished successfully, the task is removed from the executor and it returns: AdvanceResult::completed($task)

• If the fiber suspended again, it stores the new thing the fiber is waiting on and calculates when it should be checked next.

We will also create a simpler wrapper class for suspending fibers, which will be used in hobbies:

<?php

declare(strict_types=1);

namespace src;

use Fiber;
use src\Contracts\Awaitable;

final class Async
{
    public static function await(Awaitable $awaitable, ?int $timeoutSeconds = null): mixed
    {
        $now = microtime(true);

        return Fiber::suspend([
            'awaitable' => $awaitable,
            'timeoutAt' => $timeoutSeconds === null ? null : $now + max(0, $timeoutSeconds),
        ]);
    }

    public static function suspendFor(float $seconds): void
    {
        self::await(new TimerAwaitable($seconds));
    }
}

Finally, let's create a hobby that fetches and logs the data:

<?php

declare(strict_types=1);

namespace demos\concurrent;

use src\Async;
use src\Attributes\ExecuteConcurrently;
use src\Attributes\OnQueue;
use src\Contracts\Hobby;

#[ExecuteConcurrently]
readonly class CompanyLookupHobby implements Hobby
{
    public function __construct(
        private string $company,
    ) {}

    public function handle(): void
    {
        $this->log('requesting company profile');

        $profile = Async::await(
            new FakeCompanyApiRequest('profile', $this->company),
            timeoutSeconds: 5,
        );

        $this->log('received profile ' . json_encode($profile, JSON_THROW_ON_ERROR));
        $this->log('requesting risk score');

        $risk = Async::await(
            new FakeCompanyApiRequest('risk-score', $this->company),
            timeoutSeconds: 5,
        );

        $this->log('received risk score ' . json_encode($risk, JSON_THROW_ON_ERROR));
    }

    private function log(string $message): void
    {
        echo sprintf("[%s] %s: %s\n", date('H:i:s'), $this->company, $message);
    }
}

<?php

declare(strict_types=1);

namespace demos\concurrent;

use src\Contracts\Awaitable;

final readonly class FakeCompanyApiRequest implements Awaitable
{
    private float $readyAt;

    public function __construct(
        private string $operation,
        private string $company,
    ) {
        $this->readyAt = microtime(true) + match ($this->operation) {
            'profile' => 0.75,
            'risk-score' => 1.25,
            default => 1.0,
        };
    }

    public function ready(): bool
    {
        return microtime(true) >= $this->readyAt;
    }

    public function result(): array
    {
        return match ($this->operation) {
            'profile' => [
                'company' => $this->company,
                'industry' => 'software',
            ],
            'risk-score' => [
                'company' => $this->company,
                'score' => 17,
            ],
            default => ['ok' => true],
        };
    }

    public function nextCheckAt(): float
    {
        return $this->readyAt;
    }
}

What is really important to understand here is that FakeCompanyApiRequest is not blocking the execution, and it is therefore possible to use fibers because a hobby can suspend before the fake API result is ready, and the hobbyist can run other hobbies in the meanwhile.

If we had something like:

$result = $client->blockingRequest();

fibers wouldn't mean anything, as PHP needs to be able to suspend the fiber.

The TLDR version of how this hobby works would be something like this:

• CompanyLookupHobby::handle() starts.
• It creates FakeCompanyApiRequest.
• Async::await() calls Fiber::suspend(...).
• The executor receives the awaitable from start() or resume().
• The executor stores it in Task::$waitingOn.
• The executor calculates Task::$runAt from nextCheckAt().
• The worker loop continues and runs other hobbies.
• Later, the executor sees the awaitable is ready.
• It calls result().
• It passes that result into $fiber->resume($result).
• The suspended Async::await() returns that value.
• The hobby continues and logs the data.

Running demo

I have created two scenarios, one where "normal" hobbies are used, and the other where fibers are used. Each scenario dispatches and executes 10 hobbies that take ~2 seconds to finish. We can run them:

# terminal 1
php demos/concurrent/hobbyist.php

# terminal 2
php demos/concurrent/dispatch.php

and:

# terminal 1
php demos/sequential-api/hobbyist.php

# terminal 2
php demos/sequential-api/dispatch.php

We can see from the logs that sequential execution takes around 20 seconds:

[18:10:37] acme: requesting company profile
[18:10:37] acme: received profile {"company":"acme","industry":"software"}
[18:10:37] acme: requesting risk score
[18:10:39] acme: received risk score {"company":"acme","score":17}
[18:10:39] ✓ demos\sequentialApi\CompanyLookupHobby succeeded (attempt 1/3)

...

[18:10:55] vandelay: requesting risk score
[18:10:57] vandelay: received risk score {"company":"vandelay","score":17}
[18:10:57] ✓ demos\sequentialApi\CompanyLookupHobby succeeded (attempt 1/3)

and concurrent execution takes ~2 seconds:

[18:10:39] acme: requesting company profile
[18:10:39] globex: requesting company profile
[18:10:39] initech: requesting company profile

...

[18:10:41] soylent: received risk score {"company":"soylent","score":17}
[18:10:41] ✓ demos\concurrent\CompanyLookupHobby succeeded (attempt 1/3)
[18:10:41] vandelay: received risk score {"company":"vandelay","score":17}
[18:10:41] ✓ demos\concurrent\CompanyLookupHobby succeeded (attempt 1/3)

Conclusion

Our demo simulates an external API that accepts work immediately and completes it later. In a real system, this could be an API that returns a job ID, a request handle, or a polling endpoint. The important part is that PHP gets control back quickly, so the hobby can call Async::await() and suspend.

This is not a replacement for normal blocking work, and it will not make a blocking API client concurrent by itself. But for async or pollable external work, it is a useful pattern worth keeping in mind.

The code can be found on my GitHub page.