EV-Gearman
view release on metacpan or search on metacpan
src/EV__Gearman.xs view on Meta::CPAN
/* Pending packets awaiting transmission (e.g. before connect) */
ngx_queue_t wait_queue;
int waiting_count;
/* Active jobs (foreground) by handle */
ngx_queue_t active_jobs;
int active_count_cached;
/* Worker state */
ngx_queue_t functions;
int worker_active; /* in worker loop */
int worker_sleeping; /* PRE_SLEEP issued, awaiting NOOP */
int worker_grab_inflight;/* GRAB_JOB[_UNIQ] in flight */
int worker_grab_uniq; /* prefer GRAB_JOB_UNIQ */
SV *worker_on_idle; /* fires when no jobs available */
int worker_one_shot; /* if 1, exit worker loop after first job */
char *client_id; /* for SET_CLIENT_ID */
/* Connection target */
char *host;
int port;
char *path;
/* getaddrinfo fallback list (TCP): live only while a connect
sequence walks it; freed on success, exhaustion or disconnect. */
struct addrinfo *ai_list;
struct addrinfo *ai_cur;
/* Reconnect */
int reconnect;
int reconnect_delay_ms;
int max_reconnect_attempts;
int reconnect_attempts;
ev_timer reconnect_timer;
int reconnect_timer_active;
int intentional_disconnect;
/* Timeouts */
int connect_timeout_ms;
ev_timer connect_timer;
int connect_timer_active;
int command_timeout_ms;
ev_timer cmd_timer;
int cmd_timer_active;
/* Safety */
int callback_depth;
int in_cb_cleanup;
int in_wait_cleanup;
int in_active_cleanup;
/* Live EV::Gearman::Job objects holding a tombstone reference on
this struct (via sv_magicext on the job HV). While > 0 the
struct is never Safefreed — a DESTROYed connection becomes an
inert tombstone until the last job releases it, so job_resolve's
magic-word check never reads freed memory. */
U32 job_refs;
/* Options */
int priority;
int keepalive;
/* Options sent on connect */
int opt_exceptions;
};
/* ================================================================
* Forward declarations
* ================================================================ */
static void io_cb(EV_P_ ev_io *w, int revents);
static void reconnect_timer_cb(EV_P_ ev_timer *w, int revents);
static void connect_timeout_cb(EV_P_ ev_timer *w, int revents);
static void cmd_timeout_cb(EV_P_ ev_timer *w, int revents);
static void start_reading(ev_gm_t *self);
static void stop_reading(ev_gm_t *self);
static void start_writing(ev_gm_t *self);
static void stop_writing(ev_gm_t *self);
static void start_connect(pTHX_ ev_gm_t *self);
static void cleanup_connection(pTHX_ ev_gm_t *self);
static void emit_error(pTHX_ ev_gm_t *self, const char *msg);
static void handle_disconnect(pTHX_ ev_gm_t *self, const char *reason);
static void schedule_reconnect(pTHX_ ev_gm_t *self);
static void apply_keepalive(ev_gm_t *self);
static void report_connect_error(pTHX_ ev_gm_t *self, const char *errbuf);
static void finish_connect_success(pTHX_ ev_gm_t *self);
static void stop_connect_timer(ev_gm_t *self);
static void stop_reconnect_timer(ev_gm_t *self);
static int check_destroyed(ev_gm_t *self);
static void clear_addrinfo(ev_gm_t *self);
static void connect_try_from(pTHX_ ev_gm_t *self, const char *prev_err);
static void cancel_pending(pTHX_ ev_gm_t *self, const char *reason);
static void cancel_waiting(pTHX_ ev_gm_t *self, const char *reason);
static void cancel_active(pTHX_ ev_gm_t *self, const char *reason);
static void send_pending_waits(pTHX_ ev_gm_t *self);
static void arm_cmd_timer(ev_gm_t *self);
static void disarm_cmd_timer(ev_gm_t *self);
static void enqueue_packet(pTHX_ ev_gm_t *self,
uint32_t cmd, const char *data, size_t data_len, ev_gm_req_t *req);
static void worker_continue(pTHX_ ev_gm_t *self);
static void worker_send_grab(pTHX_ ev_gm_t *self);
static void worker_send_pre_sleep(pTHX_ ev_gm_t *self);
static ev_gm_func_t* find_function(ev_gm_t *self, const char *name, STRLEN len);
/* ================================================================
* Big-endian helpers (no unaligned access)
* ================================================================ */
static void gm_write_u32(char *buf, uint32_t val) {
val = htonl(val);
memcpy(buf, &val, 4);
}
static uint32_t gm_read_u32(const char *buf) {
uint32_t val;
memcpy(&val, buf, 4);
return ntohl(val);
}
/* atoi/strtol on a length-bounded ASCII run that may not be NUL-terminated
(gm_arg's last field returns a pointer with no trailing NUL since the
separator-NUL only lives between fields). Copies up to 31 bytes onto
the stack first. */
static IV gm_atoi_n(const char *p, size_t len) {
char buf[32];
if (len > sizeof(buf) - 1) len = sizeof(buf) - 1;
memcpy(buf, p, len);
buf[len] = '\0';
return (IV)atoi(buf);
}
/* ================================================================
* Buffer management
* ================================================================ */
static void buf_ensure_write(ev_gm_t *self, size_t needed) {
/* Compact first to reclaim already-sent prefix. */
if (self->wbuf_off > 0) {
size_t live = self->wbuf_len - self->wbuf_off;
if (live > 0)
memmove(self->wbuf, self->wbuf + self->wbuf_off, live);
self->wbuf_len = live;
self->wbuf_off = 0;
}
src/EV__Gearman.xs view on Meta::CPAN
/* Invoke Perl callback with up to two args (any may be NULL = undef).
* Mortal SV transfer: caller passes refcnt=1 SVs, we mortalize them. */
static void invoke_cb2(pTHX_ ev_gm_t *self, SV *cb, SV *a, SV *b) {
if (!cb) {
if (a) SvREFCNT_dec(a);
if (b) SvREFCNT_dec(b);
return;
}
self->callback_depth++;
dSP;
ENTER;
SAVETMPS;
PUSHMARK(SP);
EXTEND(SP, 2);
if (a) mPUSHs(a); else PUSHs(&PL_sv_undef);
if (b) mPUSHs(b); else PUSHs(&PL_sv_undef);
PUTBACK;
call_sv(cb, G_DISCARD | G_EVAL);
if (SvTRUE(ERRSV)) {
warn("EV::Gearman: callback error: %s", SvPV_nolen(ERRSV));
sv_setsv(ERRSV, &PL_sv_undef);
}
FREETMPS;
LEAVE;
self->callback_depth--;
}
/* Invoke a handler with at most one mortal arg; eat exceptions. */
static void invoke_handler(pTHX_ ev_gm_t *self, SV *cb, SV *arg, const char *label) {
if (!cb) { if (arg) SvREFCNT_dec(arg); return; }
self->callback_depth++;
dSP;
ENTER;
SAVETMPS;
PUSHMARK(SP);
if (arg) XPUSHs(sv_2mortal(arg));
PUTBACK;
call_sv(cb, G_DISCARD | G_EVAL);
if (SvTRUE(ERRSV)) {
warn("EV::Gearman: %s callback error: %s", label, SvPV_nolen(ERRSV));
sv_setsv(ERRSV, &PL_sv_undef);
}
FREETMPS;
LEAVE;
self->callback_depth--;
}
static void emit_error(pTHX_ ev_gm_t *self, const char *msg) {
invoke_handler(aTHX_ self, self->on_error, newSVpv(msg, 0), "on_error");
}
static void emit_connect(pTHX_ ev_gm_t *self) {
invoke_handler(aTHX_ self, self->on_connect, NULL, "on_connect");
}
static void emit_disconnect(pTHX_ ev_gm_t *self) {
invoke_handler(aTHX_ self, self->on_disconnect, NULL, "on_disconnect");
}
static void apply_keepalive(ev_gm_t *self) {
if (self->keepalive <= 0 || self->path) return;
int one = 1;
setsockopt(self->fd, SOL_SOCKET, SO_KEEPALIVE, &one, sizeof(one));
#ifdef TCP_KEEPIDLE
setsockopt(self->fd, IPPROTO_TCP, TCP_KEEPIDLE,
&self->keepalive, sizeof(self->keepalive));
#endif
}
/* Set O_NONBLOCK and FD_CLOEXEC on an already-open socket fd. The
CLOEXEC step prevents the fd leaking into child processes spawned
via fork+exec while the connection is up. Returns 0 on success,
-1 on failure (caller should close and error out). */
static int gm_set_socket_flags(int fd) {
int fl = fcntl(fd, F_GETFL);
if (fl < 0 || fcntl(fd, F_SETFL, fl | O_NONBLOCK) < 0) return -1;
#ifdef FD_CLOEXEC
fl = fcntl(fd, F_GETFD);
if (fl >= 0) (void)fcntl(fd, F_SETFD, fl | FD_CLOEXEC);
#endif
#ifdef SO_NOSIGPIPE
/* macOS/BSD lack MSG_NOSIGNAL; this is their per-socket way to
keep a write to a closed peer from raising SIGPIPE. A library
must never be able to kill its embedder with a signal. */
{ int one = 1; (void)setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof(one)); }
#endif
return 0;
}
/* Shared get/set body for on_error / on_connect / on_disconnect. */
static SV* handler_accessor(pTHX_ SV **slot, int items, SV *new_cb) {
if (items > 1) {
if (*slot) { SvREFCNT_dec(*slot); *slot = NULL; }
if (SvOK(new_cb) && SvROK(new_cb) && SvTYPE(SvRV(new_cb)) == SVt_PVCV)
*slot = newSVsv(new_cb);
}
return *slot ? newSVsv(*slot) : &PL_sv_undef;
}
/* ================================================================
* Allocators / cleaners
* ================================================================ */
/* Allocate a request entry with kind set; if cb_sv is a valid coderef,
bumps its refcount into r->cb. Pass NULL (or a non-coderef) for no cb. */
static ev_gm_req_t* alloc_req(int kind, SV *cb_sv) {
dTHX;
ev_gm_req_t *r;
Newxz(r, 1, ev_gm_req_t);
r->kind = kind;
if (cb_sv && SvOK(cb_sv) && SvROK(cb_sv))
r->cb = newSVsv(cb_sv);
return r;
}
static void cleanup_req(pTHX_ ev_gm_req_t *r) {
CLEAR_HANDLER(r->cb);
CLEAR_HANDLER(r->on_data);
CLEAR_HANDLER(r->on_warning);
CLEAR_HANDLER(r->on_status);
CLEAR_HANDLER(r->on_exception);
Safefree(r);
}
static void cleanup_wait(pTHX_ ev_gm_wait_t *w) {
Safefree(w->packet);
src/EV__Gearman.xs view on Meta::CPAN
}
self->connected = 0;
self->connecting = 0;
self->rbuf_len = 0;
self->wbuf_len = 0;
self->wbuf_off = 0;
/* Worker state resets: a reconnect must re-issue CAN_DO and re-start
the GRAB loop from scratch. Registered functions are kept. */
self->worker_sleeping = 0;
self->worker_grab_inflight = 0;
clear_addrinfo(self);
}
/* Drop the resolved-address list, if any. It is only live while a
connect sequence is walking it. */
static void clear_addrinfo(ev_gm_t *self) {
if (self->ai_list) {
freeaddrinfo(self->ai_list);
self->ai_list = NULL;
self->ai_cur = NULL;
}
}
static void handle_disconnect(pTHX_ ev_gm_t *self, const char *reason) {
int was_connected = self->connected;
cleanup_connection(aTHX_ self);
cancel_pending(aTHX_ self, "disconnected");
if (self->magic == GM_MAGIC_FREED) return;
cancel_active(aTHX_ self, "disconnected");
if (self->magic == GM_MAGIC_FREED) return;
cancel_waiting(aTHX_ self, "disconnected");
if (self->magic == GM_MAGIC_FREED) return;
/* Arm the reconnect BEFORE firing user callbacks: re-queueing work
from on_disconnect/on_error is the documented purpose of the
wait queue, and GM_CROAK_UNLESS_ALIVE keys off the reconnect
timer being active. schedule_reconnect may itself emit_error
("max reconnect attempts reached"), so re-check for DESTROY. */
if (!self->intentional_disconnect && self->reconnect)
schedule_reconnect(aTHX_ self);
if (check_destroyed(self)) return;
if (was_connected) {
emit_disconnect(aTHX_ self);
if (check_destroyed(self)) return;
}
if (reason) {
emit_error(aTHX_ self, reason);
if (check_destroyed(self)) return;
}
}
static void schedule_reconnect(pTHX_ ev_gm_t *self) {
/* DESTROY during a callback leaves a zombie (magic FREED, struct
kept alive by callback_depth); arming a timer on it would point
libev at freed memory once the depth unwinds. */
if (self->magic != GM_MAGIC_ALIVE) return;
if (self->reconnect_timer_active) return;
if (self->max_reconnect_attempts > 0 &&
self->reconnect_attempts >= self->max_reconnect_attempts) {
emit_error(aTHX_ self, "max reconnect attempts reached");
return;
}
self->reconnect_attempts++;
ev_tstamp delay = (ev_tstamp)self->reconnect_delay_ms / 1000.0;
if (delay < 0) delay = 0;
ev_timer_init(&self->reconnect_timer, reconnect_timer_cb, delay, 0.0);
self->reconnect_timer.data = (void *)self;
ev_timer_start(self->loop, &self->reconnect_timer);
self->reconnect_timer_active = 1;
}
static void stop_connect_timer(ev_gm_t *self) {
if (self->connect_timer_active) {
ev_timer_stop(self->loop, &self->connect_timer);
self->connect_timer_active = 0;
}
}
static void stop_reconnect_timer(ev_gm_t *self) {
if (self->reconnect_timer_active) {
ev_timer_stop(self->loop, &self->reconnect_timer);
self->reconnect_timer_active = 0;
}
}
/* The command timeout is per-request: the request at the head of
cb_queue must be answered within command_timeout_ms of its sent_at
stamp, regardless of how much unrelated traffic flows meanwhile.
Arm a one-shot timer for the current head's remaining budget;
callers re-arm whenever the head changes (pop) or the queue empties
(disarm via the empty-check here). */
static void arm_cmd_timer(ev_gm_t *self) {
if (self->cmd_timer_active) {
ev_timer_stop(self->loop, &self->cmd_timer);
self->cmd_timer_active = 0;
}
if (self->command_timeout_ms <= 0) return;
if (!self->connected) return;
if (ngx_queue_empty(&self->cb_queue)) return;
ev_gm_req_t *head = ngx_queue_data(ngx_queue_head(&self->cb_queue),
ev_gm_req_t, queue);
ev_tstamp budget = (ev_tstamp)self->command_timeout_ms / 1000.0;
ev_tstamp remaining = head->sent_at + budget - ev_now(self->loop);
if (remaining < 0) remaining = 0;
ev_timer_init(&self->cmd_timer, cmd_timeout_cb, remaining, 0.0);
self->cmd_timer.data = (void *)self;
ev_timer_start(self->loop, &self->cmd_timer);
self->cmd_timer_active = 1;
}
static void disarm_cmd_timer(ev_gm_t *self) {
if (self->cmd_timer_active) {
ev_timer_stop(self->loop, &self->cmd_timer);
self->cmd_timer_active = 0;
src/EV__Gearman.xs view on Meta::CPAN
/* 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);
q = ngx_queue_next(q))
{
ev_gm_func_t *f = ngx_queue_data(q, ev_gm_func_t, queue);
send_can_do(aTHX_ self, f->name, strlen(f->name), f->timeout);
}
}
static void send_options_and_id(pTHX_ ev_gm_t *self) {
if (self->client_id) {
enqueue_packet(aTHX_ self, GM_CMD_SET_CLIENT_ID,
self->client_id, strlen(self->client_id), NULL);
}
if (self->opt_exceptions) {
const char *opt = "exceptions";
ev_gm_req_t *r = alloc_req(CB_OPTION, NULL);
enqueue_packet(aTHX_ self, GM_CMD_OPTION_REQ, opt, strlen(opt), r);
}
}
static void report_connect_error(pTHX_ ev_gm_t *self, const char *errbuf) {
self->callback_depth++;
emit_error(aTHX_ self, errbuf);
self->callback_depth--;
if (check_destroyed(self)) return;
if (!self->intentional_disconnect && self->reconnect)
schedule_reconnect(aTHX_ self);
}
static void finish_connect_success(pTHX_ ev_gm_t *self) {
self->reconnect_attempts = 0;
start_reading(self);
apply_keepalive(self);
send_options_and_id(aTHX_ self);
register_all_functions(aTHX_ self);
emit_connect(aTHX_ self);
if (check_destroyed(self)) return;
send_pending_waits(aTHX_ self);
if (check_destroyed(self)) return;
worker_continue(aTHX_ self);
}
/* ================================================================
* Response routing
* ================================================================ */
/* Return pointer to first byte of the i-th NUL-separated argument
* (zero-indexed) and its length in *out_len. Returns NULL if not
* present. The final argument is whatever remains after the last
* NUL — it may itself contain NUL bytes for binary data. */
static const char* gm_arg(const char *body, size_t body_len, int idx,
int total, size_t *out_len)
{
/* Walk through, splitting by NUL. The (total-1) NULs separate
total fields; the final field has no trailing NUL. */
const char *p = body;
const char *end = body + body_len;
int cur = 0;
while (cur < idx) {
const char *nul = memchr(p, '\0', end - p);
if (!nul) return NULL;
p = nul + 1;
cur++;
}
/* p is start of arg `idx`; find its length */
if (idx == total - 1) {
/* Last arg runs to end */
*out_len = end - p;
} else {
const char *nul = memchr(p, '\0', end - p);
if (!nul) return NULL;
*out_len = nul - p;
}
return p;
}
static ev_gm_active_t* find_active_by_handle(ev_gm_t *self,
const char *handle, STRLEN handle_len)
{
ngx_queue_t *q;
for (q = ngx_queue_head(&self->active_jobs); q != ngx_queue_sentinel(&self->active_jobs);
q = ngx_queue_next(q))
{
ev_gm_active_t *a = ngx_queue_data(q, ev_gm_active_t, queue);
if (a->handle_len == handle_len &&
memcmp(a->handle, handle, handle_len) == 0)
return a;
}
return NULL;
src/EV__Gearman.xs view on Meta::CPAN
ev_gm_req_t *r = pop_req_expect(aTHX_ self, cmd,
GM_KIND_MASK(CB_GRAB_JOB));
if (!r) return;
worker_dispatch_job(aTHX_ self, r,
cmd == GM_CMD_JOB_ASSIGN_UNIQ ? 1 : 0, body, body_len);
cleanup_req(aTHX_ r);
break;
}
case GM_CMD_NOOP: {
self->worker_sleeping = 0;
worker_continue(aTHX_ self);
break;
}
case GM_CMD_ERROR: {
/* ERRCODE\0ERR_TEXT. ERROR carries no handle, so attribution is
impossible in general — and "pop whatever is at the head" is
actively destructive: gearmand emits JOB_NOT_FOUND
asynchronously for a WORK_* we sent (e.g. a double
complete()), which on a worker arrives while the head is the
in-flight GRAB_JOB; popping it wedges the worker for good.
So: pop only a head kind that can actually fail this way
(in-order rejection of that request — GRAB_JOB is never
answered with ERROR, ADMIN expects text); anything else is
surfaced at connection level with the FIFO left intact. */
size_t cl, tl;
const char *c = gm_arg(body, body_len, 0, 2, &cl);
const char *t = gm_arg(body, body_len, 1, 2, &tl);
SV *err;
if (c && t)
err = newSVpvf("%.*s: %.*s", (int)cl, c, (int)tl, t);
else
err = newSVpvn(body, body_len);
ev_gm_req_t *head = ngx_queue_empty(&self->cb_queue) ? NULL
: ngx_queue_data(ngx_queue_head(&self->cb_queue),
ev_gm_req_t, queue);
if (head && head->kind != CB_GRAB_JOB && head->kind != CB_ADMIN) {
pop_req(self);
if (head->cb) {
invoke_cb2(aTHX_ self, head->cb, NULL, err);
} else {
emit_error(aTHX_ self, SvPV_nolen(err));
SvREFCNT_dec(err);
}
cleanup_req(aTHX_ head);
} else {
emit_error(aTHX_ self, SvPV_nolen(err));
SvREFCNT_dec(err);
}
break;
}
default:
break;
}
}
/* Dispatch a JOB_ASSIGN[_UNIQ]: invoke registered function callback,
* deliver result back to server (for sync mode). Async mode keeps a
* job object alive in Perl-land until user invokes complete/fail. */
static void worker_dispatch_job(pTHX_ ev_gm_t *self, ev_gm_req_t *r,
int with_unique, const char *body, size_t body_len)
{
int total = with_unique ? 4 : 3;
size_t hl, fl, ul, wl;
const char *h = gm_arg(body, body_len, 0, total, &hl);
const char *f = gm_arg(body, body_len, 1, total, &fl);
const char *u = with_unique ? gm_arg(body, body_len, 2, total, &ul) : NULL;
const char *w = gm_arg(body, body_len, with_unique ? 3 : 2, total, &wl);
if (!h || !f) return;
/* Build a job hashref blessed into EV::Gearman::Job. */
HV *job = newHV();
hv_stores(job, "handle", newSVpvn(h, hl));
hv_stores(job, "function", newSVpvn(f, fl));
hv_stores(job, "unique", u ? newSVpvn(u, ul) : newSVpvn("", 0));
hv_stores(job, "workload", w ? newSVpvn(w, wl) : newSVpvn("", 0));
/* Tombstone reference on the client, carried as PERL_MAGIC_ext on
the job HV — not a hash key, so user code, hash walkers and
serializers can neither see nor clobber it, and a forged job
hash has no magic to pass job_resolve's check. svt_free
(ev_gm_job_magic_free) releases the reference; while any job
lives, job_refs > 0 keeps this struct allocated (as an inert
tombstone once DESTROY has run), so job_resolve's magic-word
check never reads freed memory.
Do NOT "simplify" this into storing the client's blessed SV with
a bumped refcount: the old comment's double-DESTROY fear was
unfounded (DESTROY fires once when the one blessed RV's refcount
hits zero, no matter how many references a job holds), but a
strong ref would invert ownership — the connection would stay
connected and grabbing jobs, invisibly, for as long as any job
is stashed, with no handle left to stop it. The RV is the sole
owner; jobs are observers. */
self->job_refs++;
sv_magicext((SV *)job, NULL, PERL_MAGIC_ext,
&ev_gm_job_magic_vtbl, (const char *)self, 0);
/* Owned reference, NOT a mortal: io_cb is a raw libev callback and
never runs inside a per-callback Perl scope, so no FREETMPS would
ever pop a mortal off the tmps stack — every dispatched job would
stay pinned until EV::run returns (i.e. never, for a worker
daemon). We release this reference on each exit path below. */
SV *jobref = newRV_noinc((SV*)job);
sv_bless(jobref, gv_stashpv("EV::Gearman::Job", GV_ADD));
/* grab_job mode: caller supplied an explicit cb on the request,
deliver the job to it instead of the registered-function path.
invoke_cb2 mortalizes its arg, taking ownership of jobref. */
if (r->cb) {
invoke_cb2(aTHX_ self, r->cb, jobref, NULL);
worker_continue(aTHX_ self);
return;
}
ev_gm_func_t *fn = find_function(self, f, fl);
if (!fn || !fn->cb) {
/* A plain can_do() entry records the ability (so reconnect
re-sends CAN_DO and grab_job can pull such jobs) but carries
no handler; calling it would eat "Not a CODE reference"
under G_EVAL and answer every job with a silent WORK_FAIL.
Fail loudly (once per function) instead. */
if (fn && !fn->no_cb_warned) {
fn->no_cb_warned = 1;
warn("EV::Gearman: no handler for function '%.*s' "
"(registered via can_do without a callback); "
"use register_function or grab_job", (int)fl, f);
}
SvREFCNT_dec(jobref);
send_work_event(aTHX_ self, GM_CMD_WORK_FAIL, h, hl, NULL, 0);
worker_continue(aTHX_ self);
return;
}
/* Snapshot fn->async before invoking the user callback: the
callback is allowed to call cant_do() or reset_abilities(),
which would free `fn`, leaving us with a stale pointer. The
SV behind fn->cb stays alive during call_sv via Perl's own
sub-context refcount, so reading it is safe. */
int is_async = fn->async;
self->callback_depth++;
dSP;
ENTER; SAVETMPS;
PUSHMARK(SP);
XPUSHs(jobref);
PUTBACK;
int count = call_sv(fn->cb, is_async ? (G_DISCARD | G_EVAL) : (G_SCALAR | G_EVAL));
SPAGAIN;
SV *retval = NULL;
int had_error = SvTRUE(ERRSV) ? 1 : 0;
SV *err_sv = had_error ? newSVsv(ERRSV) : NULL;
if (had_error)
sv_setsv(ERRSV, &PL_sv_undef);
if (!is_async && count > 0) {
retval = POPs;
SvREFCNT_inc(retval);
}
PUTBACK;
FREETMPS; LEAVE;
self->callback_depth--;
/* Release our owned jobref; if the callback stashed the job (async
mode) its own reference keeps the job alive. */
SvREFCNT_dec(jobref);
if (self->magic == GM_MAGIC_FREED) {
if (retval) SvREFCNT_dec(retval);
if (err_sv) SvREFCNT_dec(err_sv);
return;
}
if (!is_async) {
if (had_error) {
/* WORK_EXCEPTION is terminal at the server (gearmand calls
fail() on the job after forwarding the exception data),
so send it INSTEAD OF WORK_FAIL when the option is on —
sending both produces a JOB_NOT_FOUND error on the
second packet. */
if (self->opt_exceptions && err_sv) {
STRLEN el; const char *ep = SvPV(err_sv, el);
send_work_event(aTHX_ self, GM_CMD_WORK_EXCEPTION, h, hl, ep, el);
} else {
send_work_event(aTHX_ self, GM_CMD_WORK_FAIL, h, hl, NULL, 0);
}
} else {
STRLEN dl = 0;
const char *dp = "";
if (retval && SvOK(retval))
dp = SvPV(retval, dl);
send_work_event(aTHX_ self, GM_CMD_WORK_COMPLETE, h, hl, dp, dl);
}
} else if (had_error && err_sv) {
/* Async: the user keeps the job around and calls complete/fail
later. We immediately grab the next job so async workers can
process multiple jobs concurrently — bounded only by what the
server has queued. Users who want a concurrency cap call
work_stop in their callback (and work() again from
complete()/fail() to resume). */
warn("EV::Gearman: async worker callback raised: %s",
SvPV_nolen(err_sv));
}
if (retval) SvREFCNT_dec(retval);
if (err_sv) SvREFCNT_dec(err_sv);
if (self->worker_one_shot) {
self->worker_active = 0;
self->worker_one_shot = 0;
} else {
worker_continue(aTHX_ self);
}
}
/* ================================================================
* Read path: parse binary or text-protocol responses
*
* The server may emit two stream styles:
* - Binary packets with magic "\0RES" (response to binary requests)
* - Text lines (response to admin commands like "status\n")
*
* We choose dispatch based on the head request kind: if head is
* CB_ADMIN, we expect text; otherwise binary. If both styles are
* intermixed we parse whichever the head expects.
*
src/EV__Gearman.xs view on Meta::CPAN
/* ================================================================
* Helper to encode a multi-arg body: NUL-separated.
* ================================================================ */
/* Concatenate args with NUL separators into out (caller frees).
* args: array of {ptr, len} pairs; n_args entries.
* Returns total length in *out_len. NULL ptr is treated as empty. */
static char* gm_encode_args(int n_args, const char **ptrs, const STRLEN *lens, size_t *out_len) {
size_t total = 0;
int i;
for (i = 0; i < n_args; i++) total += (ptrs[i] ? lens[i] : 0);
if (n_args > 1) total += (n_args - 1);
char *out;
Newx(out, total ? total : 1, char);
char *p = out;
for (i = 0; i < n_args; i++) {
if (i > 0) { *p++ = '\0'; }
if (ptrs[i] && lens[i] > 0) {
memcpy(p, ptrs[i], lens[i]);
p += lens[i];
}
}
*out_len = total;
return out;
}
/* ================================================================
* XS interface
* ================================================================ */
MODULE = EV::Gearman PACKAGE = EV::Gearman
BOOT:
{
I_EV_API("EV::Gearman");
}
EV::Gearman
new(char *class, ...)
CODE:
{
PERL_UNUSED_VAR(class);
if ((items - 1) % 2 != 0) croak("odd number of arguments");
Newxz(RETVAL, 1, ev_gm_t);
RETVAL->magic = GM_MAGIC_ALIVE;
RETVAL->fd = -1;
RETVAL->port = 4730;
ngx_queue_init(&RETVAL->cb_queue);
ngx_queue_init(&RETVAL->wait_queue);
ngx_queue_init(&RETVAL->active_jobs);
ngx_queue_init(&RETVAL->functions);
Newx(RETVAL->rbuf, BUF_INIT_SIZE, char);
RETVAL->rbuf_cap = BUF_INIT_SIZE;
Newx(RETVAL->wbuf, BUF_INIT_SIZE, char);
RETVAL->wbuf_cap = BUF_INIT_SIZE;
/* Default error handler. eval_pv hands back the result without
transferring a reference we own (it stays mortal), so take one
to keep the sub alive past the current statement. */
RETVAL->on_error = eval_pv("sub { warn \"EV::Gearman error: @_\\n\" }", TRUE);
SvREFCNT_inc_simple_void_NN(RETVAL->on_error);
SV *host_sv = NULL, *path_sv = NULL;
int port = 4730;
int do_reconnect = 0, reconnect_delay = 1000, max_reconnect_attempts = 0;
RETVAL->loop = EV_DEFAULT;
int i;
for (i = 1; i < items; i += 2) {
const char *k = SvPV_nolen(ST(i));
SV *v = ST(i + 1);
if (strEQ(k, "host")) host_sv = v;
else if (strEQ(k, "port")) port = SvIV(v);
else if (strEQ(k, "path")) path_sv = v;
else if (strEQ(k, "on_error")) {
CLEAR_HANDLER(RETVAL->on_error);
if (SvOK(v) && SvROK(v)) RETVAL->on_error = newSVsv(v);
}
else if (strEQ(k, "on_connect")) {
if (SvOK(v) && SvROK(v)) RETVAL->on_connect = newSVsv(v);
}
else if (strEQ(k, "on_disconnect")) {
if (SvOK(v) && SvROK(v)) RETVAL->on_disconnect = newSVsv(v);
}
else if (strEQ(k, "connect_timeout")) RETVAL->connect_timeout_ms = SvIV(v);
else if (strEQ(k, "command_timeout")) RETVAL->command_timeout_ms = SvIV(v);
else if (strEQ(k, "priority")) RETVAL->priority = SvIV(v);
else if (strEQ(k, "keepalive")) RETVAL->keepalive = SvIV(v);
else if (strEQ(k, "reconnect")) do_reconnect = SvTRUE(v) ? 1 : 0;
else if (strEQ(k, "reconnect_delay")) reconnect_delay = SvIV(v);
else if (strEQ(k, "max_reconnect_attempts")) max_reconnect_attempts = SvIV(v);
else if (strEQ(k, "exceptions")) RETVAL->opt_exceptions = SvTRUE(v) ? 1 : 0;
else if (strEQ(k, "client_id")) {
if (SvOK(v)) RETVAL->client_id = savepv(SvPV_nolen(v));
}
else if (strEQ(k, "grab_unique")) RETVAL->worker_grab_uniq = SvTRUE(v) ? 1 : 0;
else if (strEQ(k, "loop")) {
if (!SvROK(v) || !sv_derived_from(v, "EV::Loop")) {
Safefree(RETVAL->rbuf);
Safefree(RETVAL->wbuf);
CLEAR_HANDLER(RETVAL->on_error);
CLEAR_HANDLER(RETVAL->on_connect);
CLEAR_HANDLER(RETVAL->on_disconnect);
CLEAR_HANDLER(RETVAL->loop_sv);
if (RETVAL->client_id) Safefree(RETVAL->client_id);
Safefree(RETVAL);
croak("loop => must be an EV::Loop object");
}
/* An EV::Loop object is a blessed scalar holding the
struct ev_loop pointer in its IV slot (T_PTROBJ style);
the PV slot is NULL, so SvPVX would read garbage. */
RETVAL->loop = INT2PTR(struct ev_loop *, SvIV(SvRV(v)));
/* Hold a strong ref so the loop outlives the client even
when the caller drops their own reference to it. */
CLEAR_HANDLER(RETVAL->loop_sv);
RETVAL->loop_sv = newSVsv(v);
}
else {
/* An unknown key is almost always a typo ("reconect") that
would silently leave the intended option unset — fail
loudly. Tear the partial object down exactly like the
loop-validation branch above. */
Safefree(RETVAL->rbuf);
Safefree(RETVAL->wbuf);
CLEAR_HANDLER(RETVAL->on_error);
CLEAR_HANDLER(RETVAL->on_connect);
CLEAR_HANDLER(RETVAL->on_disconnect);
CLEAR_HANDLER(RETVAL->loop_sv);
if (RETVAL->client_id) Safefree(RETVAL->client_id);
Safefree(RETVAL);
croak("EV::Gearman: unknown constructor option '%s'", k);
}
}
if (host_sv && path_sv) {
Safefree(RETVAL->rbuf);
Safefree(RETVAL->wbuf);
CLEAR_HANDLER(RETVAL->on_error);
CLEAR_HANDLER(RETVAL->on_connect);
CLEAR_HANDLER(RETVAL->on_disconnect);
CLEAR_HANDLER(RETVAL->loop_sv);
if (RETVAL->client_id) Safefree(RETVAL->client_id);
Safefree(RETVAL);
croak("cannot specify both 'host' and 'path'");
}
RETVAL->port = port;
if (do_reconnect) {
src/EV__Gearman.xs view on Meta::CPAN
void
reconnect(EV::Gearman self, ...)
CODE:
{
/* Optional args: only overwrite delay / max_attempts when the
caller supplies them, so reconnect(1) re-enables without
resetting the values configured at construction time. */
if (items < 2) croak("reconnect: enable arg required");
self->reconnect = SvTRUE(ST(1)) ? 1 : 0;
if (items > 2) {
int d = SvIV(ST(2));
self->reconnect_delay_ms = d >= 0 ? d : 0;
}
if (items > 3) {
int m = SvIV(ST(3));
self->max_reconnect_attempts = m >= 0 ? m : 0;
}
if (!self->reconnect) {
self->reconnect_attempts = 0;
stop_reconnect_timer(self);
}
}
int
reconnect_enabled(EV::Gearman self)
CODE:
RETVAL = self->reconnect;
OUTPUT:
RETVAL
int
priority(EV::Gearman self, ...)
CODE:
{
if (items > 1) {
self->priority = SvIV(ST(1));
if (self->priority < -2) self->priority = -2;
if (self->priority > 2) self->priority = 2;
if (self->reading) {
ev_io_stop(self->loop, &self->rio);
ev_set_priority(&self->rio, self->priority);
ev_io_start(self->loop, &self->rio);
} else {
ev_set_priority(&self->rio, self->priority);
}
if (self->writing) {
ev_io_stop(self->loop, &self->wio);
ev_set_priority(&self->wio, self->priority);
ev_io_start(self->loop, &self->wio);
} else {
ev_set_priority(&self->wio, self->priority);
}
}
RETVAL = self->priority;
}
OUTPUT:
RETVAL
int
keepalive(EV::Gearman self, ...)
CODE:
{
if (items > 1) {
self->keepalive = SvIV(ST(1));
if (self->keepalive < 0) self->keepalive = 0;
if (self->connected && self->fd >= 0)
apply_keepalive(self);
}
RETVAL = self->keepalive;
}
OUTPUT:
RETVAL
# ===== Job object methods (called from Perl via $job->complete etc.) =====
MODULE = EV::Gearman PACKAGE = EV::Gearman::Job
void
_send_event(SV *job_sv, int kind, SV *data_sv = &PL_sv_undef)
CODE:
{
/* kind: 0=complete, 1=fail, 2=exception, 3=data, 4=warning */
STRLEN hl;
const char *h;
ev_gm_t *self = job_resolve(aTHX_ job_sv, &h, &hl, "_send_event");
STRLEN dl = 0;
const char *dp = SvOK(data_sv) ? SvPV(data_sv, dl) : NULL;
static const uint32_t cmds[] = {
GM_CMD_WORK_COMPLETE, GM_CMD_WORK_FAIL, GM_CMD_WORK_EXCEPTION,
GM_CMD_WORK_DATA, GM_CMD_WORK_WARNING,
};
if (kind < 0 || kind >= (int)(sizeof(cmds)/sizeof(cmds[0])))
croak("_send_event: unknown kind %d", kind);
/* WORK_FAIL takes no data; the rest take handle\0data even if data
is empty (the trailing NUL is the separator, not a terminator). */
send_work_event(aTHX_ self, cmds[kind], h, hl,
kind == 1 ? NULL : (dp ? dp : ""), dl);
}
void
_send_status(SV *job_sv, SV *num_sv, SV *denom_sv)
CODE:
{
STRLEN hl;
const char *h;
ev_gm_t *self = job_resolve(aTHX_ job_sv, &h, &hl, "_send_status");
STRLEN nl, dl;
const char *n = SvPV(num_sv, nl);
const char *d = SvPV(denom_sv, dl);
const char *ptrs[3] = { h, n, d };
STRLEN lens[3] = { hl, nl, dl };
size_t blen;
char *body = gm_encode_args(3, ptrs, lens, &blen);
enqueue_packet(aTHX_ self, GM_CMD_WORK_STATUS, body, blen, NULL);
Safefree(body);
}
( run in 1.386 second using v1.01-cache-2.11-cpan-14f38c9f855 )