ks_socket_posix.c

Go to the documentation of this file.

/*
 * libkarmaclient - A C Library to the Karmasphere Reputation Server
 * Copyright (C) 2006 Karmasphere <http://www.karmasphere.com/>
 *  - Shevek <shevek@karmasphere.com>
 *  - Dave Stewart <dave.stewart@karmasphere.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
 */

/*
 * XXX NOTE TO PORTERS AND MAINTAINERS:
 *
 * If you want to do your own socket or I/O management, you do not
 * need either this file or ks_socket.h. You can simply use
 * ks_bquery_packet() and ks_response_bparse() with byte arrays
 * from your own I/O stack.
 *
 * If you do not have POSIX socket functions but still want to use
 * a blocking I/O stack, please contribute a socket implementation
 * for your own system conforming to the interface in ks_socket.h.
 *
 * Greets to Giorgio, Mario and Michael, this is for you.
 */

#include "ks_config.h"
#define RECV_SIZE     65536

#ifdef HAVE_STRING_H
#include <string.h>
#endif

#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif

#ifdef HAVE_SYS_TYPES_H
#include <sys/types.h>
#endif

#ifdef HAVE_SYS_SOCKET_H
#include <sys/socket.h>
#endif

#ifdef HAVE_NETINET_IN_H
#include <netinet/in.h>
#endif

#ifdef HAVE_ARPA_INET_H
#include <arpa/inet.h>
#endif

#ifdef HAVE_NETDB_H
#include <netdb.h>
#endif

/*
#ifdef HAVE_ALLOCA_H
#include <alloca.h>
#endif
*/

/* Required. */
#include <pthread.h>

#ifdef STDC_HEADERS
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#endif

#ifdef HAVE_SYS_TIME_H
# include <sys/time.h>
# ifdef TIME_WITH_SYS_TIME
#  ifdef HAVE_TIME_H
#  include <sys/time.h>
#  endif
# endif
#else
# ifdef HAVE_TIME_H
# include <sys/time.h>
# endif
#endif

#ifdef HAVE_SYS_SELECT_H
#include <sys/select.h>
#endif

#include "ks_socket.h"
#include "ks_malloc.h"

#define MSECS 1000
#define USECS 1000000
#define NSECS 1000000000

struct _ks_socket_t {
        char            *hostname; 
        char            *port;     
        int                      proto;    
        struct sockaddr_in      
                                 addr;     
        int                      socket;   
        pthread_mutex_t
                                 m_send;   
        pthread_mutex_t
                                 m_recv;   
        pthread_mutex_t
                                 m_cache;  
        pthread_cond_t
                                 c_cache;  
        ks_array_t       cache;    
        char            *princ;    
        char            *creds;    
        /* XXX
         * Any results from getaddrinfo,
         * Timeouts for those results.
         */
};

void
ks_socket_set_auth(ks_socket_t *s,
                                const char *principal, const char *credentials)
{
        int      len;

        if (s->princ) {
                ks_free(s->princ);
                s->princ = NULL;
        }

        if (s->creds) {
                ks_free(s->creds);
                s->creds = NULL;
        }
        
        if (principal != NULL) {
                len = strlen(principal);
                s->princ = ks_malloc(len + 1);
                strcpy(s->princ, principal);
        }

        if (credentials != NULL) {
                len = strlen(credentials);
                s->creds = ks_malloc(len + 1);
                strcpy(s->creds, credentials);
        }
}

#define QUEUE_SIZE_MAX  64

static ks_response_t *
ks_socket_cache_get(ks_socket_t *k, ks_string_t *id)
{
        ks_response_t   *r;
        int                              i;

        /* It's nonobvious, but this isn't the place to do this. */
        // pthread_mutex_lock(&(k->m_cache));

        for (i = 0; i < ks_array_length(&(k->cache)); i++) {
                r = (ks_response_t *)ks_array_get(&(k->cache), i);
                if (ks_string_equals(ks_response_id(r), id)) {
                        ks_array_remove(&(k->cache), i);
                        return r;
                }
        }

        // pthread_mutex_unlock(&(k->m_cache));

        return NULL;
}

static void
ks_socket_cache_add(ks_socket_t *k, ks_response_t *r)
{
        /* It's nonobvious, but this isn't the place to do this. */
        // pthread_mutex_lock(&(k->m_cache));

        if (ks_array_length(&(k->cache)) >= QUEUE_SIZE_MAX)
                ks_array_remove(&(k->cache), 0);
        ks_array_add(&(k->cache), (ks_base_t *)r);

        // pthread_mutex_unlock(&(k->m_cache));
}

/* 
 * Cache the [relevant bits of the] answer?
 * Set an expiry time of 3600s for the info.
 * This optional, feel free to do it in connect().
 */
