/*

+----------------------------------------------------------------------+

| Copyright (c) 1997-2009 The PHP Group |

+----------------------------------------------------------------------+

| This source file is subject to version 3.01 of the PHP license, |

| that is bundled with this package in the file LICENSE, and is |

| available through the world-wide-web at the following url: |

| http://www.php.net/license/3_01.txt |

| If you did not receive a copy of the PHP license and are unable to |

| obtain it through the world-wide-web, please send a note to |

| license@php.net so we can mail you a copy immediately. |

+----------------------------------------------------------------------+

| Author: Nicolas Favre-Felix <n.favre-felix@owlient.eu> |

| Maintainer: Michael Grunder <michael.grunder@gmail.com> |

+----------------------------------------------------------------------+

*/

#include "redis_array_impl.h"

#include "php_redis.h"

#include "library.h"

#include "php_variables.h"

#include "SAPI.h"

#include "ext/standard/url.h"

#include "ext/standard/crc32.h"

#include "ext/standard/md5.h"

#include "ext/hash/php_hash.h"

#define PHPREDIS_INDEX_NAME "__phpredis_array_index__"

extern zend_class_entry *redis_ce;

static RedisArray *

ra_load_hosts(RedisArray *ra, HashTable *hosts, zend_string *user,

zend_string *pass, long retry_interval, zend_bool b_lazy_connect)

{

int i = 0, host_len;

char *host, *p;

short port;

zval *zpData;

redis_object *redis;

/* init connections */

ZEND_HASH_FOREACH_VAL(hosts, zpData) {

if (Z_TYPE_P(zpData) != IS_STRING) {

return NULL;

}

/* default values */

host = Z_STRVAL_P(zpData);

host_len = Z_STRLEN_P(zpData);

ra->hosts[i] = zend_string_init(host, host_len, 0);

port = 6379;

if((p = strrchr(host, ':'))) { /* found port */

host_len = p - host;

port = (short)atoi(p+1);

} else if(strchr(host,'/') != NULL) { /* unix socket */

port = -1;

}

/* create Redis object */

object_init_ex(&ra->redis[i], redis_ce);

redis = PHPREDIS_ZVAL_GET_OBJECT(redis_object, &ra->redis[i]);

/* create socket */

redis->sock = redis_sock_create(host, host_len, port, ra->connect_timeout,

ra->read_timeout, ra->pconnect, NULL,

retry_interval);

redis_sock_set_auth(redis->sock, user, pass);

if (!b_lazy_connect) {

if (redis_sock_server_open(redis->sock) < 0) {

ra->count = ++i;

return NULL;

}

}

ra->count = ++i;

} ZEND_HASH_FOREACH_END();

return ra;

}

/* List pure functions */

void

ra_init_function_table(RedisArray *ra)

{

ALLOC_HASHTABLE(ra->pure_cmds);

zend_hash_init(ra->pure_cmds, 0, NULL, NULL, 0);

zend_hash_str_update_ptr(ra->pure_cmds, "EXISTS", sizeof("EXISTS") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "GET", sizeof("GET") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "GETBIT", sizeof("GETBIT") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "GETRANGE", sizeof("GETRANGE") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "HEXISTS", sizeof("HEXISTS") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "HGET", sizeof("HGET") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "HGETALL", sizeof("HGETALL") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "HKEYS", sizeof("HKEYS") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "HLEN", sizeof("HLEN") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "HMGET", sizeof("HMGET") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "HVALS", sizeof("HVALS") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "LINDEX", sizeof("LINDEX") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "LLEN", sizeof("LLEN") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "LRANGE", sizeof("LRANGE") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "OBJECT", sizeof("OBJECT") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "SCARD", sizeof("SCARD") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "SDIFF", sizeof("SDIFF") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "SINTER", sizeof("SINTER") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "SISMEMBER", sizeof("SISMEMBER") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "SMEMBERS", sizeof("SMEMBERS") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "SRANDMEMBER", sizeof("SRANDMEMBER") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "STRLEN", sizeof("STRLEN") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "SUNION", sizeof("SUNION") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "TYPE", sizeof("TYPE") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "ZCARD", sizeof("ZCARD") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "ZCOUNT", sizeof("ZCOUNT") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "ZRANGE", sizeof("ZRANGE") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "ZRANK", sizeof("ZRANK") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "ZREVRANGE", sizeof("ZREVRANGE") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "ZREVRANGEBYSCORE", sizeof("ZREVRANGEBYSCORE") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "ZREVRANK", sizeof("ZREVRANK") - 1, NULL);

zend_hash_str_update_ptr(ra->pure_cmds, "ZSCORE", sizeof("ZSCORE") - 1, NULL);

}

