view release on metacpan or  search on metacpan

src/EV__Gearman.xs  view on Meta::CPAN

              len, (unsigned)GM_MAX_PACKET);
}

static void send_pending_waits(pTHX_ ev_gm_t *self) {
    while (!ngx_queue_empty(&self->wait_queue) && self->connected) {
        ngx_queue_t *q = ngx_queue_head(&self->wait_queue);
        ev_gm_wait_t *w = ngx_queue_data(q, ev_gm_wait_t, queue);
        ngx_queue_remove(q);
        self->waiting_count--;

        buf_append_write(self, w->packet, w->packet_len);

        if (w->req) {
            ngx_queue_insert_tail(&self->cb_queue, &w->req->queue);
            self->pending_count++;
            w->req = NULL;
            arm_cmd_timer(self);
        }

        Safefree(w->packet); w->packet = NULL;
        Safefree(w);

        start_writing(self);
    }
}

/* ================================================================
 * Worker helpers (state machine for GRAB / NOOP / PRE_SLEEP)
 * ================================================================ */

static ev_gm_func_t* find_function(ev_gm_t *self, const char *name, STRLEN len) {
    ngx_queue_t *q;
    for (q = ngx_queue_head(&self->functions); q != ngx_queue_sentinel(&self->functions);
         q = ngx_queue_next(q))
    {
        ev_gm_func_t *f = ngx_queue_data(q, ev_gm_func_t, queue);
        if (strlen(f->name) == len && memcmp(f->name, name, len) == 0)
            return f;
    }
    return NULL;
}

static void worker_send_grab(pTHX_ ev_gm_t *self) {
    if (!self->connected) return;
    if (self->worker_grab_inflight) return;
    if (ngx_queue_empty(&self->functions)) return;
    self->worker_grab_inflight = 1;
    ev_gm_req_t *r = alloc_req(CB_GRAB_JOB, NULL);
    enqueue_packet(aTHX_ self,
        self->worker_grab_uniq ? GM_CMD_GRAB_JOB_UNIQ : GM_CMD_GRAB_JOB,
        NULL, 0, r);
}

static void worker_send_pre_sleep(pTHX_ ev_gm_t *self) {
    if (!self->connected) return;
    if (self->worker_sleeping) return;
    self->worker_sleeping = 1;
    enqueue_packet(aTHX_ self, GM_CMD_PRE_SLEEP, NULL, 0, NULL);
}

/* Drive the worker loop forward: if active and not busy, GRAB. */
static void worker_continue(pTHX_ ev_gm_t *self) {
    if (!self->worker_active || !self->connected) return;
    if (self->worker_grab_inflight || self->worker_sleeping) return;
    if (ngx_queue_empty(&self->functions)) return;
    worker_send_grab(aTHX_ self);
}

/* ================================================================
 * Connect-success path: re-register worker functions, drain queue
 * ================================================================ */

/* Send a WORK_* packet whose body is "handle" + optional "\0data".
   For WORK_FAIL pass dp=NULL; otherwise pass payload bytes. */
static void send_work_event(pTHX_ ev_gm_t *self, uint32_t cmd,
    const char *h, STRLEN hl, const char *dp, STRLEN dl)
{
    /* No size croak here: this runs on the worker completion / event
       path (worker_dispatch_job, $job->complete), where a longjmp would
       unwind the EV loop mid-iteration. A >4 GiB result is not a real
       concern (the worker would OOM building it first). */
    size_t plen = dp ? hl + 1 + dl : hl;
    char *body;
    Newx(body, plen ? plen : 1, char);
    memcpy(body, h, hl);
    if (dp) {
        body[hl] = '\0';
        if (dl) memcpy(body + hl + 1, dp, dl);
    }
    enqueue_packet(aTHX_ self, cmd, body, plen, NULL);
    Safefree(body);
}

/* Send CAN_DO[_TIMEOUT] for one function name. Builds the
   "name\0timeout" body on the heap to avoid the 512-byte stack-buffer
   truncation that the older inline pattern was prone to. */
static void send_can_do(pTHX_ ev_gm_t *self,
                        const char *name, STRLEN nlen, int timeout)
{
    if (!self->connected) return;
    if (timeout > 0) {
        char tbuf[16];
        int tlen = snprintf(tbuf, sizeof(tbuf), "%d", timeout);
        if (tlen <= 0) return;
        size_t plen = nlen + 1 + (size_t)tlen;
        char *body;
        Newx(body, plen, char);
        memcpy(body, name, nlen);
        body[nlen] = '\0';
        memcpy(body + nlen + 1, tbuf, tlen);
        enqueue_packet(aTHX_ self, GM_CMD_CAN_DO_TIMEOUT, body, plen, NULL);
        Safefree(body);
    } else {
        enqueue_packet(aTHX_ self, GM_CMD_CAN_DO, name, nlen, NULL);
    }
}

static void register_all_functions(pTHX_ ev_gm_t *self) {
    if (!self->connected) return;
    ngx_queue_t *q;
    for (q = ngx_queue_head(&self->functions); q != ngx_queue_sentinel(&self->functions);