/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */

// Functions related to the backend handler thread.

#include "proxy.h"

enum proxy_be_failures {

};

const char *proxy_be_failure_text[] = {

[P_BE_FAIL_TIMEOUT] = "timeout",

[P_BE_FAIL_DISCONNECTED] = "disconnected",

[P_BE_FAIL_CONNECTING] = "connecting",

[P_BE_FAIL_CONNTIMEOUT] = "conntimeout",

[P_BE_FAIL_READVALIDATE] = "readvalidate",

[P_BE_FAIL_BADVALIDATE] = "badvalidate",

[P_BE_FAIL_WRITING] = "writing",

[P_BE_FAIL_READING] = "reading",

[P_BE_FAIL_PARSING] = "parsing",

[P_BE_FAIL_CLOSED] = "closedsock",

[P_BE_FAIL_UNHANDLEDRES] = "unhandledres",

[P_BE_FAIL_OOM] = "outofmemory",

[P_BE_FAIL_ENDSYNC] = "missingend",

[P_BE_FAIL_TRAILINGDATA] = "trailingdata",

[P_BE_FAIL_INVALIDPROTOCOL] = "invalidprotocol",

NULL

};

static void proxy_backend_handler(const int fd, const short which, void *arg);

static void proxy_beconn_handler(const int fd, const short which, void *arg);

static void proxy_event_handler(evutil_socket_t fd, short which, void *arg);

static void proxy_event_beconn(evutil_socket_t fd, short which, void *arg);

static int _prep_pending_write(mcp_backend_t *be);

static void _post_pending_write(mcp_backend_t *be, ssize_t sent);

static int _flush_pending_write(mcp_backend_t *be);

static void _cleanup_backend(mcp_backend_t *be);

static int _reset_bad_backend(mcp_backend_t *be, enum proxy_be_failures err);

static void _backend_failed(mcp_backend_t *be);

static void _set_main_event(mcp_backend_t *be, struct event_base *base, int flags, struct timeval *t, event_callback_fn callback);

static void _stop_main_event(mcp_backend_t *be);

static void _start_write_event(mcp_backend_t *be);

static void _stop_write_event(mcp_backend_t *be);

static void _start_timeout_event(mcp_backend_t *be);

static void _stop_timeout_event(mcp_backend_t *be);

static int proxy_backend_drive_machine(mcp_backend_t *be);

/* Helper routines common to io_uring and libevent modes */

// TODO (v3): doing an inline syscall here, not ideal for uring mode.

// leaving for now since this should be extremely uncommon.

static int _beconn_send_validate(mcp_backend_t *be) {

const char *str = "version\r\n";

const ssize_t len = strlen(str);

ssize_t res = write(mcmc_fd(be->client), str, len);

if (res == -1) {

return -1;

}

// I'm making an opinionated statement that we should be able to write

// "version\r\n" into a fresh socket without hitting EAGAIN.

if (res < len) {

return -1;

}

return 1;

}

// FIXME: make _backend_failed conditionally use _ur() so we can have one call

// in the code and reuse more code like this.

static int _proxy_beconn_checkconnect(mcp_backend_t *be) {

int err = 0;

// We were connecting, now ensure we're properly connected.

if (mcmc_check_nonblock_connect(be->client, &err) != MCMC_OK) {

P_DEBUG("%s: backend failed to connect (%s:%s)\n", __func__, be->name, be->port);

// kick the bad backend, clear the queue, retry later.

// FIXME (v2): if a connect fails, anything currently in the queue

// should be safe to hold up until their timeout.

_reset_bad_backend(be, P_BE_FAIL_CONNECTING);

_backend_failed(be);

return -1;

}

P_DEBUG("%s: backend connected (%s:%s)\n", __func__, be->name, be->port);

be->connecting = false;

be->state = mcp_backend_read;

be->bad = false;

be->failed_count = 0;

be->validating = true;

// TODO: make validation optional.

if (_beconn_send_validate(be) == -1) {

_reset_bad_backend(be, P_BE_FAIL_BADVALIDATE);

_backend_failed(be);

return -1;

} else {

// buffer should be empty during validation stage.

assert(be->rbufused == 0);

return 0;

}

}

