Pool

__construct()#source

function __construct(
    callable $factory,
    int $max,
    ?callable $reset = null,
    ?callable $ping = null,
    float $maxLifetime = 0.0,
    float $idleTimeout = 0.0,
    ?callable $close = null
)

Create a pool.

Parameters

$factory callable

creates a new object; must return an object

$max int

maximum number of live objects, at least 1

$reset ?callable = null

called with the object on every release(); if it throws, the object is discarded

$ping ?callable = null

called with an idle object before acquire() hands it out; a false result or an exception discards the object (a freshly created object is not pinged)

$maxLifetime float = 0.0

seconds after which an object is retired when next considered; 0 means unlimited

$idleTimeout float = 0.0

seconds an object may sit idle before it is retired; 0 means unlimited

$close ?callable = null

called with every object the pool discards

Throws

\ValueError

when $max is below 1

pdo()#source

static function pdo(string $dsn, ?string $user, ?string $pass, array $options, int $max): Pool

A pool of PDO connections.

release() calls PDO::resetSession() (rolling back an open transaction and clearing session state) and acquire() pings an idle connection with PDO::ping(). For $maxLifetime, $idleTimeout or a $close callback build the pool with new Pool(...). The pool holds no connection until the first acquire().

$pool = Async\Pool::pdo('pgsql:host=db;dbname=app', 'u', 'pw', [], 10);
$rows = $pool->with(fn(PDO $db) => $db->query('select 1')->fetchAll());

Parameters

$dsn string

the PDO data source name

$user ?string

the database user

$pass ?string

the database password

$options array<int,mixed>

$max int

maximum number of connections, at least 1

Returns

Pool — a pool whose objects are PDO instances

Throws

\ValueError

when $max is below 1

acquire()#source

function acquire(?float $timeout = null): mixed

Take an object, creating one while there is capacity, else park FIFO until one is free.

The wait for a free object is bounded by $timeout; creating or validating the object is not. Pair every acquire() with release() or discard(), or use with().

Parameters

$timeout ?float = null

seconds to wait for a free object, or null to wait without a limit

Returns

T — an object checked out to the caller

Throws

TimeoutException

when $timeout seconds pass first

\RuntimeException

when the pool is closed, or closed while waiting

CancelledException

when the task is cancelled while waiting

See also: with, release

release()#source

function release(mixed $conn): void

Return an object to the pool.

Runs the $reset hook (shielded, so it may suspend), then hands the object to the first waiter or keeps it idle. If the hook throws, the object is discarded instead. When the pool is closed the object is dropped.

Parameters

$conn mixed

an object obtained from acquire()

Throws

\LogicException

on a second release, or an object that is not checked out from this pool

discard()#source

function discard(mixed $conn): void

Throw an object away instead of returning it, because it is broken or its state is unknown.

The capacity it held goes to the first waiter, and the $close hook, if any, is called with the object.

Parameters

$conn mixed

an object obtained from acquire()

Throws

\LogicException

when the object is not checked out from this pool

with()#source

function with(callable $fn): mixed

Acquire an object, run $fn($object), and release it in a finally.

The object is discarded instead of released when the body was cancelled, or failed with a PDOException whose SQLSTATE is class 08. with() takes no timeout of its own: bound the whole call with timeout().

Parameters

$fn callable

called with the object

Returns

mixed — the value returned by $fn

Throws

\Throwable

whatever $fn throws, after the object is released or discarded

Example

use Async\Pool;
use Async\TimeoutException;
use function Async\async;
use function Async\awaitAll;
use function Async\delay;
use function Async\spawn;

final class Connection
{
    public static int $made = 0;
    public int $id;

    public function __construct()
    {
        $this->id = ++self::$made;
    }
}

async(function () {
    // At most two connections ever exist; waiters are served in arrival order.
    $pool = new Pool(fn() => new Connection(), 2);

    $jobs = [];
    for ($i = 1; $i <= 4; $i++) {
        $jobs[] = spawn(fn() => $pool->with(function (Connection $c) {
            delay(0.01);
            return $c->id;
        }));
    }
    echo implode(",", awaitAll(...$jobs)), "\n";

    $stats = $pool->stats();
    echo "created {$stats['created']}, idle {$stats['idle']}, busy {$stats['busy']}\n";

    // Bound the wait for a free connection with a timeout.
    $a = $pool->acquire();
    $b = $pool->acquire();
    try {
        $pool->acquire(0.01);
    } catch (TimeoutException $e) {
        echo "no connection free\n";
    }
    $pool->release($a);
    $pool->discard($b);
    echo "discarded: {$pool->stats()['discarded']}\n";

    $pool->close();
});

Output

1,2,1,2
created 2, idle 2, busy 0
no connection free
discarded: 1

See also: acquire

close()#source

function close(): void

Close the pool. Calling it again does nothing.

New acquire() calls are refused, every waiter wakes with a RuntimeException, and idle objects are dropped. Objects still checked out are dropped when they come back.

See also: acquire

stats()#source

function stats(): array

Counters for monitoring the pool.

Returns

array<string,int> — idle, busy, waiting, created and discarded

Example

use Async\Pool;
use Async\TimeoutException;
use function Async\async;
use function Async\awaitAll;
use function Async\delay;
use function Async\spawn;

final class Connection
{
    public static int $made = 0;
    public int $id;

    public function __construct()
    {
        $this->id = ++self::$made;
    }
}

async(function () {
    // At most two connections ever exist; waiters are served in arrival order.
    $pool = new Pool(fn() => new Connection(), 2);

    $jobs = [];
    for ($i = 1; $i <= 4; $i++) {
        $jobs[] = spawn(fn() => $pool->with(function (Connection $c) {
            delay(0.01);
            return $c->id;
        }));
    }
    echo implode(",", awaitAll(...$jobs)), "\n";

    $stats = $pool->stats();
    echo "created {$stats['created']}, idle {$stats['idle']}, busy {$stats['busy']}\n";

    // Bound the wait for a free connection with a timeout.
    $a = $pool->acquire();
    $b = $pool->acquire();
    try {
        $pool->acquire(0.01);
    } catch (TimeoutException $e) {
        echo "no connection free\n";
    }
    $pool->release($a);
    $pool->discard($b);
    echo "discarded: {$pool->stats()['discarded']}\n";

    $pool->close();
});

Output

1,2,1,2
created 2, idle 2, busy 0
no connection free
discarded: 1