root/lang/perl/Cache-Memcached-LibMemcached/trunk/perl-libmemcached.c @ 5242

/* $Id$
 *
 * Copyright (c) 2008 Daisuke Maki <daisuke@endeworks.jp>
 * All rights reserved
 */

#ifndef __PERL_LIBMEMCACHED_C__
#define __PERL_LIBMEMCACHED_C__
#include "perl-libmemcached.h"

#define ASSERT_CALLBACK(x) \
    if (! SvOK(x) || ! SvROK(x) || SvTYPE(SvRV(x) ) != SVt_PVCV) { \
        croak("provided argument is not a coderef!"); \
    }

#define CONSTSUBi(name, x) \
    { \
        SV *sv = newSViv(x); \
        newCONSTSUB(stash, name, sv); \
        av_push(export, newSVpv(name, PL_na)); \
    }

void
Cache_LibMemcached_bootstrap()
{
    HV *stash;
    AV *export;

    stash = gv_stashpv("Cache::Memcached::LibMemcached::Constants", 1);
    export = get_av("Cache::Memcached::LibMemcached::Constants::EXPORT_OK", 1);

    CONSTSUBi( "F_STORABLE", F_STORABLE );
    CONSTSUBi( "F_COMPRESS", F_COMPRESS );

    /* There has to be an easier way, eh? */
    CONSTSUBi( "MEMCACHED_SUCCESS", MEMCACHED_SUCCESS );
    CONSTSUBi( "MEMCACHED_FAILURE", MEMCACHED_FAILURE );
    CONSTSUBi( "MEMCACHED_HOST_LOOKUP_FAILURE", MEMCACHED_HOST_LOOKUP_FAILURE );
    CONSTSUBi( "MEMCACHED_CONNECTION_BIND_FAILURE", MEMCACHED_CONNECTION_BIND_FAILURE );
    CONSTSUBi( "MEMCACHED_WRITE_FAILURE", MEMCACHED_WRITE_FAILURE );
    CONSTSUBi( "MEMCACHED_READ_FAILURE", MEMCACHED_READ_FAILURE );
    CONSTSUBi( "MEMCACHED_UNKNOWN_READ_FAILURE", MEMCACHED_UNKNOWN_READ_FAILURE );
    CONSTSUBi( "MEMCACHED_PROTOCOL_ERROR", MEMCACHED_PROTOCOL_ERROR );
    CONSTSUBi( "MEMCACHED_CLIENT_ERROR", MEMCACHED_CLIENT_ERROR );
    CONSTSUBi( "MEMCACHED_SERVER_ERROR", MEMCACHED_SERVER_ERROR );
    CONSTSUBi( "MEMCACHED_CONNECTION_SOCKET_CREATE_FAILURE", MEMCACHED_CONNECTION_SOCKET_CREATE_FAILURE );
    CONSTSUBi( "MEMCACHED_DATA_EXISTS", MEMCACHED_DATA_EXISTS );
    CONSTSUBi( "MEMCACHED_DATA_DOES_NOT_EXIST", MEMCACHED_DATA_DOES_NOT_EXIST );
    CONSTSUBi( "MEMCACHED_NOTSTORED", MEMCACHED_NOTSTORED );
    CONSTSUBi( "MEMCACHED_STORED", MEMCACHED_STORED );
    CONSTSUBi( "MEMCACHED_NOTFOUND", MEMCACHED_NOTFOUND );
    CONSTSUBi( "MEMCACHED_MEMORY_ALLOCATION_FAILURE", MEMCACHED_MEMORY_ALLOCATION_FAILURE );
    CONSTSUBi( "MEMCACHED_PARTIAL_READ", MEMCACHED_PARTIAL_READ );
    CONSTSUBi( "MEMCACHED_SOME_ERRORS", MEMCACHED_SOME_ERRORS );
    CONSTSUBi( "MEMCACHED_NO_SERVERS", MEMCACHED_NO_SERVERS );
    CONSTSUBi( "MEMCACHED_END", MEMCACHED_END );
    CONSTSUBi( "MEMCACHED_DELETED", MEMCACHED_DELETED );
    CONSTSUBi( "MEMCACHED_VALUE", MEMCACHED_VALUE );
    CONSTSUBi( "MEMCACHED_STAT", MEMCACHED_STAT );
    CONSTSUBi( "MEMCACHED_ERRNO", MEMCACHED_ERRNO );
    CONSTSUBi( "MEMCACHED_FAIL_UNIX_SOCKET", MEMCACHED_FAIL_UNIX_SOCKET );
    CONSTSUBi( "MEMCACHED_NOT_SUPPORTED", MEMCACHED_NOT_SUPPORTED );
    CONSTSUBi( "MEMCACHED_NO_KEY_PROVIDED", MEMCACHED_NO_KEY_PROVIDED );
    CONSTSUBi( "MEMCACHED_FETCH_NOTFINISHED", MEMCACHED_FETCH_NOTFINISHED );
    CONSTSUBi( "MEMCACHED_TIMEOUT", MEMCACHED_TIMEOUT );
    CONSTSUBi( "MEMCACHED_MAXIMUM_RETURN", MEMCACHED_MAXIMUM_RETURN );

/*
    CONSTSUBi( "MEMCACHED_BEHAVIOR_NO_BLOCK", MEMCACHED_BEHAVIOR_NO_BLOCK );
  MEMCACHED_BEHAVIOR_TCP_NODELAY,
  MEMCACHED_BEHAVIOR_HASH,
  MEMCACHED_BEHAVIOR_KETAMA,
  MEMCACHED_BEHAVIOR_SOCKET_SEND_SIZE,
  MEMCACHED_BEHAVIOR_SOCKET_RECV_SIZE,
  MEMCACHED_BEHAVIOR_CACHE_LOOKUPS,
  MEMCACHED_BEHAVIOR_SUPPORT_CAS,
  MEMCACHED_BEHAVIOR_POLL_TIMEOUT,
  MEMCACHED_BEHAVIOR_DISTRIBUTION,
*/

    CONSTSUBi( "MEMCACHED_DISTRIBUTION_MODULA", MEMCACHED_DISTRIBUTION_MODULA );
    CONSTSUBi( "MEMCACHED_DISTRIBUTION_CONSISTENT", MEMCACHED_DISTRIBUTION_CONSISTENT );

    CONSTSUBi( "MEMCACHED_HASH_DEFAULT", MEMCACHED_HASH_DEFAULT );
    CONSTSUBi( "MEMCACHED_HASH_MD5", MEMCACHED_HASH_MD5 );
    CONSTSUBi( "MEMCACHED_HASH_CRC", MEMCACHED_HASH_CRC );
    CONSTSUBi( "MEMCACHED_HASH_FNV1_64", MEMCACHED_HASH_FNV1_64 );
    CONSTSUBi( "MEMCACHED_HASH_FNV1A_64", MEMCACHED_HASH_FNV1A_64 );
    CONSTSUBi( "MEMCACHED_HASH_FNV1_32", MEMCACHED_HASH_FNV1_32 );
    CONSTSUBi( "MEMCACHED_HASH_FNV1A_32", MEMCACHED_HASH_FNV1A_32 );
    CONSTSUBi( "MEMCACHED_HASH_KETAMA", MEMCACHED_HASH_KETAMA );
    CONSTSUBi( "MEMCACHED_HASH_HSIEH", MEMCACHED_HASH_HSIEH );
}