static int _proxy_event_handler_dequeue(proxy_event_thread_t *t) {

io_head_t head;

STAILQ_INIT(&head);

STAILQ_INIT(&t->be_head);

// Pull the entire stack of inbound into local queue.

pthread_mutex_lock(&t->mutex);

STAILQ_CONCAT(&head, &t->io_head_in);

pthread_mutex_unlock(&t->mutex);

int io_count = 0;

int be_count = 0;

while (!STAILQ_EMPTY(&head)) {

io_pending_proxy_t *io = STAILQ_FIRST(&head);

io->flushed = false;

// _no_ mutex on backends. they are owned by the event thread.

STAILQ_REMOVE_HEAD(&head, io_next);

// paranoia about moving items between lists.

io->io_next.stqe_next = NULL;

mcp_backend_t *be = io->backend;

// So the backend can retrieve its event base.

if (be->bad) {

P_DEBUG("%s: fast failing request to bad backend\n", __func__);

io->client_resp->status = MCMC_ERR;

return_io_pending((io_pending_t *)io);

continue;

}

STAILQ_INSERT_TAIL(&be->io_head, io, io_next);

be->depth++;

io_count++;

if (!be->stacked) {

be->stacked = true;

// more paranoia about be_next not being overwritten

be->be_next.stqe_next = NULL;

STAILQ_INSERT_TAIL(&t->be_head, be, be_next);

be_count++;

}

}

//P_DEBUG("%s: io/be counts for syscalls [%d/%d]\n", __func__, io_count, be_count);

return io_count;

}

static void _cleanup_backend(mcp_backend_t *be) {

// remove any pending events.

int pending = event_pending(&be->main_event, EV_READ|EV_WRITE|EV_TIMEOUT, NULL);

if ((pending & (EV_READ|EV_WRITE|EV_TIMEOUT)) != 0) {

event_del(&be->main_event); // an error to call event_del() without event.

}

pending = event_pending(&be->write_event, EV_READ|EV_WRITE|EV_TIMEOUT, NULL);

if ((pending & (EV_READ|EV_WRITE|EV_TIMEOUT)) != 0) {

event_del(&be->write_event); // an error to call event_del() without event.

}

pending = event_pending(&be->timeout_event, EV_TIMEOUT, NULL);

if ((pending & (EV_TIMEOUT)) != 0) {

event_del(&be->timeout_event); // an error to call event_del() without event.

}

// - assert on empty queue

assert(STAILQ_EMPTY(&be->io_head));

mcmc_disconnect(be->client);

// - free be->client

free(be->client);

// - free be->rbuf

free(be->rbuf);

// - free *be

free(be);

}

// event handler for injecting backends for processing

// currently just for initiating connections the first time.

static void proxy_event_beconn(evutil_socket_t fd, short which, void *arg) {

proxy_event_thread_t *t = arg;

#ifdef USE_EVENTFD

uint64_t u;

if (read(fd, &u, sizeof(uint64_t)) != sizeof(uint64_t)) {

// Temporary error or wasn't actually ready to read somehow.

return;

}

#else

char buf[1];

if (read(fd, buf, 1) != 1) {

P_DEBUG("%s: pipe read failed\n", __func__);

return;

}

#endif

beconn_head_t head;

STAILQ_INIT(&head);

pthread_mutex_lock(&t->mutex);

STAILQ_CONCAT(&head, &t->beconn_head_in);

pthread_mutex_unlock(&t->mutex);

// Think we should reuse this code path for manually instructing backends

// to disable/etc but not coding for that generically. We just need to

// check the state of the backend when it reaches here or some flags at

// least.

// FIXME: another ->stacked flag?

// Either that or remove the STAILQ code and just using an array of

// ptr's.

mcp_backend_t *be = NULL;

// be can be freed by the loop, so can't use STAILQ_FOREACH.

while (!STAILQ_EMPTY(&head)) {

be = STAILQ_FIRST(&head);

STAILQ_REMOVE_HEAD(&head, beconn_next);

if (be->transferred) {

// If this object was already transferred here, we're being

// signalled to clean it up and free.

_cleanup_backend(be);

} else {

be->transferred = true;

// assign the initial events to the backend, so we don't have to

// constantly check if they were initialized yet elsewhere.

// note these events will not fire until event_add() is called.

int status = mcmc_connect(be->client, be->name, be->port, be->connect_flags);

event_assign(&be->main_event, be->event_thread->base, mcmc_fd(be->client), EV_WRITE|EV_TIMEOUT, proxy_beconn_handler, be);

event_assign(&be->write_event, be->event_thread->base, mcmc_fd(be->client), EV_WRITE|EV_TIMEOUT, proxy_backend_handler, be);

event_assign(&be->timeout_event, be->event_thread->base, -1, EV_TIMEOUT, proxy_backend_handler, be);

if (status == MCMC_CONNECTING || status == MCMC_CONNECTED) {

// if we're already connected for some reason, still push it

// through the connection handler to keep the code unified. It

// will auto-wake because the socket is writeable.

be->connecting = true;

be->can_write = false;

// kick off the event we intialized above.

event_add(&be->main_event, &be->tunables.connect);

} else {

_reset_bad_backend(be, P_BE_FAIL_CONNECTING);

_backend_failed(be);

}

}

}

}

