EV-Memcached
view release on metacpan or search on metacpan
src/EV__Memcached.xs view on Meta::CPAN
int fd;
int connected;
int connecting;
/* IO watchers */
ev_io rio, wio;
int reading, writing;
/* Buffers */
char *rbuf;
size_t rbuf_len, rbuf_cap;
char *wbuf;
size_t wbuf_len, wbuf_off, wbuf_cap;
/* Callbacks */
SV *on_error;
SV *on_connect;
SV *on_disconnect;
/* Command queue */
ngx_queue_t cb_queue;
ngx_queue_t wait_queue;
int pending_count;
int waiting_count;
int max_pending; /* 0 = unlimited */
uint32_t next_opaque;
/* Reconnection */
char *host;
int port;
char *path;
int reconnect;
int reconnect_delay_ms;
int max_reconnect_attempts;
int reconnect_attempts;
ev_timer reconnect_timer;
int reconnect_timer_active;
int intentional_disconnect;
int resume_waiting_on_reconnect;
/* 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;
/* Flow control */
int waiting_timeout_ms;
ev_timer waiting_timer;
int waiting_timer_active;
/* Safety */
int callback_depth;
int in_cb_cleanup;
int in_wait_cleanup;
/* Options */
int priority;
int keepalive;
/* SASL auth */
char *username;
char *password;
};
/* ================================================================
* Shared error strings (initialized in BOOT)
* ================================================================ */
static SV *err_skipped = NULL;
static SV *err_disconnected = NULL;
static SV *err_waiting_timeout = NULL;
/* ================================================================
* 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 waiting_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 arm_cmd_timer(ev_mc_t *self);
static void disarm_cmd_timer(ev_mc_t *self);
static uint32_t mc_enqueue_cmd(pTHX_ ev_mc_t *self,
uint8_t opcode, const char *key, STRLEN key_len,
const char *value, STRLEN value_len,
const char *extras, uint8_t extras_len,
uint64_t cas, int cmd, int quiet, SV *cb);
static void start_reading(ev_mc_t *self);
static void stop_reading(ev_mc_t *self);
static void start_writing(ev_mc_t *self);
static void stop_writing(ev_mc_t *self);
static void start_connect(pTHX_ ev_mc_t *self);
static void cleanup_connection(pTHX_ ev_mc_t *self);
static void emit_error(pTHX_ ev_mc_t *self, const char *msg);
static void handle_disconnect(pTHX_ ev_mc_t *self, const char *reason);
static void schedule_reconnect(pTHX_ ev_mc_t *self);
static void apply_keepalive(ev_mc_t *self);
static void report_connect_error(pTHX_ ev_mc_t *self, const char *errbuf);
static void finish_connect_success(pTHX_ ev_mc_t *self);
static void mc_send_sasl_auth(pTHX_ ev_mc_t *self, SV *cb);
static void stop_connect_timer(ev_mc_t *self);
static void stop_reconnect_timer(ev_mc_t *self);
static void stop_waiting_timer(ev_mc_t *self);
static void send_next_waiting(pTHX_ ev_mc_t *self);
static int check_destroyed(ev_mc_t *self);
static void cancel_pending(pTHX_ ev_mc_t *self, SV *err_sv);
static void cancel_waiting(pTHX_ ev_mc_t *self, SV *err_sv);
/* ================================================================
* Binary protocol helpers (portable, no unaligned access)
* ================================================================ */
static void mc_write_u16(char *buf, uint16_t val) {
val = htons(val);
memcpy(buf, &val, 2);
}
static void mc_write_u32(char *buf, uint32_t val) {
val = htonl(val);
memcpy(buf, &val, 4);
}
static void mc_write_u64(char *buf, uint64_t val) {
uint32_t hi = htonl((uint32_t)(val >> 32));
uint32_t lo = htonl((uint32_t)(val & 0xFFFFFFFF));
memcpy(buf, &hi, 4);
memcpy(buf + 4, &lo, 4);
}
static uint16_t mc_read_u16(const char *buf) {
uint16_t val;
memcpy(&val, buf, 2);
return ntohs(val);
}
static uint32_t mc_read_u32(const char *buf) {
uint32_t val;
memcpy(&val, buf, 4);
return ntohl(val);
}
static uint64_t mc_read_u64(const char *buf) {
uint32_t hi, lo;
memcpy(&hi, buf, 4);
memcpy(&lo, buf + 4, 4);
return ((uint64_t)ntohl(hi) << 32) | ntohl(lo);
}
static void mc_encode_header(char *buf, uint8_t opcode, uint16_t key_len,
uint8_t extras_len, uint32_t body_len, uint32_t opaque, uint64_t cas)
{
buf[0] = MC_REQ_MAGIC;
buf[1] = opcode;
mc_write_u16(buf + 2, key_len);
buf[4] = extras_len;
buf[5] = 0; /* data_type = raw */
mc_write_u16(buf + 6, 0); /* vbucket / reserved */
src/EV__Memcached.xs view on Meta::CPAN
if (result)
mPUSHs(result);
else
PUSHs(&PL_sv_undef);
if (error)
mPUSHs(error);
else
PUSHs(&PL_sv_undef);
PUTBACK;
call_sv(cb, G_DISCARD | G_EVAL);
if (SvTRUE(ERRSV)) {
warn("EV::Memcached: callback error: %s", SvPV_nolen(ERRSV));
sv_setsv(ERRSV, &PL_sv_undef);
}
FREETMPS;
LEAVE;
self->callback_depth--;
}
/* Invoke a user handler with at most one mortal arg; catch exceptions
so a die in user code can't unwind through libev. */
static void invoke_handler(pTHX_ ev_mc_t *self, SV *cb, SV *arg, const char *label) {
if (NULL == cb) { if (arg) SvREFCNT_dec(arg); return; }
/* pin: the callback may clear its own handler ($mc->on_error(undef)),
which would otherwise free the CV while call_sv is still running it */
SvREFCNT_inc_simple_void_NN(cb);
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::Memcached: %s callback error: %s", label, SvPV_nolen(ERRSV));
sv_setsv(ERRSV, &PL_sv_undef);
}
FREETMPS;
LEAVE;
self->callback_depth--;
SvREFCNT_dec(cb);
}
static void emit_error(pTHX_ ev_mc_t *self, const char *msg) {
invoke_handler(aTHX_ self, self->on_error, newSVpv(msg, 0), "on_error");
}
static void emit_connect(pTHX_ ev_mc_t *self) {
invoke_handler(aTHX_ self, self->on_connect, NULL, "on_connect");
}
static void emit_disconnect(pTHX_ ev_mc_t *self) {
invoke_handler(aTHX_ self, self->on_disconnect, NULL, "on_disconnect");
}
static void apply_keepalive(ev_mc_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
}
/* Common tail for synchronous connect-failure paths in start_connect:
emit error, run pending callbacks, and arm reconnect if configured.
Caller returns immediately after invoking. */
static void report_connect_error(pTHX_ ev_mc_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);
} else {
/* terminal failure: waiting commands would otherwise hang until DESTROY */
self->callback_depth++;
cancel_waiting(aTHX_ self, err_disconnected);
self->callback_depth--;
check_destroyed(self);
}
}
/* Shared post-connect-success path. Caller must already have set
self->connected = 1 and stopped/initialized the io watchers. */
static void finish_connect_success(pTHX_ ev_mc_t *self) {
self->reconnect_attempts = 0;
start_reading(self);
apply_keepalive(self);
if (self->username && self->password) {
/* Auto-auth: on_connect fires from the SASL_AUTH success path;
on failure the connection drops without it ever firing. */
mc_send_sasl_auth(aTHX_ self, NULL);
return;
}
emit_connect(aTHX_ self);
if (check_destroyed(self)) return;
send_next_waiting(aTHX_ self);
}
/* ================================================================
* Callback entry management
* ================================================================ */
static ev_mc_cb_t* alloc_cbt(void) {
ev_mc_cb_t *cbt;
Newxz(cbt, 1, ev_mc_cb_t);
return cbt;
}
static void cleanup_cbt(pTHX_ ev_mc_cb_t *cbt) {
CLEAR_HANDLER(cbt->cb);
if (cbt->stats_hv) {
SvREFCNT_dec((SV*)cbt->stats_hv);
cbt->stats_hv = NULL;
}
if ((cbt->cmd == CB_CMD_MGET_FENCE || cbt->cmd == CB_CMD_MGETS_FENCE) && cbt->mget_results) {
SvREFCNT_dec((SV*)cbt->mget_results);
cbt->mget_results = NULL;
}
/* MGET_ENTRY has borrowed ref - don't decrement */
Safefree(cbt);
}
static void cleanup_wait(pTHX_ ev_mc_wait_t *wt) {
CLEAR_HANDLER(wt->cb);
if (wt->packet) { Safefree(wt->packet); wt->packet = NULL; }
if (wt->stats_hv) {
SvREFCNT_dec((SV*)wt->stats_hv);
wt->stats_hv = NULL;
}
if ((wt->cmd == CB_CMD_MGET_FENCE || wt->cmd == CB_CMD_MGETS_FENCE) && wt->mget_results) {
SvREFCNT_dec((SV*)wt->mget_results);
wt->mget_results = NULL;
}
Safefree(wt);
}
/* Free every queued cb/wait entry WITHOUT invoking Perl callbacks.
* Used by the DESTROY teardown paths (deferred and global destruction). */
static void free_all_queues(pTHX_ ev_mc_t *self) {
while (!ngx_queue_empty(&self->cb_queue)) {
ngx_queue_t *q = ngx_queue_head(&self->cb_queue);
ev_mc_cb_t *cbt = ngx_queue_data(q, ev_mc_cb_t, queue);
ngx_queue_remove(q);
cleanup_cbt(aTHX_ cbt);
src/EV__Memcached.xs view on Meta::CPAN
if (strEQ(k, "loop") &&
!(SvROK(v) && SvOBJECT(SvRV(v)) && sv_derived_from(v, "EV::Loop")))
croak("loop must be an EV::Loop object");
if ((strEQ(k, "on_error") || strEQ(k, "on_connect") ||
strEQ(k, "on_disconnect")))
(void)mc_handler_arg(aTHX_ v); /* croaks on bad value */
}
}
Newxz(RETVAL, 1, ev_mc_t);
RETVAL->magic = MC_MAGIC_ALIVE;
RETVAL->fd = -1;
RETVAL->port = 11211;
RETVAL->next_opaque = 1; /* reserve 0 for fire-and-forget quiet ops */
ngx_queue_init(&RETVAL->cb_queue);
ngx_queue_init(&RETVAL->wait_queue);
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: warn. Callback exceptions are caught by
G_EVAL in emit_error, so a `die` would be demoted to a warning
anyway — emit it directly and avoid the double prefix. */
RETVAL->on_error = eval_pv("sub { warn \"EV::Memcached error: @_\\n\" }", TRUE);
SvREFCNT_inc_simple_void_NN(RETVAL->on_error);
/* Parse options */
SV *host_sv = NULL, *path_sv = NULL;
int port = 11211;
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);
SV *h = mc_handler_arg(aTHX_ v);
if (h) RETVAL->on_error = newSVsv(h);
}
else if (strEQ(k, "on_connect")) {
SV *h = mc_handler_arg(aTHX_ v);
if (h) RETVAL->on_connect = newSVsv(h);
}
else if (strEQ(k, "on_disconnect")) {
SV *h = mc_handler_arg(aTHX_ v);
if (h) RETVAL->on_disconnect = newSVsv(h);
}
else if (strEQ(k, "max_pending")) RETVAL->max_pending = SvIV(v);
else if (strEQ(k, "waiting_timeout")) RETVAL->waiting_timeout_ms = SvIV(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, "resume_waiting_on_reconnect")) RETVAL->resume_waiting_on_reconnect = SvTRUE(v) ? 1 : 0;
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, "username")) {
if (SvOK(v)) RETVAL->username = savepv(SvPV_nolen(v));
}
else if (strEQ(k, "password")) {
if (SvOK(v)) RETVAL->password = savepv(SvPV_nolen(v));
}
else if (strEQ(k, "loop")) {
/* EV's T_LOOP convention: pointer in the IV slot of the
blessed referent (SvPVX there is NULL) */
RETVAL->loop = INT2PTR(struct ev_loop *, SvIVX(SvRV(v)));
RETVAL->loop_sv = newSVsv(v); /* keep a user loop alive */
}
}
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->username) Safefree(RETVAL->username);
if (RETVAL->password) Safefree(RETVAL->password);
Safefree(RETVAL);
croak("cannot specify both 'host' and 'path'");
}
RETVAL->port = port;
if (do_reconnect) {
RETVAL->reconnect = 1;
RETVAL->reconnect_delay_ms = reconnect_delay >= 0 ? reconnect_delay : 0;
RETVAL->max_reconnect_attempts = max_reconnect_attempts >= 0 ? max_reconnect_attempts : 0;
}
if (host_sv && SvOK(host_sv)) {
RETVAL->host = savepv(SvPV_nolen(host_sv));
start_connect(aTHX_ RETVAL);
}
else if (path_sv && SvOK(path_sv)) {
RETVAL->path = savepv(SvPV_nolen(path_sv));
start_connect(aTHX_ RETVAL);
}
}
OUTPUT:
RETVAL
void
DESTROY(EV::Memcached self)
CODE:
{
if (self->magic == MC_MAGIC_FREED) return;
/* If we're inside a callback, defer destruction.
* check_destroyed() in io_cb/timer_cb will Safefree after unwind. */
if (self->callback_depth > 0) {
/* Fire pending/waiting callbacks once with "disconnected" (the
DESTRUCTION contract). callback_depth is already >0 and pins
self; refcount 0 means no callback can resurrect it. Under
global destruction, calling into Perl is unsafe: free silently. */
if (!PL_dirty) {
cancel_pending(aTHX_ self, err_disconnected);
cancel_waiting(aTHX_ self, err_disconnected);
}
self->magic = MC_MAGIC_FREED;
/* Stop watchers */
stop_reading(self);
stop_writing(self);
stop_connect_timer(self);
disarm_cmd_timer(self);
src/EV__Memcached.xs view on Meta::CPAN
if (self->command_timeout_ms == 0)
disarm_cmd_timer(self);
else if (!ngx_queue_empty(&self->cb_queue))
arm_cmd_timer(self);
}
RETVAL = self->command_timeout_ms;
}
OUTPUT:
RETVAL
void
reconnect(EV::Memcached self, int enable, int delay_ms = 1000, int max_attempts = 0)
CODE:
{
self->reconnect = enable ? 1 : 0;
self->reconnect_delay_ms = delay_ms >= 0 ? delay_ms : 0;
self->max_reconnect_attempts = max_attempts >= 0 ? max_attempts : 0;
if (!enable) {
self->reconnect_attempts = 0;
stop_reconnect_timer(self);
}
}
int
reconnect_enabled(EV::Memcached self)
CODE:
RETVAL = self->reconnect;
OUTPUT:
RETVAL
int
priority(EV::Memcached self, ...)
CODE:
{
if (items > 1) {
self->priority = SvIV(ST(1));
if (self->priority < -2) self->priority = -2;
if (self->priority > 2) self->priority = 2;
/* Apply to active watchers */
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::Memcached 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
void
skip_pending(EV::Memcached self)
CODE:
{
self->callback_depth++;
skip_pending_impl(aTHX_ self, err_skipped);
self->callback_depth--;
check_destroyed(self);
}
void
skip_waiting(EV::Memcached self)
CODE:
{
self->callback_depth++;
cancel_waiting(aTHX_ self, err_skipped);
self->callback_depth--;
check_destroyed(self);
}
( run in 2.383 seconds using v1.01-cache-2.11-cpan-14f38c9f855 )