/* The next functions (compress, uncompress, freeze, thaw) are implemented
 * as simple static method calls emulating a Perl method calls to
 * Compress::Zlib and Storable, respectively
 */

inline static SV *
Cache_LibMemcached_uncompress(cache, raw_value, value_len, flags)
        Cache_LibMemcached *cache;
        char *raw_value;
        size_t value_len;
        uint16_t flags;
{
    SV *sv;

    if (! MEMCACHED_HAVE_ZLIB(cache) || ! (flags & F_COMPRESS)) {
        /* no compression, just create an sv */
        sv = newSVpv(raw_value, value_len);
    } else {
        dSP;

        ENTER;
        SAVETMPS;

        PUSHMARK(SP);
        EXTEND(SP, 1);
        PUSHs(sv_2mortal(newSVpv(raw_value, value_len)));
        PUTBACK;

        if (call_sv(MEMCACHED_UNCOMPRESS_METHOD(cache), G_SCALAR) <= 0) {
            croak("uncompress_method did not return a proper value");
        }
        SPAGAIN;

        sv = POPs;
        SvREFCNT_inc(sv);

        FREETMPS;
        LEAVE;
    }
    return sv;
}

inline static void
Cache_LibMemcached_thaw(cache, sv, flags)
        Cache_LibMemcached *cache;
        SV *sv;
        uint16_t flags;
{
    if (flags & F_STORABLE) {
        SV *rv;
        dSP;
        ENTER;
        SAVETMPS;

        PUSHMARK(SP);
        EXTEND(SP, 1);
        PUSHs(sv);
        PUTBACK;

        if (call_sv(MEMCACHED_DESERIALIZE_METHOD(cache), G_SCALAR) <= 0) {
            croak("deserialize_method did not return a proper value");
        }
        SPAGAIN;

        rv = POPs;
        
        SvSetSV(sv, rv);

        FREETMPS;
        LEAVE;
    }
}