void proxy_run_backend_queue(be_head_t *head) {

mcp_backend_t *be;

STAILQ_FOREACH(be, head, be_next) {

be->stacked = false;

int flags = 0;

if (be->bad) {

// flush queue if backend is still bad.

// TODO: duplicated from _reset_bad_backend()

io_pending_proxy_t *io = NULL;

while (!STAILQ_EMPTY(&be->io_head)) {

io = STAILQ_FIRST(&be->io_head);

STAILQ_REMOVE_HEAD(&be->io_head, io_next);

io->client_resp->status = MCMC_ERR;

be->depth--;

return_io_pending((io_pending_t *)io);

}

} else if (be->connecting || be->validating) {

P_DEBUG("%s: deferring IO pending connecting (%s:%s)\n", __func__, be->name, be->port);

} else {

flags = _flush_pending_write(be);

if (flags == -1) {

_reset_bad_backend(be, P_BE_FAIL_WRITING);

_backend_failed(be);

} else if (flags & EV_WRITE) {

// only get here because we need to kick off the write handler

_start_write_event(be);

}

if (be->pending_read) {

_start_timeout_event(be);

}

}

}

}

// event handler for executing backend requests

static void proxy_event_handler(evutil_socket_t fd, short which, void *arg) {

proxy_event_thread_t *t = arg;

#ifdef USE_EVENTFD

uint64_t u;

if (read(fd, &u, sizeof(uint64_t)) != sizeof(uint64_t)) {

// Temporary error or wasn't actually ready to read somehow.

return;

}

#else

char buf[1];

// TODO (v2): This is a lot more fatal than it should be. can it fail? can

// it blow up the server?

// TODO (v2): a cross-platform method of speeding this up would be nice. With

// event fds we can queue N events and wakeup once here.

// If we're pulling one byte out of the pipe at a time here it'll just

// wake us up too often.

// If the pipe is O_NONBLOCK then maybe just a larger read would work?

if (read(fd, buf, 1) != 1) {

P_DEBUG("%s: pipe read failed\n", __func__);

return;

}

#endif

if (_proxy_event_handler_dequeue(t) == 0) {

//P_DEBUG("%s: no IO's to complete\n", __func__);

return;

}

// Re-walk each backend and check set event as required.

proxy_run_backend_queue(&t->be_head);

}

void *proxy_event_thread(void *arg) {

proxy_event_thread_t *t = arg;

logger_create(); // TODO (v2): add logger ptr to structure

event_base_loop(t->base, 0);

event_base_free(t->base);

// TODO (v2): join bt threads, free array.

return NULL;

}

static void _set_main_event(mcp_backend_t *be, struct event_base *base, int flags, struct timeval *t, event_callback_fn callback) {

int pending = event_pending(&be->main_event, EV_READ|EV_WRITE, NULL);

if ((pending & (EV_READ|EV_WRITE)) != 0) {

event_del(&be->main_event); // replace existing event.

}

event_assign(&be->main_event, base, mcmc_fd(be->client),

flags, callback, be);

event_add(&be->main_event, t);

}

static void _stop_main_event(mcp_backend_t *be) {

int pending = event_pending(&be->main_event, EV_READ|EV_WRITE, NULL);

if ((pending & (EV_READ|EV_WRITE|EV_TIMEOUT)) == 0) {

return;

}

event_del(&be->write_event);

}