static int

ra_find_name(const char *name) {

const char *ini_names, *p, *next;

/* php_printf("Loading redis array with name=[%s]\n", name); */

ini_names = INI_STR("redis.arrays.names");

for(p = ini_names; p;) {

next = strchr(p, ',');

if(next) {

if(strncmp(p, name, next - p) == 0) {

return 1;

}

} else {

if(strcmp(p, name) == 0) {

return 1;

}

break;

}

p = next + 1;

}

return 0;

}

/* load array from INI settings */

RedisArray *ra_load_array(const char *name) {

zval *z_data, z_tmp, z_fun, z_dist;

zval z_params_hosts;

zval z_params_prev;

RedisArray *ra = NULL;

zend_string *algorithm = NULL, *user = NULL, *pass = NULL;

zend_bool b_index = 0, b_autorehash = 0, b_pconnect = 0, consistent = 0;

zend_long l_retry_interval = 0;

zend_bool b_lazy_connect = 0;

double d_connect_timeout = 0, read_timeout = 0.0;

HashTable *hHosts = NULL, *hPrev = NULL;

size_t name_len = strlen(name);

char *iptr;

/* find entry */

if(!ra_find_name(name))

return ra;

ZVAL_NULL(&z_fun);

ZVAL_NULL(&z_dist);

/* find hosts */

array_init(&z_params_hosts);

if ((iptr = INI_STR("redis.arrays.hosts")) != NULL) {

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_params_hosts);

if ((z_data = zend_hash_str_find(Z_ARRVAL(z_params_hosts), name, name_len)) != NULL) {

hHosts = Z_ARRVAL_P(z_data);

}

}

/* find previous hosts */

array_init(&z_params_prev);

if ((iptr = INI_STR("redis.arrays.previous")) != NULL) {

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_params_prev);

if ((z_data = zend_hash_str_find(Z_ARRVAL(z_params_prev), name, name_len)) != NULL) {

if (Z_TYPE_P(z_data) == IS_ARRAY) {

hPrev = Z_ARRVAL_P(z_data);

}

}

}

/* find function */