static int
ks_socket_addr(ks_socket_t *k)
{
        struct addrinfo  hh;
        struct addrinfo *ad;
        struct addrinfo *ap;
        int              ss;
        memset(&hh, 0, sizeof(struct addrinfo));
        hh.ai_family = AF_INET;
        if (k->proto == IPPROTO_TCP)
                hh.ai_socktype = SOCK_STREAM;
        else if (k->proto == IPPROTO_UDP)
                hh.ai_socktype = SOCK_DGRAM;
        else
                hh.ai_socktype = 0;
        hh.ai_protocol = k->proto;
        hh.ai_flags = AI_ADDRCONFIG;
        ss = getaddrinfo(k->hostname, k->port, &hh, &ad);
        if (ss != 0) {
                ks_error("getaddrinfo(%s, %s): %s",
                                k->hostname, k->port, gai_strerror(ss));
                return ss;
        }
        for (ap = ad; ap; ap = ap->ai_next) {
                if (ap->ai_family == AF_INET) {
                        memcpy(&(k->addr), ap->ai_addr, ap->ai_addrlen);
                        freeaddrinfo(ad);
                        return 0;
                }
        }
        freeaddrinfo(ad);
        return 1;
}

ks_socket_t *
ks_socket_new(const char *hostname, const char *port, int proto)
{
        ks_socket_t     *k;

        if (hostname == NULL)
                hostname = KS_SLAVE_HOST_DEFAULT;
        if (port == NULL)
                port = KS_SLAVE_PORT_DEFAULT;

        if (proto == IPPROTO_TCP)
                ;
        else if (proto == IPPROTO_UDP)
                ;
        else {
                ks_error("Unsupported protocol %d\n", proto);
                return NULL;
        }

        /* XXX: Unchecked malloc */
        k = (ks_socket_t *)ks_malloc(sizeof(ks_socket_t));
        memset(k, 0, sizeof(ks_socket_t));
        k->hostname = (char *)ks_malloc(strlen(hostname) + 1);
        strcpy(k->hostname, hostname);
        k->port = (char *)ks_malloc(strlen(port) + 1);
        strcpy(k->port, port);
        k->proto = proto;
        k->socket = -1;
        k->princ = NULL;
        k->creds = NULL;

        pthread_mutex_init(&(k->m_send), NULL);
        pthread_mutex_init(&(k->m_recv), NULL);
        pthread_mutex_init(&(k->m_cache), NULL);
        pthread_cond_init(&(k->c_cache), NULL);

        ks_array_init(&(k->cache));
        return k;
}

void
ks_socket_free(ks_socket_t *k)
{
        ks_socket_close(k);
        ks_free(k->hostname);
        ks_free(k->port);

        pthread_mutex_destroy(&(k->m_send));
        pthread_mutex_destroy(&(k->m_recv));
        pthread_mutex_destroy(&(k->m_cache));
        pthread_cond_destroy(&(k->c_cache));

        ks_array_fini(&(k->cache));
        if (k->princ)
                ks_free(k->princ);
        if (k->creds)
                ks_free(k->creds);
        ks_free(k);
}

/* getaddrinfo() in here */
int
ks_socket_connect(ks_socket_t *k)
{
        if (k->socket != -1)
                ks_socket_close(k);

        if (ks_socket_addr(k) != 0)
                return -1;
        if (k->proto == IPPROTO_TCP) {
                k->socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        }
        else if (k->proto == IPPROTO_UDP) {
                k->socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
        }
        else {
                /* NOTREACHED */
                return -1;
        }

        if (k->socket == -1) {
                ks_error("ks_socket_connect: socket: %s", strerror(errno));
                return -1;
        }
        if (connect(k->socket,
                        (struct sockaddr *)&(k->addr),
                        sizeof(struct sockaddr_in)) < 0) {
                ks_error("ks_socket_connect: connect: %s", strerror(errno));
                return -1;
        }

        return 0;
}

void
ks_socket_close(ks_socket_t *k)
{
        if (k->socket != -1)
                close(k->socket);
        k->socket = -1;
}

static int
ks_socket_send_real(ks_socket_t *k, void *buf, int len) {
        int              ret;
        int              off = 0;

        do {
                ret = write(k->socket, buf + off, len - off);
                if (ret < 0) {
                        ks_error("ks_socket_send: write: %s", strerror(errno));
                        ks_socket_close(k);
                        return ret;
                }
                off += ret;
        } while (off < len);

        return off;
}