static void _start_write_event(mcp_backend_t *be) {

int pending = event_pending(&be->main_event, EV_WRITE, NULL);

if ((pending & (EV_WRITE|EV_TIMEOUT)) != 0) {

return;

}

// FIXME: wasn't there a write timeout?

event_add(&be->write_event, &be->tunables.read);

}

static void _stop_write_event(mcp_backend_t *be) {

int pending = event_pending(&be->main_event, EV_WRITE, NULL);

if ((pending & (EV_WRITE|EV_TIMEOUT)) == 0) {

return;

}

event_del(&be->write_event);

}

// handle the read timeouts with a side event, so we can stick with a

// persistent listener (optimization + catch disconnects faster)

static void _start_timeout_event(mcp_backend_t *be) {

int pending = event_pending(&be->timeout_event, EV_TIMEOUT, NULL);

if ((pending & (EV_TIMEOUT)) != 0) {

return;

}

event_add(&be->timeout_event, &be->tunables.read);

}

static void _stop_timeout_event(mcp_backend_t *be) {

int pending = event_pending(&be->timeout_event, EV_TIMEOUT, NULL);

if ((pending & (EV_TIMEOUT)) == 0) {

return;

}

event_del(&be->timeout_event);

}

static void _drive_machine_next(mcp_backend_t *be, io_pending_proxy_t *p) {

struct timeval end;

// set the head here. when we break the head will be correct.

STAILQ_REMOVE_HEAD(&be->io_head, io_next);

be->depth--;

be->pending_read--;

// stamp the elapsed time into the response object.

gettimeofday(&end, NULL);

p->client_resp->elapsed = (end.tv_sec - p->client_resp->start.tv_sec) * 1000000 +

(end.tv_usec - p->client_resp->start.tv_usec);

// have to do the q->count-- and == 0 and redispatch_conn()

// stuff here. The moment we call return_io here we

// don't own *p anymore.

return_io_pending((io_pending_t *)p);

be->state = mcp_backend_read;

}

// NOTES:

// - mcp_backend_read: grab req_stack_head, do things

// read -> next, want_read -> next | read_end, etc.

