rough framework

[lemu] / connections.c

#define _GNU_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdarg.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>

#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netdb.h>
#ifndef NI_MAXHOST
#       define NI_MAXHOST 1025
#endif
#ifndef NI_MAXSERV
#       define NI_MAXSERV 32
#endif

#include <sys/resource.h> // needs _GNU_SOURCE for RUSAGE_THREAD

#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <event2/event.h>
#include <event2/util.h>

#include <ossp/uuid.h>

#include "common.h"
#include "notify.h"
#include "connections.h"
#include "command.h"
#include "server.h"

/**
 *  A reference-counted data buffer.
 *  Used internally here for sending out one message to multiple connections, without
 *  creating a copy for each connection.
 *  Frees data pointer when the last reference is removed.
 *  A connection thread increases the reference count as it sends the message to each
 *  destination connection, while the libevent thread decrements as it completes each
 *  transmission.
 */
struct rc_data_ {
        pthread_mutex_t mutex;
        size_t reference_count;
        size_t data_len;
        char *data;
};

/**
 *  Allocates, initializes, and returns a new struct rc_data_ entity.
 */
static struct rc_data_ *
rc_data_new_(char *data, size_t len)
{
        struct rc_data_ *rc;
        int r;

        if (!data) {
                NOTIFY_DEBUG("data:%p len:%zu", data, len);
                return NULL;
        }

        rc = malloc(sizeof *rc);
        if (!rc) {
                NOTIFY_ERROR("malloc(%zu): %s", sizeof *rc, strerror(errno));
                return NULL;
        }

        if ( (r = pthread_mutex_init(&rc->mutex, NULL)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_init", strerror(r));
                free(rc);
                return NULL;
        }

        rc->reference_count = 1;
        rc->data_len = len;
        rc->data = data;

        return rc;
}

/**
 *  Increments the reference count of a struct rc_data_ entity.
 */
static void
rc_data_ref_inc_(struct rc_data_ *rc)
{
        int r;

        if ( (r = pthread_mutex_lock(&rc->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
                return;
        }

        rc->reference_count += 1;

        if ( (r = pthread_mutex_unlock(&rc->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
        }
}

/**
 *  Decrements the reference count of a struct rc_data_ entity
 *  and frees everything if there are no more references.
 */
static void
rc_data_free_(struct rc_data_ *rc)
{
        int r;

        if ( (r = pthread_mutex_lock(&rc->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
                return;
        }
        rc->reference_count -= 1;
        if (rc->reference_count > 0) {
                if ( (r = pthread_mutex_unlock(&rc->mutex)) ) {
                        NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
                }
                return;
        }

        free(rc->data);
        rc->data = NULL;
        if ( (r = pthread_mutex_unlock(&rc->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
        }
        if ( (r = pthread_mutex_destroy(&rc->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_destroy", strerror(r));
        }

        memset(rc, 0, sizeof *rc);

        free(rc);
}

/**
 *  Wrapper for rc_data_free_() of the proper callback type to be used by
 *  evbuffer_add_reference() in connection_multi_send().
 */
static void
rc_cleanup_cb_(const void *data UNUSED, size_t len UNUSED, void *arg)
{
        struct rc_data_ *r = arg;

        rc_data_free_(r);
}

/**
 *  Initializes a struct connections entity.
 */
int
connections_init(struct connections *cs)
{
        int r;

        cs->count = 0;
        TAILQ_INIT(&cs->head);

        if ( (r = pthread_mutex_init(&cs->mutex, NULL)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_init", strerror(r));
                return -1;
        }

        return 0;
}

/**
 * Cleanup a struct connections.
 */
void
connections_fini(struct connections *cs)
{
        struct connection *c1, *c2;
        int r;

#if DEBUG_SILLY
        NOTIFY_DEBUG("cs:%p", cs);
#endif

        /* free any connection entities */
        c1 = TAILQ_FIRST(&cs->head);
        while (c1 != NULL) {
                c2 = TAILQ_NEXT(c1, tailq_entry);
                connection_free(c1);
                c1 = c2;
        }
        TAILQ_INIT(&cs->head);
        cs->count = 0;

        if ( (r = pthread_mutex_destroy(&cs->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_destroy", strerror(r));
        }
}


/**
 *  Inserts a connection at the end of connections list.
 */
void
connections_append(struct connection *c)
{
        struct connections *cs = &c->server->connections;
        int r;

#if DEBUG_SILLY
        NOTIFY_DEBUG("c:%p", c);
#endif

        if ( (r = pthread_mutex_lock(&cs->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
                return;
        }

        TAILQ_INSERT_TAIL(&cs->head, c, tailq_entry);
        cs->count += 1;

        if ( (r = pthread_mutex_unlock(&cs->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
                return;
        }
}

/**
 *  Removes a connection from its struct connections list.
 */
void
connections_remove(struct connection *c)
{
        struct connections *cs = &c->server->connections;
        int r;

#if DEBUG_SILLY
        NOTIFY_DEBUG("c:%p", c);
#endif

        if ( !c || !cs) {
                NOTIFY_DEBUG("c:%p connections:%p", c, cs);
                return;
        }

        if ( (r = pthread_mutex_lock(&cs->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
                return;
        }

        TAILQ_REMOVE(&cs->head, c, tailq_entry);
        cs->count -= 1;

        if ( (r = pthread_mutex_unlock(&cs->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
                return;
        }
}

/**
 *  Allocates and returns a null-terminated array of pointers to all connection entities in
 *  connections list, minus the 'c_exclude' connection, if set.
 *  Used as a simple implementation of broadcasting a message.
 *  This increments the reference count of each connection added to the array
 *  so they need to individually be freed before freeing the array.
 */
struct connection **
connections_all_as_array(struct connections *cs, struct connection *c_exclude)
{
        struct connection **all;
        struct connection *c;
        size_t i = 0;
        int r;

        if ( (r = pthread_mutex_lock(&cs->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
                return NULL;
        }

        all = calloc(cs->count + 1, sizeof *all);
        if (!all) {
                NOTIFY_ERROR("calloc(%zu, %zu): %s", cs->count + 1, sizeof *all, strerror(errno));
                goto err_unlock;
        }

        TAILQ_FOREACH(c, &cs->head, tailq_entry) {
                if (c != c_exclude
                &&  c->state < CONNECTION_STATE_WANT_CLOSE
                &&  c->state > CONNECTION_STATE_INIT) {
                        connection_inc_ref(c);
                        all[i++] = c;
                }
        }

        if ( (r = pthread_mutex_unlock(&cs->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
                while (i) {
                        i--;
                        connection_free(c);
                }
                free(all);
                return NULL;
        }

        return all;

err_unlock:
        if ( (r = pthread_mutex_unlock(&cs->mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
        }
        return NULL;
}


/**
 * Wrap ev timeout event to re-resolve hostname.
 */
static void
connection_resolve_retry_event_(evutil_socket_t fd UNUSED, short what UNUSED, void *arg)
{
        struct connection *c = arg;

        NOTIFY_DEBUG("retrying hostname resolution");

        c->dns_retry_ev = NULL;
        c->dns_retries += 1;
        connection_resolve_hostname(c);
}

/**
 *  A callback used by connection_init to populate the reverse-lookup of a connection's
 *  client_address.
**/
static void
connection_reverse_dns_cb_(int result, char type, int count, int ttl UNUSED, void *addresses, void *ctx)
{
        struct connection *c = ctx;
        char *old_client_address = c->client_address;

        if (result != DNS_ERR_NONE) {
                NOTIFY_ERROR("Error resolving: %s",
                             evdns_err_to_string(result) );
                c->evdns_request = NULL;

                // TODO: configurable
                // if (server_get_config_integer(c->server, "dns_retries_max")) {
                // if (server_get_config_integer(c->server, "dns_retry_seconds")) {

                if (c->dns_retries < 4) {
                        struct timeval _seconds = {30, 0};
                        c->dns_retry_ev = event_new(c->server->base, -1, EV_TIMEOUT, connection_resolve_retry_event_, c);
                        event_add(c->dns_retry_ev, &_seconds);
                };
                return;
        }

        switch (type) {
                case DNS_PTR:
                        if (count < 1) {
                                return;
                        }
                        if (count > 1) {
                                NOTIFY_DEBUG("multiple records returned, using first");
                        }
                        c->client_address = strdup(((char **)addresses)[0]);
                        NOTIFY_DEBUG("resolved [%s] as %s for '%s' (%p)", old_client_address, c->client_address, c->name, c);
                        break;

                default:
                        NOTIFY_DEBUG("reverse lookup of [%s] returned dns type %d", c->client_address, type);
                        break;
        }
        c->evdns_request = NULL;

        if (old_client_address) {
                free(old_client_address);
        }
}


/**
 * Perform reverse lookup of connection address.
 */
void
connection_resolve_hostname(struct connection *c)
{
        if (c->evdns_request) {
                NOTIFY_DEBUG("resolution already in progress");
                return;
        }

        /* does libevent's evdns have a getnameinfo style call? */

        switch (c->sa->sa_family) {
        case AF_INET:
                c->evdns_request = evdns_base_resolve_reverse(c->server->evdns_base, &((struct sockaddr_in *)c->sa)->sin_addr, 0, connection_reverse_dns_cb_, c);
                break;

        case AF_INET6:
                c->evdns_request = evdns_base_resolve_reverse_ipv6(c->server->evdns_base, &((struct sockaddr_in6 *)c->sa)->sin6_addr, 0, connection_reverse_dns_cb_, c);
                break;

        default:
                NOTIFY_DEBUG("unhandled address family %u", c->sa->sa_family);
        }
        if (!c->evdns_request) {
                NOTIFY_DEBUG("could not submit PTR lookup request");
        }
}


/**
 *  Populates a connection with initial information and sets reference count.
 */
int
connection_init(struct server *s, struct connection *c, struct bufferevent *bev, struct sockaddr *sock, int socklen, connection_flags_t flags)
{
        char hbuf[NI_MAXHOST], sbuf[NI_MAXSERV];
        int r;
        uuid_rc_t rc;

        if ( (r = pthread_mutex_init(&c->rc_mutex, NULL)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_init", strerror(r));
                goto err;
        }

        if ( (r = pthread_mutex_init(&c->commands_mutex, NULL)) ) {
                NOTIFY_ERROR("%s:%s", "pthead_mutex_init", strerror(r));
                goto err_rc_mutex_destroy;
        }

        rc = uuid_create(&c->uuid);
        if (rc != UUID_RC_OK) {
                NOTIFY_ERROR("%s:%s", "uuid_create", uuid_error(rc));
                goto err_commands_mutex_destroy;
        }

        rc = uuid_make(c->uuid, UUID_MAKE_V1|UUID_MAKE_MC);
        if (rc != UUID_RC_OK) {
                NOTIFY_ERROR("%s:%s", "uuid_make", uuid_error(rc));
                goto err_uuid_destroy;
        }

        c->state = CONNECTION_STATE_INIT;
        c->bev = bev;
        c->flags = flags;
        c->server = s;

        /* render an IP from a sockaddr */
        r = getnameinfo(sock, socklen, hbuf, sizeof hbuf, sbuf, sizeof sbuf, NI_NUMERICHOST | NI_NUMERICSERV);
        if (r) {
                NOTIFY_ERROR("%s:%s", "getnameinfo", gai_strerror(r));
                strncpy(hbuf, "(unknown)", sizeof hbuf);
                strncpy(sbuf, "(?)", sizeof sbuf);
        }
        c->client_address = strdup(hbuf);
        if (!c->client_address) {
                NOTIFY_ERROR("%s:%s", "strdup", strerror(errno));
                goto err_uuid_destroy;
        }

        /* Past this point, errors are non-fatal. */
        c->reference_count = 1;

        /* Now try to resolve a name from client ip, in background. */
        c->sa = sock;
        c->sa_len = socklen;
        c->dns_retries = 0;
        c->dns_retry_ev = NULL;
        connection_resolve_hostname(c);

        /* FIXME: temporary name of connection for POC */
        static int current_connection_number;
        r = snprintf((char *)c->name, sizeof(c->name), "conn %d", current_connection_number++);
        if ((size_t)r >= sizeof c->name) {
                NOTIFY_ERROR("buffer truncated [%s:%d]", __FILE__, __LINE__);
        }

#if DARWIN
        /* Darwin systems <10.12 lack clock_gettime (!) */
        if (gettimeofday(&c->connect_time, NULL)) {
                NOTIFY_ERROR("%s:%s", "gettimeofday", strerror(errno));
        }
        NOTIFY_DEBUG("c:%p [%s] from %s at %.24s", c, c->name, c->client_address, ctime(&c->connect_time.tv_sec));
#else
        if (clock_gettime(CLOCK_REALTIME, &c->connect_timespec)) {
                NOTIFY_ERROR("%s:%s", "clock_gettime", strerror(errno));
        }
        NOTIFY_DEBUG("c:%p [%s] from %s at %.24s", c, c->name, c->client_address, ctime(&c->connect_timespec.tv_sec));
#endif /* DARWIN */

        return 0;

err_uuid_destroy:
        rc = uuid_destroy(c->uuid);
        if (rc != UUID_RC_OK) {
                NOTIFY_ERROR("%s:%s", "uuid_destroy", uuid_error(rc));
        }
err_commands_mutex_destroy:
        if ( (r = pthread_mutex_destroy(&c->commands_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_destroy", strerror(r));
        }
err_rc_mutex_destroy:
        if ( (r = pthread_mutex_destroy(&c->rc_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_destroy", strerror(r));
        }
err:
        return -1;
}


/**
 * Process a connection's commands as long as there are commands to process.
 * FIXME: track how much time has been spend consuming commands, re-queue
 *        processor job when over some threshold considering pending workqueue
 */
static void
connection_process_queue_(void *data, void *context, size_t id)
{
        struct connection *c = data;
        struct server_worker_context *ctx = context;
        struct command *command;
        struct rusage ru_start, ru_end;
        struct timespec ts_start, ts_current;
        int r;

        if (clock_gettime(CLOCK_MONOTONIC, &ts_start) < 0) {
                NOTIFY_DEBUG("%s:%s", "clock_gettime", strerror(errno));
        }
        while (1) {
                if ( (r = pthread_mutex_lock(&c->commands_mutex)) ) {
                        NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
                        return;
                }

                command = STAILQ_FIRST(&c->commands_head);
                if (command) {
                        STAILQ_REMOVE_HEAD(&c->commands_head, stailq_entry);
                }

                if ( (r = pthread_mutex_unlock(&c->commands_mutex)) ) {
                        NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
                        return;
                }

                if (command == NULL) {
                        break;
                }

                if ( (r = getrusage(RUSAGE_THREAD, &ru_start)) ) {
                        NOTIFY_ERROR("%s:%s", "getrusage", strerror(r));
                }

                command_parse(c, context, id, command);

                if ( (r = getrusage(RUSAGE_THREAD, &ru_end)) ) {
                        NOTIFY_ERROR("%s:%s", "getrusage", strerror(r));
                }

                /* track how much time was spent procesing */
                connection_accounting_increment(c, &ru_start, &ru_end);

                command_free(command);

                if (clock_gettime(CLOCK_MONOTONIC, &ts_current) < 0) {
                        NOTIFY_ERROR("%s:%s", "clock_gettime", strerror(errno));
                }

        }
        connection_free(c);

        // Done processing for a connection, force GC cycle to release additional c references.
        lua_gc(ctx->L, LUA_GCCOLLECT);
}


/**
 *  Adds a command to a connection's list of things to process.
 */
int
connection_command_enqueue(struct connection *c, struct command *command)
{
        bool was_empty;
        int r;

        if ( (r = pthread_mutex_lock(&c->commands_mutex)) ) {
                command_free(command);
                NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
                return -1;
        }

        was_empty = STAILQ_EMPTY(&c->commands_head) ? true : false;
        if (was_empty == true) {
                STAILQ_INSERT_HEAD(&c->commands_head, command, stailq_entry);
        } else {
                STAILQ_INSERT_TAIL(&c->commands_head, command, stailq_entry);
        }

#if DARWIN
        /* Darwin systems <10.12 lack clock_gettime (!) */
        if (gettimeofday(&c->last_received_time, NULL)) {
                NOTIFY_ERROR("%s:%s", "gettimeofday", strerror(errno));
        }
#else
        if (clock_gettime(CLOCK_REALTIME, &c->last_received_timespec)) {
                NOTIFY_ERROR("%s:%s", "clock_gettime", strerror(errno));
        }
#endif /* DARWIN */

        if ( (r = pthread_mutex_unlock(&c->commands_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
                return -1;
        }

        /* if there were already queued commands, a processor should already be scheduled */
        if (was_empty == true) {
                connection_inc_ref(c);
                if (workqueue_add(&c->server->workqueue, connection_process_queue_, c)) {
                        NOTIFY_ERROR("%s:%s", "workqueue_add", "failed");
                        return -1;
                }
        }

        return 0;
}


/**
 *  Add the utime and stime values to a connection's totals.
 */
void
connection_accounting_increment(struct connection *c, struct rusage *ru_start, struct rusage *ru_end)
{
        struct timeval utime_diff, stime_diff;

        timeval_diff(&utime_diff, &ru_end->ru_utime, &ru_start->ru_utime);
        timeval_diff(&stime_diff, &ru_end->ru_stime, &ru_start->ru_stime);

        timeval_increment(&c->utime, &utime_diff);
        timeval_increment(&c->stime, &stime_diff);
}

/**
 *  Locks the output buffer, for sending multiple
 *  contiguous messages without possible interleaving.
 */
void
connection_lock_output(struct connection *c)
{
        struct evbuffer *output = bufferevent_get_output(c->bev);

        evbuffer_lock(output);
}

/**
 *  Unlocks the output buffer.
 */
void
connection_unlock_output(struct connection *c)
{
        struct evbuffer *output = bufferevent_get_output(c->bev);

        evbuffer_unlock(output);
}

/**
 *  Sends a text string to a connection.
 */
int
connection_printf(struct connection *c, const char *fmt, ...)
{
        struct evbuffer *output = bufferevent_get_output(c->bev);
        va_list ap;
        int r;

        va_start(ap, fmt);
        r = evbuffer_add_vprintf(output, fmt, ap);
        va_end(ap);

        return r;
}


/**
 *  Sends a text string to a connection.
 */
int
connection_vprintf(struct connection *c, const char *fmt, va_list ap)
{
        struct evbuffer *output = bufferevent_get_output(c->bev);
        int r;

        r = evbuffer_add_vprintf(output, fmt, ap);

        return r;
}


/**
 *  Send one data buffer to multiple connections.
 *  Data buffer must be malloced, and will be freed when last connection
 *  has finished with it.
 *  Returns number of items sent, or negative on error.
 */
int
connection_multi_send(struct connection **clist, char *data, size_t len)
{
        struct connection **clist_iter;
        struct rc_data_ *rc;
        int r = 0;

        if (!clist) {
                NOTIFY_DEBUG("null clist [%s:%d]", __FILE__, __LINE__);
                return -1;
        }

        /* Track our data so it can be freed once all connections are done with it. */
        rc = rc_data_new_(data, len);
        if (rc == NULL) {
                NOTIFY_ERROR("rc_data_new_('%s', %zu) failed", data, len);
                return -1;
        }

        clist_iter = clist;
        while (*clist_iter) {
                struct evbuffer *output = bufferevent_get_output((*clist_iter)->bev);

                /* FIXME: if connection is still valid... */

                /* FIXME: clang claims this is a use-after-free if evbuffer_add_reference fails, but I don't yet believe it. */
                rc_data_ref_inc_(rc);
                if (evbuffer_add_reference(output, rc->data, rc->data_len, rc_cleanup_cb_, rc)) {
                        NOTIFY_ERROR("%s:%s", "evbuffer_add_reference", "failed");
                        rc_data_free_(rc);
                } else {
                        r++;
                }
                clist_iter++;
        }

        /* FIXME: clang also claims this is use-after-free, same as above. */
        rc_data_free_(rc);

        return r;
}


int
connection_printf_broadcast(struct connection *sender, bool exclude_sender, const char *fmt, ...)
{
        struct connection **clist;
        char *message = NULL;
        ssize_t message_len = 0;
        va_list ap;
        int r;

        va_start(ap, fmt);
        message_len = vsnprintf(message, message_len, fmt, ap);
        va_end(ap);
        if (message_len < 0) {
                NOTIFY_ERROR("%s:%s (%zd)", "vsnprintf", strerror(errno), message_len);
                return -1;
        }
        message_len += 1;
        message = malloc(message_len);
        if (!message) {
                NOTIFY_ERROR("%s(%zd):%s", "malloc", message_len, strerror(errno));
                return -1;
        }
        va_start(ap, fmt);
        message_len = vsnprintf(message, message_len, fmt, ap);
        va_end(ap);
        if (message_len < 0) {
                NOTIFY_ERROR("%s:%s (%zd)", "vsnprintf", strerror(errno), message_len);
                free(message);
                return -1;
        }

#ifdef DEBUG_SILLY
                NOTIFY_DEBUG("message_len:%zu message:%s", message_len, message);
#endif

        clist = connections_all_as_array(&sender->server->connections, exclude_sender == true ? sender : NULL);
        r = connection_multi_send(clist, message, message_len);
        if (r < 0) {
                NOTIFY_ERROR("%s:%s", "connection_multi_send", "failed");
        }
        for (struct connection **clist_iter = clist; *clist_iter; clist_iter++) {
                connection_free(*clist_iter);
        }
        free(clist);

        return r;
}

/**
 *
 */
void
connection_inc_ref(struct connection *c)
{
        int r;

        if ( (r = pthread_mutex_lock(&c->rc_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
                return;
        }

        c->reference_count += 1;

#if DEBUG_SILLY
        NOTIFY_DEBUG("c:%p new reference_count:%zu", c, c->reference_count);
#endif

        if ( (r = pthread_mutex_unlock(&c->rc_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
                return;
        }
}

/**
 *  Frees a connection (and all data owned by a connection) if there are no more references to it.
 */
void
connection_free(struct connection *c)
{
        struct evbuffer *tmp;
        evutil_socket_t fd;
        int r;

        if ( (r = pthread_mutex_lock(&c->rc_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
                return;
        }

        c->reference_count -= 1;

#if DEBUG_SILLY
        NOTIFY_DEBUG("c:%p new reference_count:%zu", c, c->reference_count);
#endif

        if (c->reference_count > 0) {
                if ( (r = pthread_mutex_unlock(&c->rc_mutex)) ) {
                        NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
                }

                return;
        }

        NOTIFY_DEBUG("c:%p freeing", c);

        connections_remove(c);

        bufferevent_disable(c->bev, EV_READ|EV_WRITE);
        tmp = bufferevent_get_output(c->bev);
        evbuffer_drain(tmp, evbuffer_get_length(tmp));
        tmp = bufferevent_get_input(c->bev);
        evbuffer_drain(tmp, evbuffer_get_length(tmp));

        fd = bufferevent_getfd(c->bev);
        if (fd != -1) {
                EVUTIL_CLOSESOCKET(fd);
                bufferevent_setfd(c->bev, -1);
        }

        if (c->dns_retry_ev) {
                if (event_del(c->dns_retry_ev)) {
                        NOTIFY_ERROR("%s:%s", "event_del", strerror(errno));
                }
                c->dns_retry_ev = NULL;
        }

        if (c->evdns_request) {
                evdns_cancel_request(c->server->evdns_base, c->evdns_request);
                c->evdns_request = NULL;
        }

        if (c->client_address) {
                free(c->client_address);
                c->client_address = NULL;
        }

        bufferevent_free(c->bev);

        if (c->uuid) {
                uuid_rc_t rc;

                rc = uuid_destroy(c->uuid);
                if (rc != UUID_RC_OK) {
                        NOTIFY_ERROR("%s:%s", "uuid_destroy", uuid_error(rc));
                }
                c->uuid = NULL;
        }

        /* empty out the command queue */
        if ( (r = pthread_mutex_lock(&c->commands_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_lock", strerror(r));
        }

        struct command *c1, *c2;
        c1 = STAILQ_FIRST(&c->commands_head);
        while (c1 != NULL) {
                c2 = STAILQ_NEXT(c1, stailq_entry);
                free(c1);
                c1 = c2;
        }
        STAILQ_INIT(&c->commands_head);

        if ( (r = pthread_mutex_unlock(&c->commands_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
        }
        if ( (r = pthread_mutex_destroy(&c->commands_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_destroy", strerror(r));
        }

        if ( (r = pthread_mutex_unlock(&c->rc_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_unlock", strerror(r));
        }
        if ( (r = pthread_mutex_destroy(&c->rc_mutex)) ) {
                NOTIFY_ERROR("%s:%s", "pthread_mutex_destroy", strerror(r));
        }

        memset(c, 0, sizeof *c);

        free(c);
}