inline static void
Cache_LibMemcached_freeze(cache, sv, flags)
        Cache_LibMemcached *cache;
        SV *sv;
        uint16_t *flags;
{
    if (SvROK(sv)) {
        SV *rv;
        dSP;
        ENTER;
        SAVETMPS;

        PUSHMARK(SP);
        EXTEND(SP, 1);
        PUSHs(sv);
        PUTBACK;

        if (call_sv(MEMCACHED_SERIALIZE_METHOD(cache), G_SCALAR) <= 0) {
            croak("serialize_method did not return a proper value");
        }
        SPAGAIN;

        rv = POPs;
        
        SvSetSV(sv, rv);

        FREETMPS;
        LEAVE;

        *flags |= F_STORABLE;
    }
}

inline static void
Cache_LibMemcached_compress(cache, sv, flags)
        Cache_LibMemcached *cache;
        SV *sv;
        uint16_t *flags;
{
    size_t compress_threshold;
    bool   compress_enabled;
    size_t sv_length = sv_len(sv);

    compress_threshold = MEMCACHED_COMPRESS_THRESHOLD(cache);
    if (MEMCACHED_HAVE_ZLIB(cache) && 
        MEMCACHED_COMPRESS_ENABLED(cache) &&
        compress_threshold > 0 &&
        sv_length > compress_threshold
    ) {
        SV *rv;
        size_t rv_length;
        dSP;

        ENTER;
        SAVETMPS;

        PUSHMARK(SP);
        EXTEND(SP, 1);
        PUSHs(sv);
        PUTBACK;

        if (call_sv(MEMCACHED_COMPRESS_METHOD(cache), G_SCALAR) <= 0) {
            croak("compress_method did not return a proper value");
        }
        SPAGAIN;

        rv = POPs;
        rv_length = SvCUR(rv);

        /* We won't reset the contents of the original SV unless
         * the savings are actually greater than the savings threshold
         */
        if (rv_length < sv_length * ( 1 - MEMCACHED_COMPRESS_SAVINGS(cache) ) ) {
            SvSetSV(sv, rv);
            *flags |= F_COMPRESS;
        }
        FREETMPS;
        LEAVE;

    }
}

SV *
Cache_LibMemcached_create(pkg, have_zlib, compress_enabled, compress_threshold, compress_savings, compress_method, uncompress_method, serialize_method, deserialize_method)
        char *pkg;
        bool have_zlib;
        bool compress_enabled;
        size_t compress_threshold;
        float compress_savings;
        SV *compress_method;
        SV *uncompress_method;
        SV *serialize_method;
        SV *deserialize_method;
{
    SV *sv;
    Cache_LibMemcached *xs;
    memcached_st *cache;

    Newxz(xs, 1, Cache_LibMemcached);

    MEMCACHED_CACHE(xs)              = memcached_create(NULL);
    MEMCACHED_HAVE_ZLIB(xs)          = have_zlib;
    MEMCACHED_COMPRESS_ENABLED(xs)   = compress_enabled;
    MEMCACHED_COMPRESS_THRESHOLD(xs) = compress_threshold;
    MEMCACHED_COMPRESS_SAVINGS(xs)   = compress_savings;
    MEMCACHED_COMPRESS_METHOD(xs)    = compress_method;
    MEMCACHED_UNCOMPRESS_METHOD(xs)  = uncompress_method;
    MEMCACHED_SERIALIZE_METHOD(xs)   = serialize_method;
    MEMCACHED_DESERIALIZE_METHOD(xs) = deserialize_method;

    SvREFCNT_inc(compress_method);
    SvREFCNT_inc(uncompress_method);
    SvREFCNT_inc(serialize_method);
    SvREFCNT_inc(deserialize_method);
    ASSERT_CALLBACK( MEMCACHED_SERIALIZE_METHOD(xs) );

    XS_STRUCT2OBJ(sv, "Cache::Memcached::LibMemcached", xs);

    return sv;
}