static int proxy_backend_drive_machine(mcp_backend_t *be) {

bool stop = false;

io_pending_proxy_t *p = NULL;

int flags = 0;

p = STAILQ_FIRST(&be->io_head);

if (p == NULL) {

// got a read event, but nothing was queued.

// probably means a disconnect event.

// TODO (v2): could probably confirm this by attempting to read the

// socket, getsockopt, or something else simply for logging or

// statistical purposes.

// In this case we know it's going to be a close so error.

flags = P_BE_FAIL_CLOSED;

P_DEBUG("%s: read event but nothing in IO queue\n", __func__);

return flags;

}

while (!stop) {

mcp_resp_t *r;

switch(be->state) {

case mcp_backend_read:

assert(p != NULL);

// FIXME: remove the _read state?

be->state = mcp_backend_parse;

break;

case mcp_backend_parse:

r = p->client_resp;

r->status = mcmc_parse_buf(be->client, be->rbuf, be->rbufused, &r->resp);

// Quick check if we need more data.

if (r->resp.code == MCMC_WANT_READ) {

return 0;

}

// we actually don't care about anything but the value length

// TODO (v2): if vlen != vlen_read, pull an item and copy the data.

int extra_space = 0;

// if all goes well, move to the next request.

be->state = mcp_backend_next;

switch (r->resp.type) {

case MCMC_RESP_GET:

// We're in GET mode. we only support one key per

// GET in the proxy backends, so we need to later check

// for an END.

extra_space = ENDLEN;

be->state = mcp_backend_read_end;

break;

case MCMC_RESP_END:

// this is a MISS from a GET request

// or final handler from a STAT request.

assert(r->resp.vlen == 0);

if (p->ascii_multiget) {

// Ascii multiget hack mode; consume END's

be->rbufused -= r->resp.reslen;

if (be->rbufused > 0) {

memmove(be->rbuf, be->rbuf+r->resp.reslen, be->rbufused);

}

be->state = mcp_backend_next;

continue;

}

break;

case MCMC_RESP_META:

// we can handle meta responses easily since they're self

// contained.

break;

case MCMC_RESP_GENERIC:

case MCMC_RESP_NUMERIC:

break;

case MCMC_RESP_ERRMSG: // received an error message

if (r->resp.code != MCMC_CODE_SERVER_ERROR) {

// Non server errors are protocol errors; can't trust

// the connection anymore.

be->state = mcp_backend_next_close;

}

break;

case MCMC_RESP_FAIL:

P_DEBUG("%s: mcmc_read failed [%d]\n", __func__, r->status);

flags = P_BE_FAIL_PARSING;

stop = true;

break;

// TODO (v2): No-op response?

default:

P_DEBUG("%s: Unhandled response from backend: %d\n", __func__, r->resp.type);

// unhandled :(

flags = P_BE_FAIL_UNHANDLEDRES;

stop = true;

break;

}

// r->resp.reslen + r->resp.vlen is the total length of the response.

// TODO (v2): need to associate a buffer with this response...

// for now we simply malloc, but reusable buffers should be used

r->blen = r->resp.reslen + r->resp.vlen;

{

bool oom = proxy_bufmem_checkadd(r->thread, r->blen + extra_space);

if (oom) {

flags = P_BE_FAIL_OOM;

// need to zero out blen so we don't over-decrement later

r->blen = 0;

stop = true;

break;

}

}

r->buf = malloc(r->blen + extra_space);

if (r->buf == NULL) {

// Enforce accounting.

pthread_mutex_lock(&r->thread->proxy_limit_lock);

r->thread->proxy_buffer_memory_used -= r->blen + extra_space;

pthread_mutex_unlock(&r->thread->proxy_limit_lock);

flags = P_BE_FAIL_OOM;

r->blen = 0;

stop = true;

break;

}

P_DEBUG("%s: r->status: %d, r->bread: %d, r->vlen: %lu\n", __func__, r->status, r->bread, r->resp.vlen);

if (r->resp.vlen != r->resp.vlen_read) {

// shouldn't be possible to have excess in buffer

// if we're dealing with a partial value.

assert(be->rbufused == r->resp.reslen+r->resp.vlen_read);

P_DEBUG("%s: got a short read, moving to want_read\n", __func__);

// copy the partial and advance mcmc's buffer digestion.

memcpy(r->buf, be->rbuf, r->resp.reslen + r->resp.vlen_read);

r->bread = r->resp.reslen + r->resp.vlen_read;

be->rbufused = 0;

be->state = mcp_backend_want_read;

flags = 0;

stop = true;

break;

} else {

// mcmc's already counted the value as read if it fit in

// the original buffer...

memcpy(r->buf, be->rbuf, r->resp.reslen+r->resp.vlen_read);

}

// had a response, advance the buffer.

be->rbufused -= r->resp.reslen + r->resp.vlen_read;

if (be->rbufused > 0) {

memmove(be->rbuf, be->rbuf+r->resp.reslen+r->resp.vlen_read, be->rbufused);

}

break;

case mcp_backend_read_end:

r = p->client_resp;

// we need to ensure the next data in the stream is "END\r\n"

// if not, the stack is desynced and we lose it.

if (be->rbufused >= ENDLEN) {

if (memcmp(be->rbuf, ENDSTR, ENDLEN) != 0) {

flags = P_BE_FAIL_ENDSYNC;

stop = true;

break;

} else {

// response is good.

// FIXME (v2): copy what the server actually sent?

if (!p->ascii_multiget) {

// sigh... if part of a multiget we need to eat the END

// markers down here.

memcpy(r->buf+r->blen, ENDSTR, ENDLEN);

r->blen += 5;

} else {

r->extra = 5;

}

// advance buffer

be->rbufused -= ENDLEN;

if (be->rbufused > 0) {

memmove(be->rbuf, be->rbuf+ENDLEN, be->rbufused);

}

}

} else {

flags = 0;

stop = true;

break;

}

be->state = mcp_backend_next;

break;

case mcp_backend_want_read:

// Continuing a read from earlier

r = p->client_resp;

// take bread input and see if we're done reading the value,

// else advance, set buffers, return next.

P_DEBUG("%s: [want_read] r->bread: %d vlen: %lu\n", __func__, r->bread, r->resp.vlen);

assert(be->rbufused != 0);

size_t tocopy = be->rbufused < r->blen - r->bread ?

be->rbufused : r->blen - r->bread;

memcpy(r->buf+r->bread, be->rbuf, tocopy);

r->bread += tocopy;

if (r->bread >= r->blen) {

// all done copying data.

if (r->resp.type == MCMC_RESP_GET) {

be->state = mcp_backend_read_end;

} else {

be->state = mcp_backend_next;

}

// shuffle remaining buffer.

be->rbufused -= tocopy;

if (be->rbufused > 0) {

memmove(be->rbuf, be->rbuf+tocopy, be->rbufused);

}

} else {

assert(tocopy == be->rbufused);

// signal to caller to issue a read.

be->rbufused = 0;

flags = 0;

stop = true;

}

break;

case mcp_backend_next:

_drive_machine_next(be, p);

if (STAILQ_EMPTY(&be->io_head)) {

stop = true;

// if there're no pending requests, the read buffer

// should also be empty.

if (be->rbufused > 0) {

flags = P_BE_FAIL_TRAILINGDATA;

}

break;

} else {

p = STAILQ_FIRST(&be->io_head);

}

// if leftover, keep processing IO's.

// if no more data in buffer, need to re-set stack head and re-set

// event.

P_DEBUG("%s: [next] remain: %lu\n", __func__, be->rbufused);

if (be->rbufused != 0) {

// data trailing in the buffer, for a different request.

be->state = mcp_backend_parse;

} else {

// need to read more data, buffer is empty.

stop = true;

}

break;

case mcp_backend_next_close:

// we advance and return the current IO, then kill the conn.

_drive_machine_next(be, p);

stop = true;

flags = P_BE_FAIL_INVALIDPROTOCOL;

break;

default:

// TODO (v2): at some point (after v1?) this should attempt to recover,

// though we should only get here from memory corruption and

// bailing may be the right thing to do.

fprintf(stderr, "%s: invalid backend state: %d\n", __func__, be->state);

assert(false);

} // switch

} // while

return flags;

}

