view release on metacpan or  search on metacpan

lib/DBIx/Loop.pm  view on Meta::CPAN

    # connection's own socket) while the loop keeps serving
    $db->query("SELECT * FROM pets WHERE id = ?", $id)->on_ready(sub {
        my $res = shift->get;   # { rows => [[...],...], columns => [...] }
        ...
    });

    # transactions, pinned to one connection
    $db->txn(sub {
        my ($tx) = @_;
        $tx->do("INSERT INTO pets (name) VALUES (?)", 'rex')
           ->then(sub { $tx->do("UPDATE counts SET pets = pets + 1") });
    })->on_ready(sub { ... });

=head1 DESCRIPTION

DBIx::Loop runs DBI queries without blocking an event loop, behind one
future-returning API. It is about B<concurrency and latency isolation>, not
per-query speed. DBD::SQLite and DBD::Pg already sit at parity with their C
libraries (a direct-libsqlite3 comparison showed a dead heat), but a blocking
query inside an event-driven server stalls every connection on that worker.
DBIx::Loop keeps the loop live.

DBIx::Loop is B<not an event loop and ships none> - you always supply a loop
adapter (L<IO::Async|DBIx::Loop::Loop::IOAsync>,
L<Mojo::IOLoop|DBIx::Loop::Loop::Mojo>,
L<AnyEvent|DBIx::Loop::Loop::AnyEvent>, Hyperman, ...). The engine - object,
capability probe, both backends, transactions, and L<DBIx::Loop::Future> - is
C.

=head1 THE TWO BACKENDS

DBI has no standard async API, so DBIx::Loop carries two backends behind the
one interface and picks per driver at connect time (see L</capability>):

=over 4

=item * B<The worker pool> (universal). For drivers with no async surface -
SQLite is the extreme: in-process, synchronous, un-yieldable - the blocking
call runs on a forked worker holding its own connection, framed over a
socketpair the loop watches. Works with every DBD. Workers use
C<prepare_cached>, crashed workers respawn (a storm cap applies), and
C<max_queue> bounds backpressure.

=item * B<Native fd async> (Pg; opt-in fast path). DBD::Pg exposes a
non-blocking execute and the connection's socket, so queries fire with
C<pg_async>, the loop watches C<pg_socket>, and results collect on readiness -
no workers, no serialization. One query per connection is in flight (a libpq
limit); the rest queue.

=back

Either way the result is the same shape: C<query> resolves to
C<< { rows => [ [...], ... ], columns => [ ... ] } >> (arrayref rows - they
benchmarked ~4x faster to build than hashrefs) and C<do> to
C<< { rows_affected => $n, insert_id => $id } >> (insert_id best-effort via
C<last_insert_id>).

A note on B<SQLite and more than one worker>: each worker is another process
with its own connection, and SQLite takes one writer at a time for the whole
file. Reads run concurrently; writes serialise, and a writer that keeps losing
the race gets C<database is locked> back once DBD::SQLite's busy timeout (30
seconds by default) is spent - which on a loaded machine a deep burst of
concurrent writes can genuinely do. For write-heavy SQLite either use
C<workers =E<gt> 1>, which costs no responsiveness (the queue still keeps the
loop free) and removes the contention outright, or raise
C<sqlite_busy_timeout>. Client/server engines have no such limit.

=head1 CONSTRUCTORS

=head2 connect

    my $db = DBIx::Loop->connect($dsn, $user, $pass, \%attr,
        loop      => $adapter,   # required; or 'auto'
        workers   => 4,          # pool backend: worker count
        max_queue => 0,          # pool backend: pending cap (0 = unbounded)
    );

Connects via C<< DBI->connect >> and keeps the connect arguments so pool
workers can open their own handles (a live handle cannot cross a fork).
C<< loop => 'auto' >> adapts an already-loaded loop (Mojo::IOLoop, then
IO::Async, then AnyEvent) and croaks when none is loaded - there is no
built-in loop, by design.

=head2 new

    my $db = DBIx::Loop->new(dbh => $dbh, loop => $adapter);

Wrap an existing handle. Without connect arguments the pool backend cannot
fork workers, so queries on a bare wrapped handle run synchronously; use
C<connect> for the non-blocking pool.

=head1 METHODS

=head2 query / do

    my $future = $db->query($sql, @bind);   # SELECT-style: rows + columns
    my $future = $db->do($sql, @bind);      # writes: rows_affected, insert_id

Both return a future immediately. C<< ->on_ready >>, C<< ->then >>,
C<< ->else >> chain; C<< ->get >> returns the result once ready (awaiting a
pending future is the adapter's job: C<< $adapter->await($future) >>).
Failures carry the DBI error string.

=head2 txn

    my $future = $db->txn(sub {
        my ($tx) = @_;
        # $tx->query / $tx->do are pinned to ONE connection
        return $tx->do(...)->then(sub { $tx->do(...) });
    });

Acquires a pool slot (waiting when all are busy), runs C<BEGIN>, calls the
block with a L<DBIx::Loop::Txn> handle pinned to that connection, then
C<COMMIT>s - or C<ROLLBACK>s when the block dies or its returned future fails,
failing the outer future with the original error. The block may return a plain
value or a future; the outer future resolves to it after commit.

Plain C<$db> statements during a transaction run on B<other> slots - they
never join the transaction and never steal its connection. A C<txn> inside a
block is an independent transaction on another slot (beware awaiting one while
its parents hold every slot). If a worker dies mid-transaction the transaction