view release on metacpan or  search on metacpan

Future.xs  view on Meta::CPAN

        SvREFCNT_inc(cb);
        sv_2mortal(cb);

        PUSHMARK(SP);
        for (I32 i = 0; i < items; i++) {
            XPUSHs(sv_mortalcopy(ST(i)));
        }
        PUTBACK;

        race_cleanup(aTHX_ &ctx);

        call_sv(cb, G_DISCARD | G_VOID);

        FREETMPS;
        LEAVE;
    } else {
        race_cleanup(aTHX_ &ctx);
    }
    XSRETURN_EMPTY;
}

static evf_handle *parallel_start(pTHX_ AV *list, SV *worker, SV *final_cb, int unsafe, int want_handle) {
    I32 len = av_len(list) + 1;
    if (len <= 0) {
        if (IS_PVCV(final_cb)) {
            dSP;
            ENTER;
            SAVETMPS;
            PUSHMARK(SP);
            PUTBACK;
            call_sv(final_cb, G_DISCARD | G_VOID);
            FREETMPS;
            LEAVE;
        }
        /* Already over: hand back a dead handle so callers get a uniform
           object rather than undef. */
        return want_handle ? evf_handle_new(aTHX_ NULL, EVF_KIND_PARALLEL) : NULL;
    }

    ENTER;

    parallel_ctx *ctx;
    Newx(ctx, 1, parallel_ctx);
    ctx->is_freed_ptr = NULL;
    ctx->h = NULL;
    ctx->tasks = (AV*)SvREFCNT_inc((SV*)list);
    ctx->final_cb = SvREFCNT_inc(final_cb);
    ctx->worker = worker ? SvREFCNT_inc(worker) : NULL;
    ctx->remaining = len;
    ctx->cvs = NULL;
    ctx->num_cvs = 0;
    ctx->shared_cv = NULL;
    ctx->abandoned = 0;

    ifp_guard *guard = ifp_guard_push(aTHX_ &ctx->is_freed_ptr, NULL,
                                      NULL, &ctx->abandoned);
    int *is_freed = &guard->is_freed;

    /* Counted before anything is dispatched: a task that completes the whole
       operation synchronously runs cleanup, which drops the context's
       reference, and only this one keeps the cell alive until we return it. */
    evf_handle *h = NULL;
    if (want_handle) {
        h = evf_handle_new(aTHX_ NULL, EVF_KIND_PARALLEL);
        evf_handle_attach(aTHX_ h, ctx);
        ctx->h = h;
        guard->pending_h = h;
    }

    SV *done_rv = NULL;
    if (unsafe) {
        ctx->shared_cv = newXS(NULL, parallel_task_done, __FILE__);
        CvXSUBANY(ctx->shared_cv).any_ptr = ctx;
        done_rv = sv_2mortal(newRV_inc((SV*)ctx->shared_cv));
    } else {
        ctx->num_cvs = len;
        Newxz(ctx->cvs, len, CV*);
    }

    dSP;

    I32 i;
    /* G_VOID as well as G_DISCARD: the value is discarded either way, but
       without the want bit the task's tail expression runs in scalar context,
       so a nested primitive there would allocate a handle nobody asked for.
       The final_cb calls in this file already pass G_VOID for the same reason. */
    U32 flags = G_DISCARD | G_VOID | (unsafe ? 0 : G_EVAL);
    const int have_worker = (worker != NULL);

    for (i = 0; i < len; i++) {
        if (*is_freed) break;

        SV **task_ary = (AvREAL(list) && !SvMAGICAL(list)) ? AvARRAY(list) : NULL;
        SV **fetch_ptr = (task_ary && i <= AvFILL(list))
            ? &task_ary[i]
            : av_fetch(list, i, 0);
        SV *task_sv = fetch_ptr ? *fetch_ptr : NULL;

        if (have_worker || IS_PVCV(task_sv)) {
            CV *cv = NULL;
            if (!unsafe) {
                cv = newXS(NULL, parallel_task_done, __FILE__);
                CvXSUBANY(cv).any_ptr = ctx;
                ctx->cvs[i] = (CV*)SvREFCNT_inc((SV*)cv);
            }

            ENTER;
            SAVETMPS;
            if (!unsafe) {
                done_rv = sv_2mortal(newRV_noinc((SV*)cv));
            }
            PUSHMARK(SP);
            if (have_worker) {
                XPUSHs(task_sv ? sv_2mortal(SvREFCNT_inc(task_sv)) : &PL_sv_undef);
            }
            XPUSHs(done_rv);
            PUTBACK;

            call_sv(have_worker ? worker : task_sv, flags);
            SPAGAIN;
            if (!unsafe) {