DBIx-Loop
view release on metacpan or search on metacpan
include/dbil_pool.h view on Meta::CPAN
static AV *dbil_clos_cap(pTHX_ CV *cv) {
MAGIC *mg = mg_findext((SV *)cv, PERL_MAGIC_ext, &dbil_clos_vtbl);
return mg ? ((dbil_clos *)mg->mg_ptr)->cap : NULL;
}
/* ---- pool + workers -------------------------------------------------------- */
typedef struct dbil_worker {
pid_t pid;
int fd; /* parent's end of the duplex socketpair */
int busy;
int reserved; /* checked out to a transaction */
int gen; /* bumped on respawn: a txn pinned to an */
/* earlier gen is lost, not silently moved */
SV *future; /* in-flight future (+1), or NULL */
SV *rbuf; /* accumulated response bytes (+1) */
AV *txq; /* reserved-slot queue [bytes,future,...] (+1) */
struct dbil_pool *p; /* back-pointers: the C read-ready */
int wi; /* callback's ud is the worker */
} dbil_worker;
include/dbil_pool.h view on Meta::CPAN
_exit(0);
}
close(sp[1]);
/* The parent's end must not survive an exec either. */
{
int fl = fcntl(sp[0], F_GETFD);
if (fl >= 0) (void)fcntl(sp[0], F_SETFD, fl | FD_CLOEXEC);
}
w->pid = pid;
w->fd = sp[0];
w->busy = 0;
w->reserved = 0;
w->gen++; /* txns pinned to the old gen fail */
w->future = NULL;
if (!w->rbuf) w->rbuf = newSVpvs(""); /* reused across respawns */
if (!w->txq) w->txq = newAV();
w->p = p;
w->wi = wi;
if (p->vt) {
/* C-vtable adapter: readiness dispatches C-to-C, no Perl frame */
p->vt->add_reader(aTHX_ p->vt->ctx, w->fd, dbil_pool_reader_c, w);
include/dbil_pool.h view on Meta::CPAN
: newSVsv(err ? err : sv_2mortal(newSVpvs("failed"))));
resp = sv_2mortal(newRV_noinc((SV *)out));
}
frozen = dbil_freeze(aTHX_ resp);
if (frozen) (void)dbil_write_frame(aTHX_ fd, frozen);
}
close(fd);
_exit(0);
}
/* send request bytes to worker wi and mark it busy with `future` (+1 taken) */
/* Can anything still take general (non-transaction) work? */
static int dbil_pool_has_live_worker(const dbil_pool *p) {
int wi;
for (wi = 0; wi < p->nw; wi++)
if (p->w[wi].fd >= 0 && !p->w[wi].reserved) return 1;
return 0;
}
static void dbil_pool_send(pTHX_ dbil_pool *p, int wi, SV *bytes, SV *future) {
dbil_worker *w = &p->w[wi];
int reserved;
w->busy = 1;
w->future = SvREFCNT_inc(future);
if (dbil_write_frame(aTHX_ w->fd, bytes) >= 0) return;
/* The write failed, so the frame did not arrive whole and the statement
* provably never ran: dbil_write_frame only returns < 0 having written
* less than all of it.
*
* What matters here is the SLOT, not this one request. A worker killed
* while idle is invisible until something is written to it - death is
* otherwise only ever noticed as EOF on the read side, and an idle
include/dbil_pool.h view on Meta::CPAN
* failed every future from then on while the healthy workers sat idle.
* Treat a failed write as the death notice it is. */
reserved = w->reserved;
if (!reserved) {
/* Nothing executed, so this is a retry and not a loss. Put it back at
* the head of the queue so it keeps its place, and let the respawn
* inside worker_died pick it up. */
SvREFCNT_dec(w->future);
w->future = NULL;
w->busy = 0;
av_unshift(p->queue, 2);
av_store(p->queue, 0, newSVsv(bytes));
av_store(p->queue, 1, SvREFCNT_inc(future));
}
/* A reserved slot is a transaction pinned to that one connection. It
* cannot be moved, so leave the future in place for worker_died to fail
* alongside the rest of the transaction. */
dbil_pool_worker_died(aTHX_ p, wi);
include/dbil_pool.h view on Meta::CPAN
av_push(req, newSVsv(sql));
av_push(req, dbil_bind_rv(aTHX_ bind));
bytes = dbil_freeze(aTHX_ sv_2mortal(newRV_noinc((SV *)req)));
}
if (!bytes) {
dbil_future_settle_fail(aTHX_ future,
sv_2mortal(newSVpvs("could not serialise request")));
return future;
}
for (wi = 0; wi < p->nw; wi++)
if (!p->w[wi].busy && !p->w[wi].reserved && p->w[wi].fd >= 0)
{ dbil_pool_send(aTHX_ p, wi, bytes, future); return future; }
/* all busy: queue (bytes, future) - unless the backpressure cap is hit */
if (p->max_queue > 0 && (av_len(p->queue) + 1) / 2 >= p->max_queue) {
dbil_future_settle_fail(aTHX_ future,
sv_2mortal(newSVpvs("queue full (max_queue exceeded)")));
return future;
}
av_push(p->queue, newSVsv(bytes));
av_push(p->queue, SvREFCNT_inc(future));
return future;
}
/* pull the next queued request onto the now-free worker wi. A reserved
* worker only drains its own transaction queue; a free worker first serves a
* waiting acquire() (reserving itself), then the general queue. */
static void dbil_pool_dispatch(pTHX_ dbil_pool *p, int wi) {
dbil_worker *w = &p->w[wi];
if (w->reserved) {
if (!w->busy && av_len(w->txq) >= 1) {
SV *bytes = av_shift(w->txq);
SV *future = av_shift(w->txq);
dbil_pool_send(aTHX_ p, wi, sv_2mortal(bytes), future);
SvREFCNT_dec(future); /* send took its own +1 */
}
return;
}
if (av_len(p->acq) >= 0) { /* a txn is waiting for a slot */
SV *fut = av_shift(p->acq);
w->reserved = 1;
include/dbil_pool.h view on Meta::CPAN
/* ---- slot checkout (transactions) ------------------------------------------ */
/* acquire a slot: future resolves with the worker index (reserved for the
* caller). Resolves immediately when a slot is free. (+1) */
static SV *dbil_pool_acquire(pTHX_ dbil_pool *p) {
SV *future = dbil_future_new(aTHX_ "DBIx::Loop::Future");
int wi;
for (wi = 0; wi < p->nw; wi++) {
dbil_worker *w = &p->w[wi];
if (!w->busy && !w->reserved && w->fd >= 0) {
w->reserved = 1;
dbil_future_settle_done1(aTHX_ future, sv_2mortal(newSViv(wi)));
return future;
}
}
av_push(p->acq, SvREFCNT_inc(future));
return future;
}
/* release a reserved slot back to the pool and let it pick up work */
static void dbil_pool_release(pTHX_ dbil_pool *p, int wi) {
if (wi < 0 || wi >= p->nw) return;
p->w[wi].reserved = 0;
if (!p->w[wi].busy) dbil_pool_dispatch(aTHX_ p, wi);
}
/* send a statement on a reserved slot (FIFO behind any in-flight one).
* gen guards a txn whose worker died and respawned: that txn is lost, not
* silently continued on a fresh connection. */
static void dbil_pool_send_tx(pTHX_ dbil_pool *p, int wi, int gen,
SV *bytes, SV *future) {
dbil_worker *w;
if (wi < 0 || wi >= p->nw) {
dbil_future_settle_fail(aTHX_ future,
sv_2mortal(newSVpvs("transaction has no slot")));
return;
}
w = &p->w[wi];
if (w->fd < 0 || w->gen != gen) {
dbil_future_settle_fail(aTHX_ future,
sv_2mortal(newSVpvs("transaction lost (worker died)")));
return;
}
if (w->busy) {
av_push(w->txq, newSVsv(bytes));
av_push(w->txq, SvREFCNT_inc(future));
return;
}
dbil_pool_send(aTHX_ p, wi, bytes, future);
}
/* unregister fd from the loop via whichever seam the adapter has */
static void dbil_pool_loop_remove(pTHX_ dbil_pool *p, int fd) {
if (p->vt) {
include/dbil_pool.h view on Meta::CPAN
static void dbil_pool_worker_died(pTHX_ dbil_pool *p, int wi) {
dbil_worker *w = &p->w[wi];
if (w->future) {
dbil_future_settle_fail(aTHX_ w->future,
sv_2mortal(newSVpvs("worker exited unexpectedly")));
SvREFCNT_dec(w->future); w->future = NULL;
}
dbil_pool_loop_remove(aTHX_ p, w->fd);
close(w->fd);
w->fd = -1;
w->busy = 0;
if (w->pid > 0) { waitpid(w->pid, NULL, 0); w->pid = 0; }
if (w->rbuf) SvCUR_set(w->rbuf, 0); /* drop any partial frame */
/* a transaction pinned here is lost: fail everything queued on the slot
* (the gen bump in spawn stops later tx statements too) */
while (av_len(w->txq) >= 1) {
SV *bytes = av_shift(w->txq);
SV *fut = av_shift(w->txq);
SvREFCNT_dec(bytes);
dbil_future_settle_fail(aTHX_ fut,
sv_2mortal(newSVpvs("transaction lost (worker died)")));
include/dbil_pool.h view on Meta::CPAN
const unsigned char *b = (const unsigned char *)SvPVX(w->rbuf);
STRLEN len;
if (have < 4) break;
len = ((STRLEN)b[0] << 24) | ((STRLEN)b[1] << 16)
| ((STRLEN)b[2] << 8) | (STRLEN)b[3];
if (have < 4 + len) break;
{
SV *resp = dbil_thaw(aTHX_ (const char *)b + 4, len);
SV *fut = w->future;
w->future = NULL;
w->busy = 0;
if (fut) {
if (resp && SvROK(resp) && SvTYPE(SvRV(resp)) == SVt_PVAV) {
AV *ra = (AV *)SvRV(resp);
int ok = (int)SvIV(*av_fetch(ra, 0, 0));
SV *data = *av_fetch(ra, 1, 0);
if (ok) dbil_future_settle_done1(aTHX_ fut, data);
else dbil_future_settle_fail(aTHX_ fut, data);
} else {
dbil_future_settle_fail(aTHX_ fut,
sv_2mortal(newSVpvs("bad response frame from worker")));
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
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)
);
lib/DBIx/Loop.pm view on Meta::CPAN
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
B<fails> - it is never silently resumed on the respawned connection. Pool
t/02-pool.t view on Meta::CPAN
is($res->{rows}[0][0], 5, 'pool still serves after an error');
}
# ---- worker crash: in-flight future fails, slot respawns, pool recovers ---------
{
my @pids = $db->_worker_pids;
is(scalar @pids, 3, 'pool reports 3 worker pids');
# Kill a worker while a query is genuinely in flight on it.
#
# `busy` is the parent's bookkeeping, not the child's state: writing the
# request frame says nothing about whether the child has already read,
# run and answered it. Against SELECT COUNT(*) over five rows the child
# routinely finished before the signal landed, and the whole block then
# exercised the idle-death path instead (which is worth testing, and is
# tested separately below). A statement slow enough to still be running
# is what makes this deterministic.
my $SLOW = 'WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL '
. 'SELECT x+1 FROM c WHERE x < 3000000) SELECT COUNT(*) FROM c';
my @f = map { $db->query($SLOW) } 1 .. 3;
kill 'KILL', $pids[0];
$ad->await($_) for @f;
my $failed = grep { $_->is_failed } @f;
ok($failed >= 1, 'killing a busy worker fails its in-flight future')
or diag map { $_->is_failed ? "failed: " . $_->failure : "done" } @f;
# the slot respawned: pool still has 3 live pids, the dead one replaced
my @pids2 = $db->_worker_pids;
is(scalar(grep { $_ > 0 } @pids2), 3, 'dead slot respawned (3 live pids)');
isnt($pids2[0], $pids[0], 'slot 0 has a new pid');
# and keeps serving at full width
my @g = map { $db->query("SELECT COUNT(*) FROM t") } 1 .. 6;
$ad->await($_) for @g;
t/03-native-mock.t view on Meta::CPAN
package Mock::Attr; # anything read via ->FETCH($key)
sub new { my ($c, %h) = @_; bless {%h}, $c }
sub FETCH { $_[0]{ $_[1] } }
package Mock::PgSth;
sub new { my ($c, %h) = @_; bless {%h}, $c }
sub execute {
my $self = shift;
my $dbh = $self->{dbh};
die "mock execute: connection busy\n" if $dbh->{inflight};
$dbh->{inflight} = $self;
# completion arrives asynchronously: a forked child writes one byte to the
# far end of the socketpair after `delay` seconds
my $pid = fork; defined $pid or die "fork: $!";
if (!$pid) {
select undef, undef, undef, ($self->{delay} || 0.02);
syswrite $dbh->{notify}, "x", 1;
POSIX::_exit(0) if eval { require POSIX; 1 };
exit 0;
}
t/lib/BackendParity.pm view on Meta::CPAN
my $wide = $await->($db->query(
"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 (?,?,?)",
( run in 1.361 second using v1.01-cache-2.11-cpan-64ef6c95b5d )