static void _backend_reconnect(mcp_backend_t *be) {

int status = mcmc_connect(be->client, be->name, be->port, be->connect_flags);

if (status == MCMC_CONNECTED) {

// TODO (v2): unexpected but lets let it be here.

be->connecting = false;

be->can_write = true;

} else if (status == MCMC_CONNECTING) {

be->connecting = true;

be->can_write = false;

} else {

// failed to immediately re-establish the connection.

// need to put the BE into a bad/retry state.

be->connecting = false;

be->can_write = true;

}

// re-create the write handler for the new file descriptor.

// the main event will be re-assigned after this call.

event_assign(&be->write_event, be->event_thread->base, mcmc_fd(be->client), EV_WRITE|EV_TIMEOUT, proxy_backend_handler, be);

// do not need to re-assign the timer event because it's not tied to fd

}

// All we need to do here is schedule the backend to attempt to connect again.

static void proxy_backend_retry_handler(const int fd, const short which, void *arg) {

mcp_backend_t *be = arg;

assert(which & EV_TIMEOUT);

struct timeval tmp_time = be->tunables.connect;

_backend_reconnect(be);

_set_main_event(be, be->event_thread->base, EV_WRITE, &tmp_time, proxy_beconn_handler);

}

// must be called after _reset_bad_backend(), so the backend is currently

// clear.

// TODO (v2): extra counter for "backend connect tries" so it's still possible

// to see dead backends exist

static void _backend_failed(mcp_backend_t *be) {

if (++be->failed_count > be->tunables.backend_failure_limit) {

struct timeval tmp_time = be->tunables.retry;

if (!be->bad) {

P_DEBUG("%s: marking backend as bad\n", __func__);

STAT_INCR(be->event_thread->ctx, backend_marked_bad, 1);

LOGGER_LOG(NULL, LOG_PROXYEVENTS, LOGGER_PROXY_BE_ERROR, NULL, "markedbad", be->name, be->port, 0, NULL, 0);

}

be->bad = true;

_set_main_event(be, be->event_thread->base, EV_TIMEOUT, &tmp_time, proxy_backend_retry_handler);

} else {

struct timeval tmp_time = be->tunables.connect;

STAT_INCR(be->event_thread->ctx, backend_failed, 1);

_backend_reconnect(be);

_set_main_event(be, be->event_thread->base, EV_WRITE, &tmp_time, proxy_beconn_handler);

}

}

