DBIx-Loop

 view release on metacpan or  search on metacpan

examples/ioasync.pl  view on Meta::CPAN

#!/usr/bin/env perl
use strict;
use warnings;

# DBIx::Loop on IO::Async: three CPU-heavy queries overlap on the worker pool
# while a loop timer keeps ticking - the loop is never blocked.
#
#   perl examples/ioasync.pl

use File::Temp ();
use DBIx::Loop;
use DBIx::Loop::Loop::IOAsync;

my $dir = File::Temp->newdir;
my $ad  = DBIx::Loop::Loop::IOAsync->new;
my $db  = DBIx::Loop->connect(

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

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

t/04-native-pg.t  view on Meta::CPAN

}

# ---- a genuinely async wait: the loop is free while Pg sleeps -------------------
{
    my $ticks = 0;
    my $timer_done = 0;
    $ad->timer(0.05, sub { $ticks++; $timer_done = 1 });
    my $f = $db->query("SELECT pg_sleep(0.3), 42 AS answer");
    $ad->await($f);
    is(($f->get)[0]{rows}[0][1], 42, 'slow query resolves');
    ok($timer_done, 'a loop timer fired while the query was in flight (loop not blocked)');
}

# ---- errors fail only their future ----------------------------------------------
{
    my $bad = $db->query("SELECT * FROM table_that_does_not_exist");
    my $good = $db->query("SELECT 7 AS ok");
    $ad->await($bad); $ad->await($good);
    ok($bad->is_failed, 'bad SQL -> failed future');
    like($bad->failure, qr/does not exist|relation/i, 'failure carries the Pg error');
    is(($good->get)[0]{rows}[0][0], 7, 'the queued query after it still succeeds');

t/lib/AdapterConformance.pm  view on Meta::CPAN

        my $res = ($q->get)[0];
        is(scalar @{ $res->{rows} }, 4, 'four rows back');
        is_deeply($res->{columns}, ['id', 'v'], 'columns named');

        # -- concurrency: more in flight than workers -------------------------
        my @f = map { $db->query("SELECT COUNT(*) FROM t") } 1 .. 5;
        $ad->await($_) for @f;
        is(scalar(grep { $_->is_done && ($_->get)[0]{rows}[0][0] == 4 } @f), 5,
            '5 concurrent queries across 2 workers all resolve');

        # -- a timer fires while a query is in flight (loop not blocked) ------
        SKIP: {
            skip 'adapter has no timer', 1 unless $ad->can('timer');
            my $fired = 0;
            $ad->timer(0.02, sub { $fired = 1 });
            my $slow = $db->query(
                "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM c WHERE x < 200000) SELECT COUNT(*) FROM c");
            $ad->await($slow);
            # give the timer a chance if the query won the race
            $ad->await(do { my $f = DBIx::Loop::Future->new;
                            $ad->timer(0.03, sub { $f->done(1) }); $f });

t/lib/BackendParity.pm  view on Meta::CPAN

            "SELECT id, name, n, id+1, id+2, id+3, id+4, id+5, id+6, id+7 "
          . "FROM p WHERE id = 1"));
        is(scalar @{ $wide->{rows}[0] }, 10, 'ten columns cross intact');

        # -- a larger result set --------------------------------------------------
        #
        # The seed writes go one at a time, on purpose. Two pool workers are
        # two processes writing one SQLite file and SQLite serialises writers,
        # so fired off as a 500-deep burst they fight for the write lock; on a
        # loaded machine a starved writer sits out its whole busy timeout and
        # comes back "database is locked". Nothing here ever looked at the
        # write futures, so that arrived as a silent short row count (a CPAN
        # smoker reported 497). Nothing about crossing 500 rows needs
        # concurrent writers - the concurrent-work path is what the read burst
        # below, AdapterConformance and t/02-pool.t are for - and a failed
        # write is now a named failure rather than a missing row.
        {
            my ($written, $err) = (0, undef);
            for my $i (100 .. 599) {
                my $f = $db->do("INSERT INTO p (id, name, n) VALUES (?,?,?)",
                                $i, "row$i", $i);