PHPackages                             yangusik/thrun - PHPackages - PHPackages  [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register)

1. [Directory](/)
2. /
3. [Queues & Workers](/categories/queues)
4. /
5. yangusik/thrun

ActiveLibrary[Queues & Workers](/categories/queues)

yangusik/thrun
==============

Async queue worker engine for PHP with native threads + coroutines via TrueAsync.

v0.1.0(4w ago)401[1 issues](https://github.com/YanGusik/thrun/issues)1MITPHPPHP >=8.4

Since May 11Pushed 1w agoCompare

[ Source](https://github.com/YanGusik/thrun)[ Packagist](https://packagist.org/packages/yangusik/thrun)[ RSS](/packages/yangusik-thrun/feed)WikiDiscussions main Synced 1w ago

READMEChangelog (1)Dependencies (1)Versions (2)Used By (1)

Thrun
=====

[](#thrun)

Async queue worker for PHP built on [TrueAsync](https://github.com/true-async) - alternative PHP core that implements true asynchrony by modifying the Zend engine, I/O libraries, database and socket handling.

Goal
----

[](#goal)

The fastest async queue worker for PHP - one worker process that handles both IO-bound and CPU-bound tasks efficiently. Uses real OS threads instead of forked processes, consumes significantly less memory, and aims to outperform Symfony Messenger and Laravel Horizon.

Benchmarks
----------

[](#benchmarks)

Measured on WSL2, 8GB RAM, PHP 8.6 TrueAsync fork:

ScenarioIO throughputCPU throughputStable RSSHorizon 1 worker18/s73/s72 MBHorizon 12 workers210/s514/s949 MBTrueAsync 1x1001,869/s452/s44 MBTrueAsync 12x102,355/s2,059/s54 MBTrueAsync 12x10 uses **17x less RSS** than Horizon 12 workers, **11x more IO throughput**.

Requirements
------------

[](#requirements)

- TrueAsync PHP 8.6+ (`trueasync/php-true-async` Docker image)
- ext-pcntl (for signal handling)
- ext-redis (Edmond's TrueAsync-compatible fork for Redis transport)

Installation
------------

[](#installation)

```
composer require yangusik/thrun
```

Package is in development and may change name.

Quick Start
-----------

[](#quick-start)

```
use Thrun\Envelope\Envelope;
use Thrun\Supervisor\Supervisor;
use Thrun\Supervisor\SupervisorOptions;
use Thrun\Transport\InMemory\InMemoryTransport;
use Thrun\Worker\Worker;
use Thrun\Worker\WorkerOptions;

$transport = new InMemoryTransport();

// Send messages
$transport->send(Envelope::wrap(new SendEmailMessage('user@example.com', 'Hello')));

$supervisor = new Supervisor(
    workerFactory: fn() => new Worker(
        transport: $transport,
        handlers: [
            SendEmailMessage::class => fn(SendEmailMessage $m) => mail($m->to, $m->subject, '...'),
        ],
        options: new WorkerOptions(threads: 2, concurrency: 4),
    ),
    options: new SupervisorOptions(),
);

$supervisor->run();
```

Features
--------

[](#features)

**N OS Threads x M Coroutines**

Each Worker spawns N real OS threads. Each thread runs a TaskGroup of M coroutines. Total concurrency is N \* M. ThreadChannel provides backpressure and blocks when all coroutines are busy.

**Per-Message Timeout (Hard Cancel)**

Attach a TimeoutStamp to any message. If the handler exceeds the limit, it receives a hard cancellation that interrupts blocking operations like `sleep`, `file_get_contents`, or DB queries. `finally` blocks still execute.

```
use Thrun\Envelope\Stamp\TimeoutStamp;

$transport->send(Envelope::wrap(
    new SlowApiCall(),
    new TimeoutStamp(5000), // 5 seconds
));
```

**Retry with Delay**

Attach a RetryStamp with a strategy. Failed messages are retried with a configurable delay.

```
use Thrun\Envelope\Stamp\RetryStamp;
use Thrun\Worker\Retry\ExponentialBackoffStrategy;

$transport->send(Envelope::wrap(
    new SendEmailMessage('user@example.com', 'Hello'),
    new RetryStamp(strategy: new ExponentialBackoffStrategy(1000, 3)),
));
```

Strategies: `NoRetryStrategy`, `FixedDelayStrategy`, `ExponentialBackoffStrategy`.

**Metrics**

Inject a MetricsInterface to track throughput, failures, retries, timeouts, and average processing time.

```
use Thrun\Worker\Metrics\InMemoryMetrics;

$metrics = new InMemoryMetrics();
$worker = new Worker(transport: $transport, handlers: $handlers, metrics: $metrics);

// $metrics->processed, $metrics->failed, $metrics->retried, $metrics->timedOut
// $metrics->averageTime()
```

**Multi-Queue and Scheduling**

One Worker can serve multiple queues simultaneously with pluggable scheduling:

- `RoundRobinStrategy` - equal distribution
- `PriorityStrategy` - weighted credits, proportional distribution

**Dispatch Policies**

Limit concurrency per partition (tenant, user, etc.) via PartitionStamp:

```
use Thrun\Envelope\Stamp\PartitionStamp;
use Thrun\Transport\Policy\MaxConcurrencyPolicy;
use Thrun\Transport\PolicyAwareReceiver;

$receiver = new PolicyAwareReceiver(
    inner: $transport,
    policy: new MaxConcurrencyPolicy(maxPerPartition: 5),
);

$transport->send(Envelope::wrap($msg, new PartitionStamp('tenant-42')));
```

**Graceful Shutdown**

Supervisor handles SIGINT/SIGTERM gracefully. `Worker::stop()` cancels the scope and closes the transport. The producer unblocks from `receive()`. Pending jobs complete or timeout.

**Explicit Acknowledgement**

Handlers can accept an Acknowledger for explicit control:

```
use Thrun\Worker\Acknowledger;

SendEmailMessage::class => function (SendEmailMessage $m, Acknowledger $ack) {
    if ($m->to === 'blocked@example.com') {
        $ack->fail(new \RuntimeException('Blocked'));
        return;
    }
    // process...
    $ack->ack();
}
```

Methods: `$ack->ack()`, `$ack->retry(int $delayMs)`, `$ack->fail(?Throwable)`.

**Redis Transport**

At-least-once delivery via `LMOVE` from ready to processing. Delayed messages use a sorted set (`ZADD`). Reclaims the processing list on startup for crash recovery.

```
use Thrun\Transport\Redis\RedisTransport;
use Thrun\Transport\Redis\RedisConnection;
use Thrun\Serialization\JsonSerializer;

$redis = new \Redis();
$redis->connect('redis', 6379);

$transport = new RedisTransport(
    connection: new RedisConnection($redis, 'thrun:queue'),
    serializer: new JsonSerializer(),
    queue: 'emails',
);
```

**Failure Transport (Dead Letter)**

Exhausted retries can be sent to a separate transport for inspection:

```
$failureTransport = new InMemoryTransport(); // or RedisTransport

$worker = new Worker(
    transport: $transport,
    handlers: $handlers,
    failureTransport: $failureTransport,
);
```

Failed messages carry an ErrorDetailsStamp with the exception class, message, code, and trace.

Examples
--------

[](#examples)

See the `examples/` directory:

- `one_queue.php` - basic single-queue worker
- `two_queue.php` - multi-queue with priority
- `round_robin.php` - round-robin scheduling
- `priority.php` - priority strategy
- `max_concurrency.php` - per-partition concurrency limits
- `retry.php` - retry with fixed delay
- `metrics.php` - live console metrics reporter

Run examples directly if you have TrueAsync PHP installed, or via Docker (see below).

Architecture
------------

[](#architecture)

Detailed architecture: `docs/architecture.md`Development plan: `docs/development-plan.md`

Testing
-------

[](#testing)

Tests use `testo`. No mocking of TrueAsync internals - test with `InMemoryTransport` and real async execution.

Some transport tests require Redis running on `localhost:6379`.

```
vendor/bin/testo
```

Docker
------

[](#docker)

If you want to test this project in Docker, here is an example setup.

**Dockerfile:**

```
FROM trueasync/php-true-async:0.7.0-alpha.9-php8.6

RUN apt-get update && apt-get install -y \
    build-essential \
    autoconf \
    libtool \
    curl \
    git \
    && rm -rf /var/lib/apt/lists/*

# Build and install phpredis from trueasync fork
RUN git clone --depth 1 --branch true-async https://github.com/true-async/phpredis.git /tmp/phpredis \
    && cd /tmp/phpredis \
    && phpize \
    && ./configure \
    && make -j$(nproc) \
    && make install \
    && echo 'extension=redis.so' > /etc/php.d/redis.ini \
    && rm -rf /tmp/phpredis
```

**docker-compose.yml:**

```
services:
  dev:
    build:
      context: .
      dockerfile: Dockerfile
    working_dir: /app
    volumes:
      - .:/app

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
```

License
-------

[](#license)

MIT

###  Health Score

38

—

LowBetter than 83% of packages

Maintenance91

Actively maintained with recent releases

Popularity5

Limited adoption so far

Community9

Small or concentrated contributor base

Maturity41

Maturing project, gaining track record

 Bus Factor1

Top contributor holds 100% of commits — single point of failure

How is this calculated?**Maintenance (25%)** — Last commit recency, latest release date, and issue-to-star ratio. Uses a 2-year decay window.

**Popularity (30%)** — Total and monthly downloads, GitHub stars, and forks. Logarithmic scaling prevents top-heavy scores.

**Community (15%)** — Contributors, dependents, forks, watchers, and maintainers. Measures real ecosystem engagement.

**Maturity (30%)** — Project age, version count, PHP version support, and release stability.

###  Release Activity

Cadence

Unknown

Total

1

Last Release

29d ago

### Community

Maintainers

![](https://www.gravatar.com/avatar/ca757be9e125f187e3fc6f17f8f71463c29e0b083ef8dfb1c1616adc71381f50?d=identicon)[YanGusik](/maintainers/YanGusik)

---

Top Contributors

[![YanGusik](https://avatars.githubusercontent.com/u/28189620?v=4)](https://github.com/YanGusik "YanGusik (30 commits)")

### Embed Badge

![Health badge](/badges/yangusik-thrun/health.svg)

```
[![Health](https://phpackages.com/badges/yangusik-thrun/health.svg)](https://phpackages.com/packages/yangusik-thrun)
```

###  Alternatives

[league/geotools

Geo-related tools PHP 7.3+ library

1.4k5.5M29](/packages/league-geotools)[illuminate/bus

The Illuminate Bus package.

6145.5M491](/packages/illuminate-bus)[uecode/qpush-bundle

Asynchronous processing for Symfony using Push Queues

1672.5M2](/packages/uecode-qpush-bundle)[prooph/event-store-symfony-bundle

109256.9k10](/packages/prooph-event-store-symfony-bundle)[ezsystems/ezscriptmonitor-ls

eZ Publish extension that aims to avoid timeout problems and database corruption by moving long running processes from the GUI to the background.

13208.2k](/packages/ezsystems-ezscriptmonitor-ls)

PHPackages © 2026

[Directory](/)[Categories](/categories)[Trending](/trending)[Changelog](/changelog)[Analyze](/analyze)