// TODO (v2): add a second argument for assigning a specific error to all pending

// IO's (ie; timeout).

// The backend has gotten into a bad state (timed out, protocol desync, or

// some other supposedly unrecoverable error: purge the queue and

// cycle the socket.

// Note that some types of errors may not require flushing the queue and

// should be fixed as they're figured out.

// _must_ be called from within the event thread.

static int _reset_bad_backend(mcp_backend_t *be, enum proxy_be_failures err) {

io_pending_proxy_t *io = NULL;

P_DEBUG("%s: resetting bad backend: %s\n", __func__, proxy_be_failure_text[err]);

// Can't use STAILQ_FOREACH() since return_io_pending() free's the current

// io. STAILQ_FOREACH_SAFE maybe?

int depth = be->depth;

while (!STAILQ_EMPTY(&be->io_head)) {

io = STAILQ_FIRST(&be->io_head);

STAILQ_REMOVE_HEAD(&be->io_head, io_next);

// TODO (v2): Unsure if this is the best way of surfacing errors to lua,

// but will do for V1.

io->client_resp->status = MCMC_ERR;

be->depth--;

return_io_pending((io_pending_t *)io);

}

STAILQ_INIT(&be->io_head);

be->io_next = NULL; // also reset the write offset.

// Only log if we don't already know it's messed up.

if (!be->bad) {

LOGGER_LOG(NULL, LOG_PROXYEVENTS, LOGGER_PROXY_BE_ERROR, NULL, proxy_be_failure_text[err], be->name, be->port, depth, be->rbuf, be->rbufused);

}

// reset buffer to blank state.

be->rbufused = 0;

be->pending_read = 0;

// allow the _backend_failed() routine to connect when ready.

_stop_write_event(be);

_stop_main_event(be);

_stop_timeout_event(be);

mcmc_disconnect(be->client);

// we leave the main event alone, because be_failed() always overwrites.

return 0;

}

static int _prep_pending_write(mcp_backend_t *be) {

struct iovec *iovs = be->write_iovs;

io_pending_proxy_t *io = NULL;

int iovused = 0;

if (be->io_next == NULL) {

// separate pointer for how far into the list we've flushed.

io = STAILQ_FIRST(&be->io_head);

} else {

io = be->io_next;

}

assert(io != NULL);

for (; io; io = STAILQ_NEXT(io, io_next)) {

// TODO (v2): paranoia for now, but this check should never fire

if (io->flushed)

continue;

if (io->iovcnt + iovused > BE_IOV_MAX) {

// We will need to keep writing later.

break;

}

memcpy(&iovs[iovused], io->iov, sizeof(struct iovec)*io->iovcnt);

iovused += io->iovcnt;

}

return iovused;

}

// returns true if any pending writes were fully flushed.

static void _post_pending_write(mcp_backend_t *be, ssize_t sent) {

io_pending_proxy_t *io = be->io_next;

if (io == NULL) {

io = STAILQ_FIRST(&be->io_head);

}

for (; io; io = STAILQ_NEXT(io, io_next)) {

bool flushed = true;

if (io->flushed)

continue;

if (sent >= io->iovbytes) {

// short circuit for common case.

sent -= io->iovbytes;

} else {

io->iovbytes -= sent;

for (int x = 0; x < io->iovcnt; x++) {

struct iovec *iov = &io->iov[x];

if (sent >= iov->iov_len) {

sent -= iov->iov_len;

iov->iov_len = 0;

} else {

iov->iov_len -= sent;

iov->iov_base = (char *)iov->iov_base + sent;

sent = 0;

flushed = false;

break;

}

}

}

io->flushed = flushed;

if (flushed) {

be->pending_read++;

}

if (sent <= 0) {

// really shouldn't be negative, though.

assert(sent >= 0);

break;

}

} // for

// resume the flush from this point.

if (io != NULL && !io->flushed) {

be->io_next = io;

} else {

be->io_next = NULL;

}

}