ssize_t
ks_socket_send(ks_socket_t *k, ks_bquery_t *q)
{
        void            *buf;
        ks_string_t     *r;
        uint32_t         len;
        ssize_t          ret = -1;

        if (k->socket == -1) {
                ret = ks_socket_connect(k);
                if (ret < 0)
                        return ret;
        }

        if (k->princ != NULL) {
                int authed = ks_bquery_has_auth(q);

                if (!authed)
                        ks_bquery_set_auth(q, k->princ, k->creds);
        }

        /* DO NOT RETURN EARLY - you will forget to free this. */
        r = ks_bquery_packet(q);

        pthread_mutex_lock(&(k->m_send));

        if (k->proto == IPPROTO_TCP) {
// #if defined(KS_USE_ALLOCA) && defined(HAVE_ALLOCA)
//              buf = alloca(4 + ks_string_length(r));
// #else
                buf = ks_malloc(4 + ks_string_length(r));
// #endif
                len = htonl(ks_string_length(r));
                memcpy(buf, &len, 4);
                memcpy(buf + 4, ks_string_get(r), ks_string_length(r));
                ret = ks_socket_send_real(k, buf, 4 + ks_string_length(r));
                /* XXX This is the place to try a reconnect. */
// #if !defined(KS_USE_ALLOCA) || !defined(HAVE_ALLOCA)
                ks_free(buf);
// #endif
        }
        else if (k->proto == IPPROTO_UDP) {
                /* Under Linux, the respecification of the target address
                 * is redundant. Under BSD, it returns EISCONN - socket
                 * is already connected, and we must pass NULL, 0. */
                ret = sendto(k->socket,
                                ks_string_get(r), ks_string_length(r),
                                0,
                                /* (struct sockaddr*)&(k->addr),
                                sizeof(struct sockaddr_in) */
                                NULL, 0);
                if (ret < 0) {
                        ks_error("ks_socket_send: sendto: %s", strerror(errno));
                        ks_socket_close(k);
                }
        }

        pthread_mutex_unlock(&(k->m_send));

        ks_string_free(r);

        return ret;
}

static int
ks_socket_tv_subtract(struct timeval *out,
                                const struct timeval *x, struct timeval *y)
{
        /* Perform the carry for the later subtraction by updating y. */
        if (x->tv_usec < y->tv_usec) {
                int nsec = (y->tv_usec - x->tv_usec) / USECS + 1;
                y->tv_usec -= USECS * nsec;
                y->tv_sec += nsec;
        }
        if (x->tv_usec - y->tv_usec > USECS) {
                int nsec = (x->tv_usec - y->tv_usec) / USECS;
                y->tv_usec += USECS * nsec;
                y->tv_sec -= nsec;
        }

        /* Compute the time remaining to wait.
         * tv_usec is certainly positive. */
        out->tv_sec = x->tv_sec - y->tv_sec;
        out->tv_usec = x->tv_usec - y->tv_usec;

        /* Return 1 if result is negative. */
        return x->tv_sec < y->tv_sec;
}

static ks_response_t *
ks_socket_recv_real(ks_socket_t *k, const struct timeval *end)
{
        char                     buffer[RECV_SIZE];
        uint32_t                 len;
        ssize_t                  ret;
        uint32_t                 off;

        struct timeval   now;
        struct timeval   timeout;
        fd_set                   rfds;

        if (k->socket == -1) {
                ks_error("Cannot receive from closed/invalid socket");
                return NULL;
        }

        /* Again, it's nonobvious, but this isn't the place. */
        // pthread_mutex_lock(&(k->m_recv));

        {
                gettimeofday(&now, NULL);
                if (ks_socket_tv_subtract(&timeout, end, &now)) {
                        ks_error("ks_socket_recv: timeout on socket %d", k->socket);
                        return NULL;    /* timeout */
                }
                FD_ZERO(&rfds);
                FD_SET(k->socket, &rfds);
                select(k->socket + 1, &rfds, NULL, NULL, &timeout);
                if (!FD_ISSET(k->socket, &rfds)) {
                        ks_error("ks_socket_recv: timeout on socket %d", k->socket);
                        return NULL;    /* timeout */
                }
        }

        if (k->proto == IPPROTO_TCP) {
                len = 1;
                for (off = 0; off < 4; off += ret) {
                        ret = read(k->socket, &(buffer[off]), 4 - off);
                        if (ret < 0) {
                                ks_error("ks_socket_recv: read: %s", strerror(errno));
                                ks_socket_close(k);
                                len = 0;
                                break;
                        }
                }
                if (len > 0) {
                        len = ntohl(*(uint32_t *)buffer);
                        if (len > RECV_SIZE || len < 2) {
                                ks_error("Cannot receive packet: bad size %u", len);
                                ks_socket_close(k);
                                len = 0;
                        }
                        else {
                                for (off = 0; off < len; off += ret) {
                                        ret = read(k->socket, &(buffer[off]), len - off);
                                        if (ret < 0) {
                                                ks_error("ks_socket_recv: read: %s",
                                                                strerror(errno));
                                                ks_socket_close(k);
                                                len = 0;
                                                break;
                                        }
                                }
                        }
                }
        }
        else if (k->proto == IPPROTO_UDP) {
                ret = recvfrom(k->socket, buffer,
                                RECV_SIZE, 0,
                                NULL, NULL);
                if (ret < 0) {
                        ks_error("ks_socket_recv: recvfrom: %s", strerror(errno));
                        ks_socket_close(k);
                        len = 0;
                }
                else {
                        len = ret;
                }
        }
        else {
                len = 0;
        }

        // pthread_mutex_unlock(&(k->m_recv));

        if (len == 0)
                return NULL;

        return ks_response_bparse(buffer, len);
}