void
Cache_LibMemcached_DESTROY(cache)
        Cache_LibMemcached *cache;
{
    memcached_quit(MEMCACHED_CACHE(cache));
    memcached_free(MEMCACHED_CACHE(cache));
    SvREFCNT_dec(MEMCACHED_COMPRESS_METHOD(cache));
    SvREFCNT_dec(MEMCACHED_UNCOMPRESS_METHOD(cache));
    SvREFCNT_dec(MEMCACHED_DESERIALIZE_METHOD(cache));
    SvREFCNT_dec(MEMCACHED_SERIALIZE_METHOD(cache));
}

Cache_LibMemcached_rc
Cache_LibMemcached_server_add(cache, hostname, port)
        Cache_LibMemcached *cache;
        char *hostname;
        unsigned int port;
{
    return memcached_server_add(MEMCACHED_CACHE(cache), hostname, port);
}

Cache_LibMemcached_rc
Cache_LibMemcached_server_add_unix_socket(cache, filename)
        Cache_LibMemcached *cache;
        char *filename;
{
    return memcached_server_add_unix_socket(MEMCACHED_CACHE(cache), filename);
}

unsigned int
Cache_LibMemcached_server_list_count(cache)
        Cache_LibMemcached *cache;
{
    return memcached_server_list_count( MEMCACHED_CACHE(cache)->hosts );
}

void
Cache_LibMemcached_server_list_free(cache)
        Cache_LibMemcached *cache;
{
    memcached_server_list_free( MEMCACHED_CACHE(cache)->hosts );
}

SV *
Cache_LibMemcached_get(cache, key)
        Cache_LibMemcached *cache;
        SV *key;
{
    STRLEN key_len;
    size_t value_len = 0;
    char *key_char, *value;
    Cache_LibMemcached_rc rc;
    uint16_t flags;
    SV *sv;

    key_char = SvPVbyte(key, key_len);
    
    value = memcached_get(MEMCACHED_CACHE(cache), key_char, key_len, &value_len, &flags, &rc);
    if (!value || rc != MEMCACHED_SUCCESS) {
        return &PL_sv_undef;
    }

    sv = Cache_LibMemcached_uncompress( cache, value, value_len, flags );
    Cache_LibMemcached_thaw( cache, sv, flags );
    free(value);

    return sv;
}

Cache_LibMemcached_rc
Cache_LibMemcached_set(cache, key, value, expires)
        Cache_LibMemcached *cache;
        SV *key;
        SV *value;
        time_t expires;
{
    STRLEN key_len, value_len;
    char *key_char;
    char *value_char;
    uint16_t flags = 0;
    SV *value_copy;

    value_copy = newSVsv(value);

    Cache_LibMemcached_freeze(cache, value_copy, &flags);
    Cache_LibMemcached_compress(cache, value_copy, &flags);

    key_char = SvPVbyte(key, key_len);
    value_char = SvPVbyte(value_copy, value_len);

    return memcached_set(MEMCACHED_CACHE(cache), key_char, key_len, value_char, value_len, expires, flags);
}

Cache_LibMemcached_rc
Cache_LibMemcached_cas(cache, key, cas, value, expires)
        Cache_LibMemcached *cache;
        SV *key;
        UV cas;
        SV *value;
        time_t expires;
{
    STRLEN key_len, value_len;
    char *key_char;
    char *value_char;
    uint16_t flags = 0;
    SV *value_copy;

    value_copy = newSVsv(value);

    Cache_LibMemcached_freeze(cache, value_copy, &flags);
    Cache_LibMemcached_compress(cache, value_copy, &flags);

    key_char = SvPVbyte(key, key_len);
    value_char = SvPVbyte(value_copy, value_len);

    return memcached_cas(MEMCACHED_CACHE(cache), key_char, key_len, value_char, value_len, expires, flags, (uint64_t) cas);
}

