view release on metacpan or  search on metacpan

include/dbil_pool.h  view on Meta::CPAN

        else if (n > 0) (void)POPs;
        PUTBACK; FREETMPS; LEAVE;
    }
    for (;;) {
        SV *req = dbil_read_frame_blocking(aTHX_ fd);
        SV *reqd, *resp, *frozen;
        AV *ra;
        int is_query;
        SV *sql, *err = NULL, *res;
        if (!req) break;                       /* parent closed: exit */
        { STRLEN l; const char *pv = SvPV_const(req, l); reqd = dbil_thaw(aTHX_ pv, l); }
        if (!reqd || !SvROK(reqd) || SvTYPE(SvRV(reqd)) != SVt_PVAV) continue;
        ra       = (AV *)SvRV(reqd);
        is_query = (int)SvIV(*av_fetch(ra, 0, 0));
        sql      = *av_fetch(ra, 1, 0);
        {
            SV **bp = av_fetch(ra, 2, 0);
            AV *bind = (bp && *bp && SvROK(*bp)) ? (AV *)SvRV(*bp) : NULL;
            res = dbh ? dbil_run_dbi(aTHX_ dbh, is_query, sql, bind, &err)
                      : (err = sv_2mortal(newSVpvs("worker has no database handle")), NULL);
        }
        {
            AV *out = newAV();
            av_push(out, newSViv(res ? 1 : 0));
            av_push(out, res ? newSVsv(res)
                             : 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
     * worker's fd is not readable - so this is the first and only moment we
     * learn about it. Leaving the slot alive meant dbil_pool_run kept
     * choosing it, being the lowest-numbered idle worker with a live fd, and
     * 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);

    /* Respawn is capped and can fail. If it did, and nothing is left to run
     * on, fail what we just queued rather than leave callers waiting on a
     * pool that can no longer answer. */
    if (!reserved && !dbil_pool_has_live_worker(p)) {
        while (av_len(p->queue) >= 1) {
            SV *qb = av_shift(p->queue);
            SV *qf = av_shift(p->queue);
            SvREFCNT_dec(qb);
            dbil_future_settle_fail(aTHX_ qf,
                sv_2mortal(newSVpvs("pool has no live workers")));
            SvREFCNT_dec(qf);
        }
    }
}

/* run one statement on the pool; returns the future (+1) */
static SV *dbil_pool_run(pTHX_ dbil_pool *p, int is_query, SV *sql, AV *bind) {
    SV *future = dbil_future_new(aTHX_ "DBIx::Loop::Future");
    SV *bytes;
    int wi;
    {   /* freeze the request tuple */
        AV *req = newAV();
        av_push(req, newSViv(is_query));
        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,