/* The simple version:
ks_response_t *
ks_socket_recv(ks_socket_t *k, ks_string_t *id)
{
        ks_response_t   *r;

        r = ks_socket_cache_get(k, id);
        if (r != NULL)
                return r;

        for (;;) {
                r = ks_socket_recv_real(k);
                if (r == NULL)
                        return NULL;
                if (id == NULL)
                        return r;
                if (ks_string_equals(ks_response_id(r), id))
                        return r;
                else
                        ks_socket_cache_add(k, r);
        }
}
*/

/* XXX: If using a UDP transport and the packet is lost this will not return */
ks_response_t *
ks_socket_recv(ks_socket_t *k, ks_string_t *id, int timeout)
{
        ks_response_t   *r;
        ks_string_t             *rid;

        if (timeout <= 0)
                timeout = KS_TIMEOUT_DEFAULT;

        struct timeval   tv;
        struct timespec  ts;

        {
                gettimeofday(&tv, NULL);
                tv.tv_usec += timeout * (USECS/MSECS);
                tv.tv_sec += (tv.tv_usec / USECS);
                tv.tv_usec %= USECS;

                ts.tv_sec = tv.tv_sec;
                ts.tv_nsec = tv.tv_usec * (NSECS/USECS);
        }

        pthread_mutex_lock(&(k->m_cache));
        for (;;) {
                r = ks_socket_cache_get(k, id);
                if (r != NULL) {
                        pthread_mutex_unlock(&(k->m_cache));
                        return r;
                }
                /* We hold the lock on the cache here.
                 * Therefore, the recv thread cannot add to
                 * the cache until we are waiting on the
                 * condition. */
                if (pthread_mutex_trylock(&(k->m_recv)) != EBUSY) {
                        /* Let other people access the cache while
                         * this thread is blocked in recv() */
                        pthread_mutex_unlock(&(k->m_cache));
                        for (;;) {
                                /* We are the recv thread until we get our
                                 * own response. */
                                r = ks_socket_recv_real(k, &tv);
                                if (r == NULL)
                                        goto recv_done;
                                rid = ks_response_id(r);
                                if (rid == NULL) {
                                        /* This should never happen. But it did. */
                                        ks_error("ks_socket_recv: received packet has no id");
                                        goto recv_done;
                                }
                                if (ks_string_equals(rid, id))
                                        goto recv_done;
                                pthread_mutex_lock(&(k->m_cache));
                                ks_socket_cache_add(k, r);
                                pthread_cond_broadcast(&(k->c_cache));
                                pthread_mutex_unlock(&(k->m_cache));
                        }
                }
                if (pthread_cond_timedwait(&(k->c_cache), &(k->m_cache), &ts)
                                                == ETIMEDOUT)
                        return NULL;
        }

recv_done:
        /* Tell someone else to get their act in gear. */
        pthread_mutex_unlock(&(k->m_recv));
        pthread_mutex_lock(&(k->m_cache));
        pthread_cond_broadcast(&(k->c_cache));
        pthread_mutex_unlock(&(k->m_cache));
        return r;
}

ks_response_t *
ks_socket_ask(ks_socket_t *k, ks_bquery_t *q, int timeout)
                
{
        ks_response_t   *r;
        int                              i;

        if (timeout <= 0)
                timeout = KS_TIMEOUT_DEFAULT;

        /* XXX Consider a retry in TCP mode in case the maximum UDP
         * packet size for some network segment was exceeded. */
        for (i = 0; i < 3; i++) {
                if (ks_socket_send(k, q) < 0)
                        return NULL;
                r = ks_socket_recv(k, ks_bquery_id_get(q), timeout);
                if (r != NULL)
                        return r;
                timeout += timeout;
        }

        return NULL;
}

---