Cache_LibMemcached_rc
Cache_LibMemcached_add(cache, key, value, expires)
        Cache_LibMemcached *cache;
        SV *key;
        SV *value;
        time_t expires;
{
    STRLEN key_len, value_len;
    char *key_char;
    char *value_char;
    uint16_t flags = 0;
    SV *value_copy;

    value_copy = newSVsv(value);

    Cache_LibMemcached_freeze(cache, value_copy, &flags);
    Cache_LibMemcached_compress(cache, value_copy, &flags);

    key_char = SvPVbyte(key, key_len);
    value_char = SvPVbyte(value_copy, value_len);

    return memcached_add(MEMCACHED_CACHE(cache), key_char, key_len, value_char, value_len, expires, flags);
}

Cache_LibMemcached_rc
Cache_LibMemcached_replace(cache, key, value, expires)
        Cache_LibMemcached *cache;
        SV *key;
        SV *value;
        time_t expires;
{
    STRLEN key_len, value_len;
    char *key_char;
    char *value_char;
    uint16_t flags = 0;
    SV *value_copy;

    value_copy = newSVsv(value);

    Cache_LibMemcached_freeze(cache, value_copy, &flags);
    Cache_LibMemcached_compress(cache, value_copy, &flags);

    key_char = SvPVbyte(key, key_len);
    value_char = SvPVbyte(value_copy, value_len);

    return memcached_replace(MEMCACHED_CACHE(cache), key_char, key_len, value_char, value_len, expires, flags);
}

/* XXX - Notes from memcached spec.
 * 
 * "append" means "add this data to an existing key after existing data".
 *
 * "prepend" means "add this data to an existing key before existing data".
 *
 * The append and prepend commands do not accept flags or exptime.
 * They update existing data portions, and ignore new flag and exptime
 * settings.
 */
Cache_LibMemcached_rc
Cache_LibMemcached_prepend(cache, key_sv, value_sv)
        Cache_LibMemcached *cache;
        SV *key_sv;
        SV *value_sv;
{
    char *key, *value;
    STRLEN key_length, value_length;

    key = SvPV(key_sv, key_length);
    value = SvPV(value_sv, value_length);

    /* While libmemcached's API supports expiration and flags, the 
     * operation as defined in the spec doesn't, so we're sending 
     * 0 for these
     */
    return memcached_prepend(
        MEMCACHED_CACHE(cache), key, key_length, value, value_length,
        (time_t) 0, (uint16_t) 0);
}

Cache_LibMemcached_rc
Cache_LibMemcached_append(cache, key_sv, value_sv)
        Cache_LibMemcached *cache;
        SV *key_sv;
        SV *value_sv;
{
    char *key, *value;
    STRLEN key_length, value_length;

    key = SvPV(key_sv, key_length);
    value = SvPV(value_sv, value_length);

    /* While libmemcached's API supports expiration and flags, the 
     * operation as defined in the spec doesn't, so we're sending 
     * 0 for these
     */
    return memcached_append(
        MEMCACHED_CACHE(cache), key, key_length, value, value_length,
        (time_t) 0, (uint16_t) 0);
}

SV *
Cache_LibMemcached_get_cas_multi(cache, keys, key_len_list, keys_len)
        Cache_LibMemcached *cache;
        char **keys;
        size_t *key_len_list;
        unsigned int keys_len;
{
    Cache_LibMemcached_rc rc;
    memcached_result_st *result;
    HV *hv;
    unsigned int i;

    rc = memcached_mget(MEMCACHED_CACHE(cache), keys, key_len_list, keys_len);
    if (rc != MEMCACHED_SUCCESS) {
        memcached_quit(MEMCACHED_CACHE(cache));
        croak("memcached_mget failed :(");
    }

    hv = newHV();
    result = memcached_result_create(MEMCACHED_CACHE(cache), NULL);
    for(i = 0; i < keys_len; i++) {
        SV *key, *value;

        memcached_fetch_result(MEMCACHED_CACHE(cache), result, &rc);
        if (rc != MEMCACHED_SUCCESS) {
            break;
        }

        key = newSVpv( 
            memcached_result_key_value(result),
            memcached_result_key_length(result)
        );
        value = newSVnv( memcached_result_cas(result) );
        SvREFCNT_inc(value);
        if (hv_store_ent(hv, key, value, 0) == NULL) {
            croak("Failed to insert into hash");
        }
    }
    memcached_result_free(result);
    return newRV_noinc((SV *) hv);
}