if ((iptr = INI_STR("redis.arrays.functions")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_zval(Z_ARRVAL(z_tmp), name, name_len, &z_fun, 1, 0);

zval_dtor(&z_tmp);

}

/* find distributor */

if ((iptr = INI_STR("redis.arrays.distributor")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_zval(Z_ARRVAL(z_tmp), name, name_len, &z_dist, 1, 0);

zval_dtor(&z_tmp);

}

/* find hash algorithm */

if ((iptr = INI_STR("redis.arrays.algorithm")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_string(Z_ARRVAL(z_tmp), name, name_len, &algorithm);

zval_dtor(&z_tmp);

}

/* find index option */

if ((iptr = INI_STR("redis.arrays.index")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_zend_bool(Z_ARRVAL(z_tmp), name, name_len, &b_index);

zval_dtor(&z_tmp);

}

/* find autorehash option */

if ((iptr = INI_STR("redis.arrays.autorehash")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_zend_bool(Z_ARRVAL(z_tmp), name, name_len, &b_autorehash);

zval_dtor(&z_tmp);

}

/* find retry interval option */

if ((iptr = INI_STR("redis.arrays.retryinterval")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_long(Z_ARRVAL(z_tmp), name, name_len, &l_retry_interval);

zval_dtor(&z_tmp);

}

/* find pconnect option */

if ((iptr = INI_STR("redis.arrays.pconnect")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_zend_bool(Z_ARRVAL(z_tmp), name, name_len, &b_pconnect);

zval_dtor(&z_tmp);

}

/* find lazy connect option */

if ((iptr = INI_STR("redis.arrays.lazyconnect")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_zend_bool(Z_ARRVAL(z_tmp), name, name_len, &b_lazy_connect);

zval_dtor(&z_tmp);

}

/* find connect timeout option */

if ((iptr = INI_STR("redis.arrays.connecttimeout")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_double(Z_ARRVAL(z_tmp), name, name_len, &d_connect_timeout);

zval_dtor(&z_tmp);

}

/* find read timeout option */

if ((iptr = INI_STR("redis.arrays.readtimeout")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_double(Z_ARRVAL(z_tmp), name, name_len, &read_timeout);

zval_dtor(&z_tmp);

}

/* find consistent option */

if ((iptr = INI_STR("redis.arrays.consistent")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

if ((z_data = zend_hash_str_find(Z_ARRVAL(z_tmp), name, name_len)) != NULL) {

consistent = Z_TYPE_P(z_data) == IS_STRING && strncmp(Z_STRVAL_P(z_data), "1", 1) == 0;

}

zval_dtor(&z_tmp);

}

/* find auth option */

if ((iptr = INI_STR("redis.arrays.auth")) != NULL) {

array_init(&z_tmp);

sapi_module.treat_data(PARSE_STRING, estrdup(iptr), &z_tmp);

redis_conf_auth(Z_ARRVAL(z_tmp), name, name_len, &user, &pass);

zval_dtor(&z_tmp);

}

/* create RedisArray object */

ra = ra_make_array(hHosts, &z_fun, &z_dist, hPrev, b_index, b_pconnect, l_retry_interval,

b_lazy_connect, d_connect_timeout, read_timeout, consistent, algorithm,

user, pass);

if (ra) {

ra->auto_rehash = b_autorehash;

if(ra->prev) ra->prev->auto_rehash = b_autorehash;

}

if (algorithm) zend_string_release(algorithm);

if (user) zend_string_release(user);

if (pass) zend_string_release(pass);

zval_dtor(&z_params_hosts);

zval_dtor(&z_params_prev);

zval_dtor(&z_dist);

zval_dtor(&z_fun);

return ra;

}

static int

ra_points_cmp(const void *v1, const void *v2)

{

const ContinuumPoint *p1 = v1, *p2 = v2;

return p1->value < p2->value ? - 1 : p1->value > p2->value;

}

static Continuum *

ra_make_continuum(zend_string **hosts, int nb_hosts)

{

int i, j, k, len, idx = 0;

char host[HOST_NAME_MAX];

unsigned char digest[16];

PHP_MD5_CTX ctx;

Continuum *c;

c = ecalloc(1, sizeof(*c));

c->nb_points = nb_hosts * 160; /* 40 hashes, 4 numbers per hash = 160 points per server */

c->points = ecalloc(c->nb_points, sizeof(*c->points));

for (i = 0; i < nb_hosts; ++i) {

for (j = 0; j < 40; ++j) {

len = snprintf(host, sizeof(host), "%.*s-%u", (int)ZSTR_LEN(hosts[i]), ZSTR_VAL(hosts[i]), j);

PHP_MD5Init(&ctx);

PHP_MD5Update(&ctx, host, len);

PHP_MD5Final(digest, &ctx);

for (k = 0; k < 4; ++k) {

c->points[idx].index = i;

c->points[idx++].value = (digest[3 + k * 4] << 24)

| (digest[2 + k * 4] << 16)

| (digest[1 + k * 4] << 8)

| (digest[k * 4]);

}

}

}

qsort(c->points, c->nb_points, sizeof(*c->points), ra_points_cmp);

return c;

}

RedisArray *

ra_make_array(HashTable *hosts, zval *z_fun, zval *z_dist, HashTable *hosts_prev,

zend_bool b_index, zend_bool b_pconnect, long retry_interval,

zend_bool b_lazy_connect, double connect_timeout, double read_timeout,

zend_bool consistent, zend_string *algorithm, zend_string *user,

zend_string *pass)

{

int i, count;

RedisArray *ra;

if (!hosts || (count = zend_hash_num_elements(hosts)) == 0) return NULL;

/* create object */

ra = emalloc(sizeof(RedisArray));

ra->hosts = ecalloc(count, sizeof(*ra->hosts));

ra->redis = ecalloc(count, sizeof(*ra->redis));

ra->count = 0;

ra->z_multi_exec = NULL;

ra->index = b_index;

ra->auto_rehash = 0;

ra->pconnect = b_pconnect;

ra->connect_timeout = connect_timeout;

ra->read_timeout = read_timeout;

ra->continuum = NULL;

ra->algorithm = NULL;

if (ra_load_hosts(ra, hosts, user, pass, retry_interval, b_lazy_connect) == NULL || !ra->count) {

for (i = 0; i < ra->count; ++i) {

zval_dtor(&ra->redis[i]);

zend_string_release(ra->hosts[i]);

}

efree(ra->redis);

efree(ra->hosts);

efree(ra);

return NULL;

}

ra->prev = hosts_prev ? ra_make_array(hosts_prev, z_fun, z_dist, NULL, b_index, b_pconnect, retry_interval, b_lazy_connect, connect_timeout, read_timeout, consistent, algorithm, user, pass) : NULL;

/* init array data structures */

ra_init_function_table(ra);

/* Set hash function and distribtor if provided */

ZVAL_ZVAL(&ra->z_fun, z_fun, 1, 0);

ZVAL_ZVAL(&ra->z_dist, z_dist, 1, 0);

if (algorithm) ra->algorithm = zend_string_copy(algorithm);

/* init continuum */

if (consistent) {

ra->continuum = ra_make_continuum(ra->hosts, ra->count);

}

return ra;

}

/* call userland key extraction function */

zend_string *

ra_call_extractor(RedisArray *ra, const char *key, int key_len)

{

zend_string *out = NULL;

zval z_ret, z_argv;

/* check that we can call the extractor function */

if (!zend_is_callable_ex(&ra->z_fun, NULL, 0, NULL, NULL, NULL)) {

php_error_docref(NULL, E_ERROR, "Could not call extractor function");

return NULL;

}

ZVAL_NULL(&z_ret);

/* call extraction function */

ZVAL_STRINGL(&z_argv, key, key_len);

call_user_function(EG(function_table), NULL, &ra->z_fun, &z_ret, 1, &z_argv);

if (Z_TYPE(z_ret) == IS_STRING) {

out = zval_get_string(&z_ret);

}

zval_dtor(&z_argv);

zval_dtor(&z_ret);

return out;

}

static zend_string *

ra_extract_key(RedisArray *ra, const char *key, int key_len)

{

char *start, *end;

if (Z_TYPE(ra->z_fun) != IS_NULL) {

return ra_call_extractor(ra, key, key_len);

} else if ((start = strchr(key, '{')) == NULL || (end = strchr(start + 1, '}')) == NULL) {

return zend_string_init(key, key_len, 0);

}

/* found substring */

return zend_string_init(start + 1, end - start - 1, 0);

}

/* call userland key distributor function */

int

ra_call_distributor(RedisArray *ra, const char *key, int key_len)

{

int ret;

zval z_ret, z_argv;

/* check that we can call the extractor function */

if (!zend_is_callable_ex(&ra->z_dist, NULL, 0, NULL, NULL, NULL)) {

php_error_docref(NULL, E_ERROR, "Could not call distributor function");

return -1;

}

ZVAL_NULL(&z_ret);

/* call extraction function */

ZVAL_STRINGL(&z_argv, key, key_len);

call_user_function(EG(function_table), NULL, &ra->z_dist, &z_ret, 1, &z_argv);

ret = (Z_TYPE(z_ret) == IS_LONG) ? Z_LVAL(z_ret) : -1;

zval_dtor(&z_argv);

zval_dtor(&z_ret);

return ret;

}

zval *

ra_find_node(RedisArray *ra, const char *key, int key_len, int *out_pos)

{

int pos;

zend_string *out;

/* extract relevant part of the key */

if ((out = ra_extract_key(ra, key, key_len)) == NULL) {

return NULL;

}

if (Z_TYPE(ra->z_dist) == IS_NULL) {

int i;

unsigned long ret = 0xffffffff;

const php_hash_ops *ops;

/* hash */

if (ra->algorithm && (ops = redis_hash_fetch_ops(ra->algorithm))) {

void *ctx = emalloc(ops->context_size);

unsigned char *digest = emalloc(ops->digest_size);

#if PHP_VERSION_ID >= 80100

ops->hash_init(ctx,NULL);

#else

ops->hash_init(ctx);

#endif

ops->hash_update(ctx, (const unsigned char *)ZSTR_VAL(out), ZSTR_LEN(out));

ops->hash_final(digest, ctx);

memcpy(&ret, digest, MIN(sizeof(ret), ops->digest_size));

ret %= 0xffffffff;

efree(digest);

efree(ctx);

} else {

for (i = 0; i < ZSTR_LEN(out); ++i) {

CRC32(ret, ZSTR_VAL(out)[i]);

}

}

/* get position on ring */

if (ra->continuum) {

int left = 0, right = ra->continuum->nb_points;

while (left < right) {

i = (int)((left + right) / 2);

if (ra->continuum->points[i].value < ret) {

left = i + 1;

} else {

right = i;

}

}

if (right == ra->continuum->nb_points) {

right = 0;

}

pos = ra->continuum->points[right].index;

} else {

pos = (int)((ret ^ 0xffffffff) * ra->count / 0xffffffff);

}

} else {

pos = ra_call_distributor(ra, key, key_len);

if (pos < 0 || pos >= ra->count) {

zend_string_release(out);

return NULL;

}

}

zend_string_release(out);

if(out_pos) *out_pos = pos;

return &ra->redis[pos];

}

zval *

ra_find_node_by_name(RedisArray *ra, zend_string *host) {

int i;

for(i = 0; i < ra->count; ++i) {

if (zend_string_equals(host, ra->hosts[i])) {

return &ra->redis[i];

}

}

return NULL;

}

void

ra_index_multi(zval *z_redis, long multi_value) {

zval z_fun_multi, z_ret;

zval z_args[1];

/* run MULTI */

ZVAL_STRINGL(&z_fun_multi, "MULTI", 5);

ZVAL_LONG(&z_args[0], multi_value);

call_user_function(&redis_ce->function_table, z_redis, &z_fun_multi, &z_ret, 1, z_args);

zval_dtor(&z_fun_multi);

zval_dtor(&z_ret);

}

static void

ra_index_change_keys(const char *cmd, zval *z_keys, zval *z_redis) {

int i, argc;

zval z_fun, z_ret, *z_args;

/* alloc */

argc = 1 + zend_hash_num_elements(Z_ARRVAL_P(z_keys));

z_args = ecalloc(argc, sizeof(zval));

/* prepare first parameters */

ZVAL_STRING(&z_fun, cmd);

ZVAL_STRINGL(&z_args[0], PHPREDIS_INDEX_NAME, sizeof(PHPREDIS_INDEX_NAME) - 1);

/* prepare keys */

for(i = 0; i < argc - 1; ++i) {

zval *zv = zend_hash_index_find(Z_ARRVAL_P(z_keys), i);

if (zv == NULL) {

ZVAL_NULL(&z_args[i+1]);

} else {

z_args[i+1] = *zv;

}

}

/* run cmd */

call_user_function(&redis_ce->function_table, z_redis, &z_fun, &z_ret, argc, z_args);

zval_dtor(&z_args[0]);

zval_dtor(&z_fun);

zval_dtor(&z_ret);

efree(z_args); /* free container */

}

void

ra_index_del(zval *z_keys, zval *z_redis) {

ra_index_change_keys("SREM", z_keys, z_redis);

}

void

ra_index_keys(zval *z_pairs, zval *z_redis) {

zval z_keys, *z_val;

zend_string *zkey;

zend_ulong idx;

/* Initialize key array */

array_init_size(&z_keys, zend_hash_num_elements(Z_ARRVAL_P(z_pairs)));

/* Go through input array and add values to the key array */

ZEND_HASH_FOREACH_KEY_VAL(Z_ARRVAL_P(z_pairs), idx, zkey, z_val) {

zval z_new;

PHPREDIS_NOTUSED(z_val);

if (zkey) {

ZVAL_STRINGL(&z_new, ZSTR_VAL(zkey), ZSTR_LEN(zkey));

} else {

ZVAL_LONG(&z_new, idx);

}

zend_hash_next_index_insert(Z_ARRVAL(z_keys), &z_new);

} ZEND_HASH_FOREACH_END();

/* add keys to index */

ra_index_change_keys("SADD", &z_keys, z_redis);

/* cleanup */

zval_dtor(&z_keys);

}

void

ra_index_key(const char *key, int key_len, zval *z_redis) {

zval z_fun_sadd, z_ret, z_args[2];

/* prepare args */

ZVAL_STRINGL(&z_fun_sadd, "SADD", 4);

ZVAL_STRINGL(&z_args[0], PHPREDIS_INDEX_NAME, sizeof(PHPREDIS_INDEX_NAME) - 1);

ZVAL_STRINGL(&z_args[1], key, key_len);

/* run SADD */

call_user_function(&redis_ce->function_table, z_redis, &z_fun_sadd, &z_ret, 2, z_args);

zval_dtor(&z_fun_sadd);

zval_dtor(&z_args[1]);

zval_dtor(&z_args[0]);

zval_dtor(&z_ret);

}

void

ra_index_exec(zval *z_redis, zval *return_value, int keep_all) {

zval z_fun_exec, z_ret, *zp_tmp;

/* run EXEC */

ZVAL_STRINGL(&z_fun_exec, "EXEC", 4);

call_user_function(&redis_ce->function_table, z_redis, &z_fun_exec, &z_ret, 0, NULL);

zval_dtor(&z_fun_exec);

/* extract first element of exec array and put into return_value. */

if(Z_TYPE(z_ret) == IS_ARRAY) {

if(return_value) {

if(keep_all) {

zp_tmp = &z_ret;

RETVAL_ZVAL(zp_tmp, 1, 0);

} else if ((zp_tmp = zend_hash_index_find(Z_ARRVAL(z_ret), 0)) != NULL) {

RETVAL_ZVAL(zp_tmp, 1, 0);

}

}

}

zval_dtor(&z_ret);

/* zval *zptr = &z_ret; */

/* php_var_dump(&zptr, 0); */

}

void

ra_index_discard(zval *z_redis, zval *return_value) {

zval z_fun_discard, z_ret;

/* run DISCARD */

ZVAL_STRINGL(&z_fun_discard, "DISCARD", 7);

call_user_function(&redis_ce->function_table, z_redis, &z_fun_discard, &z_ret, 0, NULL);

zval_dtor(&z_fun_discard);

zval_dtor(&z_ret);

}

void

ra_index_unwatch(zval *z_redis, zval *return_value) {

zval z_fun_unwatch, z_ret;

/* run UNWATCH */

ZVAL_STRINGL(&z_fun_unwatch, "UNWATCH", 7);

call_user_function(&redis_ce->function_table, z_redis, &z_fun_unwatch, &z_ret, 0, NULL);

zval_dtor(&z_fun_unwatch);

zval_dtor(&z_ret);

}

zend_bool

ra_is_write_cmd(RedisArray *ra, const char *cmd, int cmd_len) {

zend_bool ret;

int i;

char *cmd_up = emalloc(1 + cmd_len);

/* convert to uppercase */

for(i = 0; i < cmd_len; ++i)

cmd_up[i] = toupper(cmd[i]);

cmd_up[cmd_len] = 0;

ret = zend_hash_str_exists(ra->pure_cmds, cmd_up, cmd_len);

efree(cmd_up);

return !ret;

}

/* run TYPE to find the type */

static zend_bool

ra_get_key_type(zval *z_redis, const char *key, int key_len, zval *z_from, long *res) {

int i = 0;

zval z_fun, z_ret, z_arg, *z_data;

long success = 1;

/* Pipelined */

ra_index_multi(z_from, PIPELINE);

/* prepare args */

ZVAL_STRINGL(&z_arg, key, key_len);

/* run TYPE */

ZVAL_NULL(&z_ret);

ZVAL_STRINGL(&z_fun, "TYPE", 4);

call_user_function(&redis_ce->function_table, z_redis, &z_fun, &z_ret, 1, &z_arg);

zval_dtor(&z_fun);

zval_dtor(&z_ret);

/* run TYPE */

ZVAL_NULL(&z_ret);

ZVAL_STRINGL(&z_fun, "TTL", 3);

call_user_function(&redis_ce->function_table, z_redis, &z_fun, &z_ret, 1, &z_arg);

zval_dtor(&z_fun);

zval_dtor(&z_ret);

/* Get the result from the pipeline. */

ra_index_exec(z_from, &z_ret, 1);

if (Z_TYPE(z_ret) == IS_ARRAY) {

ZEND_HASH_FOREACH_VAL(Z_ARRVAL(z_ret), z_data) {

if (z_data == NULL || Z_TYPE_P(z_data) != IS_LONG) {

success = 0;

break;

}

/* Get the result - Might change in the future to handle doubles as well */

res[i++] = Z_LVAL_P(z_data);

} ZEND_HASH_FOREACH_END();

}

zval_dtor(&z_arg);

zval_dtor(&z_ret);

return success;

}

/* delete key from source server index during rehashing */

static void

ra_remove_from_index(zval *z_redis, const char *key, int key_len) {

zval z_fun_srem, z_ret, z_args[2];

/* run SREM on source index */

ZVAL_STRINGL(&z_fun_srem, "SREM", 4);

ZVAL_STRINGL(&z_args[0], PHPREDIS_INDEX_NAME, sizeof(PHPREDIS_INDEX_NAME) - 1);

ZVAL_STRINGL(&z_args[1], key, key_len);

call_user_function(&redis_ce->function_table, z_redis, &z_fun_srem, &z_ret, 2, z_args);

/* cleanup */

zval_dtor(&z_fun_srem);

zval_dtor(&z_args[1]);

zval_dtor(&z_args[0]);

zval_dtor(&z_ret);

}

/* delete key from source server during rehashing */

static zend_bool

ra_del_key(const char *key, int key_len, zval *z_from) {

zval z_fun_del, z_ret, z_args[1];

/* in a transaction */

ra_index_multi(z_from, MULTI);

/* run DEL on source */

ZVAL_STRINGL(&z_fun_del, "DEL", 3);

ZVAL_STRINGL(&z_args[0], key, key_len);

call_user_function(&redis_ce->function_table, z_from, &z_fun_del, &z_ret, 1, z_args);

zval_dtor(&z_fun_del);

zval_dtor(&z_args[0]);

zval_dtor(&z_ret);

/* remove key from index */

ra_remove_from_index(z_from, key, key_len);

/* close transaction */

ra_index_exec(z_from, NULL, 0);

return 1;

}

static zend_bool

ra_expire_key(const char *key, int key_len, zval *z_to, long ttl) {

zval z_fun_expire, z_ret, z_args[2];

if (ttl > 0)

{

/* run EXPIRE on target */

ZVAL_STRINGL(&z_fun_expire, "EXPIRE", 6);

ZVAL_STRINGL(&z_args[0], key, key_len);

ZVAL_LONG(&z_args[1], ttl);

call_user_function(&redis_ce->function_table, z_to, &z_fun_expire, &z_ret, 2, z_args);

zval_dtor(&z_fun_expire);

zval_dtor(&z_args[0]);

zval_dtor(&z_ret);

}

return 1;

}

static zend_bool

ra_move_zset(const char *key, int key_len, zval *z_from, zval *z_to, long ttl) {

zval z_fun_zrange, z_fun_zadd, z_ret, z_ret_dest, z_args[4], *z_zadd_args, *z_score_p;

int i, count;

HashTable *h_zset_vals;

zend_string *zkey;

zend_ulong idx;

/* run ZRANGE key 0 -1 WITHSCORES on source */

ZVAL_STRINGL(&z_fun_zrange, "ZRANGE", 6);

ZVAL_STRINGL(&z_args[0], key, key_len);

ZVAL_STRINGL(&z_args[1], "0", 1);

ZVAL_STRINGL(&z_args[2], "-1", 2);

ZVAL_BOOL(&z_args[3], 1);

call_user_function(&redis_ce->function_table, z_from, &z_fun_zrange, &z_ret, 4, z_args);

zval_dtor(&z_fun_zrange);

zval_dtor(&z_args[2]);

zval_dtor(&z_args[1]);

zval_dtor(&z_args[0]);

if(Z_TYPE(z_ret) != IS_ARRAY) { /* key not found or replaced */

/* TODO: report? */

zval_dtor(&z_ret);

return 0;

}

/* we now have an array of value → score pairs in z_ret. */

h_zset_vals = Z_ARRVAL(z_ret);

/* allocate argument array for ZADD */

count = zend_hash_num_elements(h_zset_vals);

z_zadd_args = ecalloc((1 + 2*count), sizeof(zval));

ZVAL_STRINGL(&z_zadd_args[0], key, key_len);

i = 1;

ZEND_HASH_FOREACH_KEY_VAL(h_zset_vals, idx, zkey, z_score_p) {

/* add score */

ZVAL_DOUBLE(&z_zadd_args[i], Z_DVAL_P(z_score_p));

/* add value */

if (zkey) {

ZVAL_STRINGL(&z_zadd_args[i+1], ZSTR_VAL(zkey), ZSTR_LEN(zkey));

} else {

ZVAL_LONG(&z_zadd_args[i+1], (long)idx);

}

i += 2;

} ZEND_HASH_FOREACH_END();

/* run ZADD on target */

ZVAL_STRINGL(&z_fun_zadd, "ZADD", 4);

call_user_function(&redis_ce->function_table, z_to, &z_fun_zadd, &z_ret_dest, 1 + 2 * count, z_zadd_args);

/* Expire if needed */

ra_expire_key(key, key_len, z_to, ttl);

/* cleanup */

zval_dtor(&z_fun_zadd);

zval_dtor(&z_ret_dest);

zval_dtor(&z_ret);

/* Free the array itself */

for (i = 0; i < 1 + 2 * count; i++) {

zval_dtor(&z_zadd_args[i]);

}

efree(z_zadd_args);

return 1;

}

static zend_bool

ra_move_string(const char *key, int key_len, zval *z_from, zval *z_to, long ttl) {

zval z_fun_get, z_fun_set, z_ret, z_args[3];

/* run GET on source */

ZVAL_STRINGL(&z_fun_get, "GET", 3);

ZVAL_STRINGL(&z_args[0], key, key_len);

call_user_function(&redis_ce->function_table, z_from, &z_fun_get, &z_ret, 1, z_args);

zval_dtor(&z_fun_get);

if(Z_TYPE(z_ret) != IS_STRING) { /* key not found or replaced */

/* TODO: report? */

zval_dtor(&z_args[0]);

zval_dtor(&z_ret);

return 0;

}

/* run SET on target */

if (ttl > 0) {

ZVAL_STRINGL(&z_fun_set, "SETEX", 5);

ZVAL_LONG(&z_args[1], ttl);

ZVAL_STRINGL(&z_args[2], Z_STRVAL(z_ret), Z_STRLEN(z_ret)); /* copy z_ret to arg 1 */

zval_dtor(&z_ret); /* free memory from our previous call */

call_user_function(&redis_ce->function_table, z_to, &z_fun_set, &z_ret, 3, z_args);

/* cleanup */

zval_dtor(&z_args[2]);

} else {

ZVAL_STRINGL(&z_fun_set, "SET", 3);

ZVAL_STRINGL(&z_args[1], Z_STRVAL(z_ret), Z_STRLEN(z_ret)); /* copy z_ret to arg 1 */

zval_dtor(&z_ret); /* free memory from our previous return value */

call_user_function(&redis_ce->function_table, z_to, &z_fun_set, &z_ret, 2, z_args);

/* cleanup */

zval_dtor(&z_args[1]);

}

zval_dtor(&z_fun_set);

zval_dtor(&z_args[0]);

zval_dtor(&z_ret);

return 1;

}

static zend_bool

ra_move_hash(const char *key, int key_len, zval *z_from, zval *z_to, long ttl) {

zval z_fun_hgetall, z_fun_hmset, z_ret_dest, z_args[2];

/* run HGETALL on source */

ZVAL_STRINGL(&z_args[0], key, key_len);

ZVAL_STRINGL(&z_fun_hgetall, "HGETALL", 7);

call_user_function(&redis_ce->function_table, z_from, &z_fun_hgetall, &z_args[1], 1, z_args);

zval_dtor(&z_fun_hgetall);

if (Z_TYPE(z_args[1]) != IS_ARRAY) { /* key not found or replaced */

/* TODO: report? */

zval_dtor(&z_args[1]);

zval_dtor(&z_args[0]);

return 0;

}

/* run HMSET on target */

ZVAL_STRINGL(&z_fun_hmset, "HMSET", 5);

call_user_function(&redis_ce->function_table, z_to, &z_fun_hmset, &z_ret_dest, 2, z_args);

zval_dtor(&z_fun_hmset);

zval_dtor(&z_ret_dest);

/* Expire if needed */

ra_expire_key(key, key_len, z_to, ttl);

/* cleanup */

zval_dtor(&z_args[1]);

zval_dtor(&z_args[0]);

return 1;

}

static zend_bool

ra_move_collection(const char *key, int key_len, zval *z_from, zval *z_to,

int list_count, const char **cmd_list,

int add_count, const char **cmd_add, long ttl) {

zval z_fun_retrieve, z_fun_sadd, z_ret, *z_retrieve_args, *z_sadd_args, *z_data_p;

int count, i;

HashTable *h_set_vals;

/* run retrieval command on source */