static int _flush_pending_write(mcp_backend_t *be) {

int flags = 0;

// Allow us to be called with an empty stack to prevent dev errors.

if (STAILQ_EMPTY(&be->io_head)) {

return 0;

}

int iovcnt = _prep_pending_write(be);

ssize_t sent = writev(mcmc_fd(be->client), be->write_iovs, iovcnt);

if (sent > 0) {

_post_pending_write(be, sent);

// still have unflushed pending IO's, check for write and re-loop.

if (be->io_next) {

be->can_write = false;

flags |= EV_WRITE;

}

} else if (sent == -1) {

if (errno == EAGAIN || errno == EWOULDBLOCK) {

be->can_write = false;

flags |= EV_WRITE;

} else {

flags = -1;

}

}

return flags;

}

// Libevent handler for backends in a connecting state.

static void proxy_beconn_handler(const int fd, const short which, void *arg) {

assert(arg != NULL);

mcp_backend_t *be = arg;

int flags = EV_TIMEOUT;

struct timeval tmp_time = be->tunables.read;

if (which & EV_TIMEOUT) {

P_DEBUG("%s: backend timed out while connecting\n", __func__);

if (be->connecting) {

_reset_bad_backend(be, P_BE_FAIL_CONNTIMEOUT);

} else {

_reset_bad_backend(be, P_BE_FAIL_READVALIDATE);

}

_backend_failed(be);

return;

}

if (which & EV_WRITE) {

be->can_write = true;

if (be->connecting) {

if (_proxy_beconn_checkconnect(be) == -1) {

return;

}

_set_main_event(be, be->event_thread->base, EV_READ, &tmp_time, proxy_beconn_handler);

}

// TODO: currently never taken, until validation is made optional.

if (!be->validating) {

int res = _flush_pending_write(be);

if (res == -1) {

_reset_bad_backend(be, P_BE_FAIL_WRITING);

_backend_failed(be);

return;

}

flags |= res;

// FIXME: set write event?

}

}

if (which & EV_READ) {

assert(be->validating);

int read = recv(mcmc_fd(be->client), be->rbuf + be->rbufused, READ_BUFFER_SIZE - be->rbufused, 0);

if (read > 0) {

mcmc_resp_t r;

be->rbufused += read;

int status = mcmc_parse_buf(be->client, be->rbuf, be->rbufused, &r);

if (status == MCMC_ERR) {

// Needed more data for a version line, somehow. I feel like

// this should set off some alarms, but it is possible.

if (r.code == MCMC_WANT_READ) {

_set_main_event(be, be->event_thread->base, EV_READ, &tmp_time, proxy_beconn_handler);

return;

}

_reset_bad_backend(be, P_BE_FAIL_READVALIDATE);

_backend_failed(be);

return;

}

if (r.code != MCMC_CODE_VERSION) {

_reset_bad_backend(be, P_BE_FAIL_BADVALIDATE);

_backend_failed(be);

return;

}

be->validating = false;

be->rbufused = 0;

} else if (read == 0) {

// not connected or error.

_reset_bad_backend(be, P_BE_FAIL_DISCONNECTED);

_backend_failed(be);

return;

} else if (read == -1) {

// sit on epoll again.

if (errno != EAGAIN && errno != EWOULDBLOCK) {

_reset_bad_backend(be, P_BE_FAIL_READING);

_backend_failed(be);

return;

}

_set_main_event(be, be->event_thread->base, EV_READ, &tmp_time, proxy_beconn_handler);

return;

}

// Passed validation, don't need to re-read, flush any pending writes.

int res = _flush_pending_write(be);

if (res == -1) {

_reset_bad_backend(be, P_BE_FAIL_WRITING);

_backend_failed(be);

return;

}

if (flags & EV_WRITE) {

_start_write_event(be);

}

if (be->pending_read) {

_start_timeout_event(be);

}

}

// switch to the primary persistent read event.

if (!be->validating) {

_set_main_event(be, be->event_thread->base, EV_READ|EV_PERSIST, NULL, proxy_backend_handler);

}

}

// The libevent backend callback handler.

// If we end up resetting a backend, it will get put back into a connecting

// state.

static void proxy_backend_handler(const int fd, const short which, void *arg) {

mcp_backend_t *be = arg;