SV *
Cache_LibMemcached_mget(cache, keys, key_len_list, keys_len)
        Cache_LibMemcached *cache;
        char **keys;
        size_t *key_len_list;
        unsigned int keys_len;
{
    Cache_LibMemcached_rc rc;
    HV *hv;
    unsigned int i;

    rc = memcached_mget(MEMCACHED_CACHE(cache), keys, key_len_list, keys_len);
    if (rc != MEMCACHED_SUCCESS) {
        memcached_quit(MEMCACHED_CACHE(cache));
        croak("memcached_mget failed :(");
    }

    hv = newHV();
    for(i = 0; i < keys_len; i++) {
        char *value = NULL;
        char key[MEMCACHED_MAX_KEY];
        size_t key_length = 0;
        size_t value_length = 0;
        uint16_t flags = 0;
        Cache_LibMemcached_rc rc;
        SV *sv;

        value = memcached_fetch(MEMCACHED_CACHE(cache), key, &key_length, &value_length, &flags, &rc);
        if (value == NULL) {
            break;
        }

        sv = Cache_LibMemcached_uncompress( cache, value, value_length, flags );
        Cache_LibMemcached_thaw( cache, sv, flags );
        free(value);

        SvREFCNT_inc(sv);
        if (hv_store_ent(hv, newSVpv(key, key_length), sv, 0) == NULL) {
            croak("Failed to insert into hash");
        }
    }

    return newRV_noinc((SV *) hv);
}

SV *
Cache_LibMemcached_delete(cache, key_sv, expiration)
        Cache_LibMemcached *cache;
        SV *key_sv;
        time_t expiration;
{
    char *key;
    size_t key_len;
    Cache_LibMemcached_rc rc;

    key = SvPVbyte(key_sv, key_len);

    rc = memcached_delete(MEMCACHED_CACHE(cache), key, key_len, expiration);
    return (rc == MEMCACHED_SUCCESS) ?
        &PL_sv_yes : &PL_sv_no
    ;
}

SV *
Cache_LibMemcached_incr(cache, key_sv, offset)
        Cache_LibMemcached *cache;
        SV *key_sv;
        unsigned int offset;
{
    char *key;
    size_t key_len;
    uint64_t value;
    Cache_LibMemcached_rc rc;

    key = SvPVbyte(key_sv, key_len);

    rc = memcached_increment(MEMCACHED_CACHE(cache), key, key_len, offset, &value);

    if (rc != MEMCACHED_SUCCESS) {
        return &PL_sv_undef;
    }

    return newSViv(value);
}

SV *
Cache_LibMemcached_decr(cache, key_sv, offset)
        Cache_LibMemcached *cache;
        SV *key_sv;
        unsigned int offset;
{
    char *key;
    size_t key_len;
    uint64_t value;
    Cache_LibMemcached_rc rc;

    key = SvPVbyte(key_sv, key_len);

    rc = memcached_decrement(MEMCACHED_CACHE(cache), key, key_len, offset, &value);

    if (rc != MEMCACHED_SUCCESS) {
        return &PL_sv_undef;
    }

    return newSViv(value);
}

Cache_LibMemcached_rc
Cache_LibMemcached_flush_all(cache, expiration) 
        Cache_LibMemcached *cache;
        time_t expiration;
{
    return memcached_flush( MEMCACHED_CACHE(cache), expiration );
}

void
Cache_LibMemcached_set_support_cas(cache, value)
        Cache_LibMemcached *cache;
        IV value;
{
    memcached_behavior_set( MEMCACHED_CACHE(cache), 
        MEMCACHED_BEHAVIOR_SUPPORT_CAS, (void *) value );
}

void
Cache_LibMemcached_set_no_block(cache, value)
        Cache_LibMemcached *cache;
        IV value;
{
    memcached_behavior_set( MEMCACHED_CACHE(cache), 
        MEMCACHED_BEHAVIOR_NO_BLOCK, (void *) value );
}

IV
Cache_LibMemcached_is_no_block(cache)
        Cache_LibMemcached *cache;
{
    return memcached_behavior_get( MEMCACHED_CACHE(cache), 
        MEMCACHED_BEHAVIOR_NO_BLOCK );
}

void
Cache_LibMemcached_set_distribution_method(cache, value)
        Cache_LibMemcached *cache;
        IV value;
{
    memcached_behavior_set( MEMCACHED_CACHE(cache), 
        MEMCACHED_BEHAVIOR_DISTRIBUTION, &value );
}

IV
Cache_LibMemcached_get_distribution_method(cache)
        Cache_LibMemcached *cache;
{
    return memcached_behavior_get( MEMCACHED_CACHE(cache), 
        MEMCACHED_BEHAVIOR_DISTRIBUTION );
}

void
Cache_LibMemcached_set_hashing_algorithm(cache, value)
        Cache_LibMemcached *cache;
        IV value;
{
    memcached_behavior_set( MEMCACHED_CACHE(cache), 
        MEMCACHED_BEHAVIOR_HASH, &value );
}

IV
Cache_LibMemcached_get_hashing_algorithm(cache)
        Cache_LibMemcached *cache;
{
    return memcached_behavior_get( MEMCACHED_CACHE(cache), 
        MEMCACHED_BEHAVIOR_HASH );
}

Cache_LibMemcached_rc
Cache_LibMemcached_populate_stathv(cache, hv, stat)
        Cache_LibMemcached *cache;
        HV *hv;
        Cache_LibMemcached_stat stat;
{
    Cache_LibMemcached_rc rc;
    char **list;
    char **ptr;

    list = memcached_stat_get_keys(MEMCACHED_CACHE(cache), &stat, &rc);
    if (rc != MEMCACHED_SUCCESS)
        return rc;

    for(ptr = list; *ptr; ptr++) {
        SV *key, *sv;
        char *value;

        value = memcached_stat_get_value(
            MEMCACHED_CACHE(cache),
            &stat,
            *ptr,
            &rc
        );
        if (rc != MEMCACHED_SUCCESS) {
            free(value);
            break;
        }
        key = newSVpvf("%s", *ptr);
        sv  = newSVpvf("%s", value);
        if (hv_store_ent(hv, key, sv, 0) == NULL) {
            /* croak("Failed to insert into hash"); */
            free(value);
            break;
        }
        free(value);
    }
    free(list);

    return rc;
}
        

/* This is so half-baked, it's not even funny */
SV *
Cache_LibMemcached_stats(cache)
        Cache_LibMemcached *cache;
{
    Cache_LibMemcached_rc   rc;
    Cache_LibMemcached_stat *stat;
    memcached_server_st *server_list;
    unsigned int i;
    unsigned int server_count;
    HV *hv, *hosts_hv, *single_host_hv;

    stat = memcached_stat(MEMCACHED_CACHE(cache), NULL, &rc);
    if (rc != MEMCACHED_SUCCESS)
        croak("memcached_stat failed: %s", memcached_strerror(MEMCACHED_CACHE(cache), rc));

    /* need a structure like
     *   { hosts => {
     *      "server:port" => {
     *          key => value
     *      }
     *    }
     */

    hosts_hv = newHV();
    server_list  = memcached_server_list(MEMCACHED_CACHE(cache));
    server_count = memcached_server_count(MEMCACHED_CACHE(cache));
    for (i = 0; i < server_count; i++) {
        SV *key;
        single_host_hv = newHV();

        rc = Cache_LibMemcached_populate_stathv(cache, single_host_hv, stat[i]);
        if (rc != MEMCACHED_SUCCESS)
            break;

        key = newSVpvf("%s:%u", 
            memcached_server_name(cache, server_list[i]),
            memcached_server_port(cache, server_list[i])
        );
        if (hv_store_ent(hosts_hv, key, newRV_inc((SV *) single_host_hv), 0) == NULL) {
            /* croak("Failed to insert into hash"); */
            break;
        }
    }
    free(stat);
    
    hv = newHV(); /* top level hash */
    if (hv_store_ent(hv, newSVpv("hosts", 5), newRV_inc((SV *) hosts_hv), 0) == NULL) {
        croak("Failed insert into hash");
    }
    return newRV_inc((SV *) hv);
}

void
Cache_LibMemcached_disconnect_all(cache)
        Cache_LibMemcached *cache;
{
    memcached_quit(MEMCACHED_CACHE(cache));
}

#endif /* __PERL_LIBMEMCACHED_C__ */