php-win-ext/phpredis
1
0
Fork 0
mirror of https://github.com/php-win-ext/phpredis.git synced 2026-03-24 17:12:15 +01:00
Code Issues Packages Projects Releases Wiki Activity
Files
1b624141c052d89d4b0bd17815768dcb50520edb
phpredis/redis_array_impl.c
michael-grunder 3e6d5b60d6 Cutting a release for 2.2.3. Added myself as a maintainer 
in the header files
2013-04-28 11:09:58 -07:00

1211 lines
33 KiB
C
Raw Blame History

/*
+----------------------------------------------------------------------+
| PHP Version 5 |
+----------------------------------------------------------------------+
| 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"
#define PHPREDIS_INDEX_NAME "__phpredis_array_index__"
extern int le_redis_sock;
extern zend_class_entry *redis_ce;
RedisArray*
ra_load_hosts(RedisArray *ra, HashTable *hosts, long retry_interval TSRMLS_DC)
{
int i, host_len, id;
int count = zend_hash_num_elements(hosts);
char *host, *p;
short port;
zval **zpData, z_cons, z_ret;
RedisSock *redis_sock = NULL;
/* function calls on the Redis object */
ZVAL_STRING(&z_cons, "__construct", 0);
/* init connections */
for(i = 0; i < count; ++i) {
if(FAILURE == zend_hash_quick_find(hosts, NULL, 0, i, (void**)&zpData)) {
efree(ra);
return NULL;
}
ra->hosts[i] = estrdup(Z_STRVAL_PP(zpData));
/* default values */
host = Z_STRVAL_PP(zpData);
host_len = Z_STRLEN_PP(zpData);
port = 6379;
if((p = strchr(host, ':'))) { /* found port */
host_len = p - host;
port = (short)atoi(p+1);
}
/* create Redis object */
MAKE_STD_ZVAL(ra->redis[i]);
object_init_ex(ra->redis[i], redis_ce);
INIT_PZVAL(ra->redis[i]);
call_user_function(&redis_ce->function_table, &ra->redis[i], &z_cons, &z_ret, 0, NULL TSRMLS_CC);
/* create socket */
redis_sock = redis_sock_create(host, host_len, port, 0, 0, NULL, retry_interval); /* TODO: persistence? */
/* connect */
redis_sock_server_open(redis_sock, 1 TSRMLS_CC);
/* attach */
#if PHP_VERSION_ID >= 50400
id = zend_list_insert(redis_sock, le_redis_sock TSRMLS_CC);
#else
id = zend_list_insert(redis_sock, le_redis_sock);
#endif
add_property_resource(ra->redis[i], "socket", id);
}
return ra;
}
/* List pure functions */
void ra_init_function_table(RedisArray *ra) {
MAKE_STD_ZVAL(ra->z_pure_cmds);
array_init(ra->z_pure_cmds);
add_assoc_bool(ra->z_pure_cmds, "HGET", 1);
add_assoc_bool(ra->z_pure_cmds, "HGETALL", 1);
add_assoc_bool(ra->z_pure_cmds, "HKEYS", 1);
add_assoc_bool(ra->z_pure_cmds, "HLEN", 1);
add_assoc_bool(ra->z_pure_cmds, "SRANDMEMBER", 1);
add_assoc_bool(ra->z_pure_cmds, "HMGET", 1);
add_assoc_bool(ra->z_pure_cmds, "STRLEN", 1);
add_assoc_bool(ra->z_pure_cmds, "SUNION", 1);
add_assoc_bool(ra->z_pure_cmds, "HVALS", 1);
add_assoc_bool(ra->z_pure_cmds, "TYPE", 1);
add_assoc_bool(ra->z_pure_cmds, "EXISTS", 1);
add_assoc_bool(ra->z_pure_cmds, "LINDEX", 1);
add_assoc_bool(ra->z_pure_cmds, "SCARD", 1);
add_assoc_bool(ra->z_pure_cmds, "LLEN", 1);
add_assoc_bool(ra->z_pure_cmds, "SDIFF", 1);
add_assoc_bool(ra->z_pure_cmds, "ZCARD", 1);
add_assoc_bool(ra->z_pure_cmds, "ZCOUNT", 1);
add_assoc_bool(ra->z_pure_cmds, "LRANGE", 1);
add_assoc_bool(ra->z_pure_cmds, "ZRANGE", 1);
add_assoc_bool(ra->z_pure_cmds, "ZRANK", 1);
add_assoc_bool(ra->z_pure_cmds, "GET", 1);
add_assoc_bool(ra->z_pure_cmds, "GETBIT", 1);
add_assoc_bool(ra->z_pure_cmds, "SINTER", 1);
add_assoc_bool(ra->z_pure_cmds, "GETRANGE", 1);
add_assoc_bool(ra->z_pure_cmds, "ZREVRANGE", 1);
add_assoc_bool(ra->z_pure_cmds, "SISMEMBER", 1);
add_assoc_bool(ra->z_pure_cmds, "ZREVRANGEBYSCORE", 1);
add_assoc_bool(ra->z_pure_cmds, "ZREVRANK", 1);
add_assoc_bool(ra->z_pure_cmds, "HEXISTS", 1);
add_assoc_bool(ra->z_pure_cmds, "ZSCORE", 1);
add_assoc_bool(ra->z_pure_cmds, "HGET", 1);
add_assoc_bool(ra->z_pure_cmds, "OBJECT", 1);
add_assoc_bool(ra->z_pure_cmds, "SMEMBERS", 1);
}
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;
}
/* laod array from INI settings */
RedisArray *ra_load_array(const char *name TSRMLS_DC) {
zval *z_params_hosts, **z_hosts;
zval *z_params_prev, **z_prev;
zval *z_params_funs, **z_data_pp, *z_fun = NULL, *z_dist = NULL;
zval *z_params_index;
zval *z_params_autorehash;
zval *z_params_retry_interval;
RedisArray *ra = NULL;
zend_bool b_index = 0, b_autorehash = 0;
long l_retry_interval = 0;
HashTable *hHosts = NULL, *hPrev = NULL;
/* find entry */
if(!ra_find_name(name))
return ra;
/* find hosts */
MAKE_STD_ZVAL(z_params_hosts);
array_init(z_params_hosts);
sapi_module.treat_data(PARSE_STRING, estrdup(INI_STR("redis.arrays.hosts")), z_params_hosts TSRMLS_CC);
if (zend_hash_find(Z_ARRVAL_P(z_params_hosts), name, strlen(name) + 1, (void **) &z_hosts) != FAILURE) {
hHosts = Z_ARRVAL_PP(z_hosts);
}
/* find previous hosts */
MAKE_STD_ZVAL(z_params_prev);
array_init(z_params_prev);
sapi_module.treat_data(PARSE_STRING, estrdup(INI_STR("redis.arrays.previous")), z_params_prev TSRMLS_CC);
if (zend_hash_find(Z_ARRVAL_P(z_params_prev), name, strlen(name) + 1, (void **) &z_prev) != FAILURE) {
hPrev = Z_ARRVAL_PP(z_prev);
}
/* find function */
MAKE_STD_ZVAL(z_params_funs);
array_init(z_params_funs);
sapi_module.treat_data(PARSE_STRING, estrdup(INI_STR("redis.arrays.functions")), z_params_funs TSRMLS_CC);
if (zend_hash_find(Z_ARRVAL_P(z_params_funs), name, strlen(name) + 1, (void **) &z_data_pp) != FAILURE) {
MAKE_STD_ZVAL(z_fun);
*z_fun = **z_data_pp;
zval_copy_ctor(z_fun);
}
/* find distributor */
MAKE_STD_ZVAL(z_params_funs);
array_init(z_params_funs);
sapi_module.treat_data(PARSE_STRING, estrdup(INI_STR("redis.arrays.distributor")), z_params_funs TSRMLS_CC);
if (zend_hash_find(Z_ARRVAL_P(z_params_funs), name, strlen(name) + 1, (void **) &z_data_pp) != FAILURE) {
MAKE_STD_ZVAL(z_dist);
*z_dist = **z_data_pp;
zval_copy_ctor(z_dist);
}
/* find index option */
MAKE_STD_ZVAL(z_params_index);
array_init(z_params_index);
sapi_module.treat_data(PARSE_STRING, estrdup(INI_STR("redis.arrays.index")), z_params_index TSRMLS_CC);
if (zend_hash_find(Z_ARRVAL_P(z_params_index), name, strlen(name) + 1, (void **) &z_data_pp) != FAILURE) {
if(Z_TYPE_PP(z_data_pp) == IS_STRING && strncmp(Z_STRVAL_PP(z_data_pp), "1", 1) == 0) {
b_index = 1;
}
}
/* find autorehash option */
MAKE_STD_ZVAL(z_params_autorehash);
array_init(z_params_autorehash);
sapi_module.treat_data(PARSE_STRING, estrdup(INI_STR("redis.arrays.autorehash")), z_params_autorehash TSRMLS_CC);
if (zend_hash_find(Z_ARRVAL_P(z_params_autorehash), name, strlen(name) + 1, (void **) &z_data_pp) != FAILURE) {
if(Z_TYPE_PP(z_data_pp) == IS_STRING && strncmp(Z_STRVAL_PP(z_data_pp), "1", 1) == 0) {
b_autorehash = 1;
}
}
/* find retry interval option */
MAKE_STD_ZVAL(z_params_retry_interval);
array_init(z_params_retry_interval);
sapi_module.treat_data(PARSE_STRING, estrdup(INI_STR("redis.arrays.retryinterval")), z_params_retry_interval TSRMLS_CC);
if (zend_hash_find(Z_ARRVAL_P(z_params_retry_interval), name, strlen(name) + 1, (void **) &z_data_pp) != FAILURE) {
if (Z_TYPE_PP(z_data_pp) == IS_LONG || Z_TYPE_PP(z_data_pp) == IS_STRING) {
if (Z_TYPE_PP(z_data_pp) == IS_LONG) {
l_retry_interval = Z_LVAL_PP(z_data_pp);
}
else {
l_retry_interval = atol(Z_STRVAL_PP(z_data_pp));
}
}
}
/* create RedisArray object */
ra = ra_make_array(hHosts, z_fun, z_dist, hPrev, b_index, l_retry_interval TSRMLS_CC);
ra->auto_rehash = b_autorehash;
/* cleanup */
zval_dtor(z_params_hosts);
efree(z_params_hosts);
zval_dtor(z_params_prev);
efree(z_params_prev);
zval_dtor(z_params_funs);
efree(z_params_funs);
zval_dtor(z_params_index);
efree(z_params_index);
zval_dtor(z_params_autorehash);
efree(z_params_autorehash);
zval_dtor(z_params_retry_interval);
efree(z_params_retry_interval);
return ra;
}
RedisArray *
ra_make_array(HashTable *hosts, zval *z_fun, zval *z_dist, HashTable *hosts_prev, zend_bool b_index, long retry_interval TSRMLS_DC) {
int count = zend_hash_num_elements(hosts);
/* create object */
RedisArray *ra = emalloc(sizeof(RedisArray));
ra->hosts = emalloc(count * sizeof(char*));
ra->redis = emalloc(count * sizeof(zval*));
ra->count = count;
ra->z_fun = NULL;
ra->z_dist = NULL;
ra->z_multi_exec = NULL;
ra->index = b_index;
ra->auto_rehash = 0;
/* init array data structures */
ra_init_function_table(ra);
if(NULL == ra_load_hosts(ra, hosts, retry_interval TSRMLS_CC)) {
return NULL;
}
ra->prev = hosts_prev ? ra_make_array(hosts_prev, z_fun, z_dist, NULL, b_index, retry_interval TSRMLS_CC) : NULL;
/* copy function if provided */
if(z_fun) {
MAKE_STD_ZVAL(ra->z_fun);
*ra->z_fun = *z_fun;
zval_copy_ctor(ra->z_fun);
}
/* copy distributor if provided */
if(z_dist) {
MAKE_STD_ZVAL(ra->z_dist);
*ra->z_dist = *z_dist;
zval_copy_ctor(ra->z_dist);
}
return ra;
}
/* call userland key extraction function */
char *
ra_call_extractor(RedisArray *ra, const char *key, int key_len, int *out_len TSRMLS_DC) {
char *out;
zval z_ret;
zval *z_argv0;
/* check that we can call the extractor function */
if(!zend_is_callable_ex(ra->z_fun, NULL, 0, NULL, NULL, NULL, NULL TSRMLS_CC)) {
php_error_docref(NULL TSRMLS_CC, E_ERROR, "Could not call extractor function");
return NULL;
}
//convert_to_string(ra->z_fun);
/* call extraction function */
MAKE_STD_ZVAL(z_argv0);
ZVAL_STRINGL(z_argv0, key, key_len, 0);
call_user_function(EG(function_table), NULL, ra->z_fun, &z_ret, 1, &z_argv0 TSRMLS_CC);
efree(z_argv0);
if(Z_TYPE(z_ret) != IS_STRING) {
zval_dtor(&z_ret);
return NULL;
}
*out_len = Z_STRLEN(z_ret);
out = emalloc(*out_len + 1);
out[*out_len] = 0;
memcpy(out, Z_STRVAL(z_ret), *out_len);
zval_dtor(&z_ret);
return out;
}
static char *
ra_extract_key(RedisArray *ra, const char *key, int key_len, int *out_len TSRMLS_DC) {
char *start, *end, *out;
*out_len = key_len;
if(ra->z_fun)
return ra_call_extractor(ra, key, key_len, out_len TSRMLS_CC);
/* look for '{' */
start = strchr(key, '{');
if(!start) return estrndup(key, key_len);
/* look for '}' */
end = strchr(start + 1, '}');
if(!end) return estrndup(key, key_len);
/* found substring */
*out_len = end - start - 1;
out = emalloc(*out_len + 1);
out[*out_len] = 0;
memcpy(out, start+1, *out_len);
return out;
}
/* call userland key distributor function */
zend_bool
ra_call_distributor(RedisArray *ra, const char *key, int key_len, int *pos TSRMLS_DC) {
zval z_ret;
zval *z_argv0;
/* check that we can call the extractor function */
if(!zend_is_callable_ex(ra->z_dist, NULL, 0, NULL, NULL, NULL, NULL TSRMLS_CC)) {
php_error_docref(NULL TSRMLS_CC, E_ERROR, "Could not call distributor function");
return 0;
}
//convert_to_string(ra->z_fun);
/* call extraction function */
MAKE_STD_ZVAL(z_argv0);
ZVAL_STRINGL(z_argv0, key, key_len, 0);
call_user_function(EG(function_table), NULL, ra->z_dist, &z_ret, 1, &z_argv0 TSRMLS_CC);
efree(z_argv0);
if(Z_TYPE(z_ret) != IS_LONG) {
zval_dtor(&z_ret);
return 0;
}
*pos = Z_LVAL(z_ret);
zval_dtor(&z_ret);
return 1;
}
zval *
ra_find_node(RedisArray *ra, const char *key, int key_len, int *out_pos TSRMLS_DC) {
uint32_t hash;
char *out;
int pos, out_len;
/* extract relevant part of the key */
out = ra_extract_key(ra, key, key_len, &out_len TSRMLS_CC);
if(!out)
return NULL;
if(ra->z_dist) {
if (!ra_call_distributor(ra, key, key_len, &pos TSRMLS_CC)) {
return NULL;
}
}
else {
/* hash */
hash = rcrc32(out, out_len);
efree(out);
/* get position on ring */
uint64_t h64 = hash;
h64 *= ra->count;
h64 /= 0xffffffff;
pos = (int)h64;
}
if(out_pos) *out_pos = pos;
return ra->redis[pos];
}
zval *
ra_find_node_by_name(RedisArray *ra, const char *host, int host_len TSRMLS_DC) {
int i;
for(i = 0; i < ra->count; ++i) {
if(strncmp(ra->hosts[i], host, host_len) == 0) {
return ra->redis[i];
}
}
return NULL;
}
char *
ra_find_key(RedisArray *ra, zval *z_args, const char *cmd, int *key_len) {
zval **zp_tmp;
int key_pos = 0; /* TODO: change this depending on the command */
if( zend_hash_num_elements(Z_ARRVAL_P(z_args)) == 0
|| zend_hash_quick_find(Z_ARRVAL_P(z_args), NULL, 0, key_pos, (void**)&zp_tmp) == FAILURE
|| Z_TYPE_PP(zp_tmp) != IS_STRING) {
return NULL;
}
*key_len = Z_STRLEN_PP(zp_tmp);
return Z_STRVAL_PP(zp_tmp);
}
void
ra_index_multi(zval *z_redis, long multi_value TSRMLS_DC) {
zval z_fun_multi, z_ret;
zval *z_args[1];
/* run MULTI */
ZVAL_STRING(&z_fun_multi, "MULTI", 0);
MAKE_STD_ZVAL(z_args[0]);
ZVAL_LONG(z_args[0], multi_value);
call_user_function(&redis_ce->function_table, &z_redis, &z_fun_multi, &z_ret, 1, z_args TSRMLS_CC);
efree(z_args[0]);
//zval_dtor(&z_ret);
}
static void
ra_index_change_keys(const char *cmd, zval *z_keys, zval *z_redis TSRMLS_DC) {
int i, argc;
zval z_fun, z_ret, **z_args;
/* alloc */
argc = 1 + zend_hash_num_elements(Z_ARRVAL_P(z_keys));
z_args = emalloc(argc * sizeof(zval*));
/* prepare first parameters */
ZVAL_STRING(&z_fun, cmd, 0);
MAKE_STD_ZVAL(z_args[0]);
ZVAL_STRING(z_args[0], PHPREDIS_INDEX_NAME, 0);
/* prepare keys */
for(i = 0; i < argc - 1; ++i) {
zval **zpp;
zend_hash_quick_find(Z_ARRVAL_P(z_keys), NULL, 0, i, (void**)&zpp);
z_args[i+1] = *zpp;
}
/* run cmd */
call_user_function(&redis_ce->function_table, &z_redis, &z_fun, &z_ret, argc, z_args TSRMLS_CC);
/* don't dtor z_ret, since we're returning z_redis */
efree(z_args[0]); /* free index name zval */
efree(z_args); /* free container */
}
void
ra_index_del(zval *z_keys, zval *z_redis TSRMLS_DC) {
ra_index_change_keys("SREM", z_keys, z_redis TSRMLS_CC);
}
void
ra_index_keys(zval *z_pairs, zval *z_redis TSRMLS_DC) {
/* Initialize key array */
zval *z_keys, **z_entry_pp;
MAKE_STD_ZVAL(z_keys);
#if PHP_VERSION_ID > 50300
array_init_size(z_keys, zend_hash_num_elements(Z_ARRVAL_P(z_pairs)));
#else
array_init(z_keys);
#endif
HashPosition pos;
/* Go through input array and add values to the key array */
zend_hash_internal_pointer_reset_ex(Z_ARRVAL_P(z_pairs), &pos);
while (zend_hash_get_current_data_ex(Z_ARRVAL_P(z_pairs), (void **)&z_entry_pp, &pos) == SUCCESS) {
char *key;
unsigned int key_len;
unsigned long num_key;
zval *z_new;
MAKE_STD_ZVAL(z_new);
switch (zend_hash_get_current_key_ex(Z_ARRVAL_P(z_pairs), &key, &key_len, &num_key, 1, &pos)) {
case HASH_KEY_IS_STRING:
ZVAL_STRINGL(z_new, key, (int)key_len - 1, 0);
zend_hash_next_index_insert(Z_ARRVAL_P(z_keys), &z_new, sizeof(zval *), NULL);
break;
case HASH_KEY_IS_LONG:
Z_TYPE_P(z_new) = IS_LONG;
Z_LVAL_P(z_new) = (long)num_key;
zend_hash_next_index_insert(Z_ARRVAL_P(z_keys), &z_new, sizeof(zval *), NULL);
break;
}
zend_hash_move_forward_ex(Z_ARRVAL_P(z_pairs), &pos);
}
/* add keys to index */
ra_index_change_keys("SADD", z_keys, z_redis TSRMLS_CC);
/* cleanup */
zval_dtor(z_keys);
efree(z_keys);
}
void
ra_index_key(const char *key, int key_len, zval *z_redis TSRMLS_DC) {
zval z_fun_sadd, z_ret, *z_args[2];
MAKE_STD_ZVAL(z_args[0]);
MAKE_STD_ZVAL(z_args[1]);
/* prepare args */
ZVAL_STRINGL(&z_fun_sadd, "SADD", 4, 0);
ZVAL_STRING(z_args[0], PHPREDIS_INDEX_NAME, 0);
ZVAL_STRINGL(z_args[1], key, key_len, 1);
/* run SADD */
call_user_function(&redis_ce->function_table, &z_redis, &z_fun_sadd, &z_ret, 2, z_args TSRMLS_CC);
/* don't dtor z_ret, since we're returning z_redis */
efree(z_args[0]);
efree(z_args[1]);
}
void
ra_index_exec(zval *z_redis, zval *return_value, int keep_all TSRMLS_DC) {
zval z_fun_exec, z_ret, **zp_tmp;
/* run EXEC */
ZVAL_STRING(&z_fun_exec, "EXEC", 0);
call_user_function(&redis_ce->function_table, &z_redis, &z_fun_exec, &z_ret, 0, NULL TSRMLS_CC);
/* extract first element of exec array and put into return_value. */
if(Z_TYPE(z_ret) == IS_ARRAY) {
if(return_value) {
if(keep_all) {
*return_value = z_ret;
zval_copy_ctor(return_value);
} else if(zend_hash_quick_find(Z_ARRVAL(z_ret), NULL, 0, 0, (void**)&zp_tmp) != FAILURE) {
*return_value = **zp_tmp;
zval_copy_ctor(return_value);
}
}
zval_dtor(&z_ret);
}
//zval *zptr = &z_ret;
//php_var_dump(&zptr, 0 TSRMLS_CC);
}
void
ra_index_discard(zval *z_redis, zval *return_value TSRMLS_DC) {
zval z_fun_discard, z_ret;
/* run DISCARD */
ZVAL_STRING(&z_fun_discard, "DISCARD", 0);
call_user_function(&redis_ce->function_table, &z_redis, &z_fun_discard, &z_ret, 0, NULL TSRMLS_CC);
zval_dtor(&z_ret);
}
void
ra_index_unwatch(zval *z_redis, zval *return_value TSRMLS_DC) {
zval z_fun_unwatch, z_ret;
/* run UNWATCH */
ZVAL_STRING(&z_fun_unwatch, "UNWATCH", 0);
call_user_function(&redis_ce->function_table, &z_redis, &z_fun_unwatch, &z_ret, 0, NULL TSRMLS_CC);
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_exists(Z_ARRVAL_P(ra->z_pure_cmds), cmd_up, cmd_len+1);
efree(cmd_up);
return !ret;
}
/* list keys from array index */
static long
ra_rehash_scan(zval *z_redis, char ***keys, int **key_lens, const char *cmd, const char *arg TSRMLS_DC) {
long count, i;
zval z_fun_smembers, z_ret, *z_arg, **z_data_pp;
HashTable *h_keys;
HashPosition pointer;
char *key;
int key_len;
/* arg */
MAKE_STD_ZVAL(z_arg);
ZVAL_STRING(z_arg, arg, 0);
/* run SMEMBERS */
ZVAL_STRING(&z_fun_smembers, cmd, 0);
call_user_function(&redis_ce->function_table, &z_redis, &z_fun_smembers, &z_ret, 1, &z_arg TSRMLS_CC);
efree(z_arg);
if(Z_TYPE(z_ret) != IS_ARRAY) { /* failure */
return -1; /* TODO: log error. */
}
h_keys = Z_ARRVAL(z_ret);
/* allocate key array */
count = zend_hash_num_elements(h_keys);
*keys = emalloc(count * sizeof(char*));
*key_lens = emalloc(count * sizeof(int));
for (i = 0, zend_hash_internal_pointer_reset_ex(h_keys, &pointer);
zend_hash_get_current_data_ex(h_keys, (void**) &z_data_pp, &pointer) == SUCCESS;
zend_hash_move_forward_ex(h_keys, &pointer), ++i) {
key = Z_STRVAL_PP(z_data_pp);
key_len = Z_STRLEN_PP(z_data_pp);
/* copy key and length */
(*keys)[i] = emalloc(1 + key_len);
memcpy((*keys)[i], key, key_len);
(*key_lens)[i] = key_len;
(*keys)[i][key_len] = 0; /* null-terminate string */
}
/* cleanup */
zval_dtor(&z_ret);
return count;
}
static long
ra_rehash_scan_index(zval *z_redis, char ***keys, int **key_lens TSRMLS_DC) {
return ra_rehash_scan(z_redis, keys, key_lens, "SMEMBERS", PHPREDIS_INDEX_NAME TSRMLS_CC);
}
/* list keys using KEYS command */
static long
ra_rehash_scan_keys(zval *z_redis, char ***keys, int **key_lens TSRMLS_DC) {
return ra_rehash_scan(z_redis, keys, key_lens, "KEYS", "*" TSRMLS_CC);
}
/* 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 TSRMLS_DC) {
int i;
zval z_fun_type, z_ret, *z_arg;
zval **z_data;
long success = 1;
MAKE_STD_ZVAL(z_arg);
/* Pipelined */
ra_index_multi(z_from, PIPELINE TSRMLS_CC);
/* prepare args */
ZVAL_STRINGL(&z_fun_type, "TYPE", 4, 0);
ZVAL_STRINGL(z_arg, key, key_len, 0);
/* run TYPE */
call_user_function(&redis_ce->function_table, &z_redis, &z_fun_type, &z_ret, 1, &z_arg TSRMLS_CC);
ZVAL_STRINGL(&z_fun_type, "TTL", 3, 0);
ZVAL_STRINGL(z_arg, key, key_len, 0);
/* run TYPE */
call_user_function(&redis_ce->function_table, &z_redis, &z_fun_type, &z_ret, 1, &z_arg TSRMLS_CC);
/* cleanup */
efree(z_arg);
/* Get the result from the pipeline. */
ra_index_exec(z_from, &z_ret, 1 TSRMLS_CC);
if(Z_TYPE(z_ret) == IS_ARRAY) {
HashTable *retHash = Z_ARRVAL(z_ret);
for(i = 0, zend_hash_internal_pointer_reset(retHash);
zend_hash_has_more_elements(retHash) == SUCCESS;
zend_hash_move_forward(retHash)) {
if(zend_hash_get_current_data(retHash, (void**)&z_data) == FAILURE) {
success = 0;
break;
}
if(Z_TYPE_PP(z_data) != IS_LONG) {
success = 0;
break;
}
/* Get the result - Might change in the future to handle doubles as well */
res[i] = Z_LVAL_PP(z_data);
i++;
}
}
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 TSRMLS_DC) {
zval z_fun_srem, z_ret, *z_args[2];
/* run SREM on source index */
ZVAL_STRINGL(&z_fun_srem, "SREM", 4, 0);
MAKE_STD_ZVAL(z_args[0]);
ZVAL_STRING(z_args[0], PHPREDIS_INDEX_NAME, 0);
MAKE_STD_ZVAL(z_args[1]);
ZVAL_STRINGL(z_args[1], key, key_len, 0);
call_user_function(&redis_ce->function_table, &z_redis, &z_fun_srem, &z_ret, 2, z_args TSRMLS_CC);
/* cleanup */
efree(z_args[0]);
efree(z_args[1]);
}
/* delete key from source server during rehashing */
static zend_bool
ra_del_key(const char *key, int key_len, zval *z_from TSRMLS_DC) {
zval z_fun_del, z_ret, *z_args;
/* in a transaction */
ra_index_multi(z_from, MULTI TSRMLS_CC);
/* run DEL on source */
MAKE_STD_ZVAL(z_args);
ZVAL_STRINGL(&z_fun_del, "DEL", 3, 0);
ZVAL_STRINGL(z_args, key, key_len, 0);
call_user_function(&redis_ce->function_table, &z_from, &z_fun_del, &z_ret, 1, &z_args TSRMLS_CC);
efree(z_args);
/* remove key from index */
ra_remove_from_index(z_from, key, key_len TSRMLS_CC);
/* close transaction */
ra_index_exec(z_from, NULL, 0 TSRMLS_CC);
return 1;
}
static zend_bool
ra_expire_key(const char *key, int key_len, zval *z_to, long ttl TSRMLS_DC) {
zval z_fun_expire, z_ret, *z_args[2];
if (ttl > 0)
{
/* run EXPIRE on target */
MAKE_STD_ZVAL(z_args[0]);
MAKE_STD_ZVAL(z_args[1]);
ZVAL_STRINGL(&z_fun_expire, "EXPIRE", 6, 0);
ZVAL_STRINGL(z_args[0], key, key_len, 0);
ZVAL_LONG(z_args[1], ttl);
call_user_function(&redis_ce->function_table, &z_to, &z_fun_expire, &z_ret, 2, z_args TSRMLS_CC);
/* cleanup */
efree(z_args[0]);
efree(z_args[1]);
}
return 1;
}
static zend_bool
ra_move_zset(const char *key, int key_len, zval *z_from, zval *z_to, long ttl TSRMLS_DC) {
zval z_fun_zrange, z_fun_zadd, z_ret, *z_args[4], **z_zadd_args, **z_score_pp;
int count;
HashTable *h_zset_vals;
char *val;
unsigned int val_len;
int i;
unsigned long idx;
/* run ZRANGE key 0 -1 WITHSCORES on source */
ZVAL_STRINGL(&z_fun_zrange, "ZRANGE", 6, 0);
for(i = 0; i < 4; ++i) {
MAKE_STD_ZVAL(z_args[i]);
}
ZVAL_STRINGL(z_args[0], key, key_len, 0);
ZVAL_STRINGL(z_args[1], "0", 1, 0);
ZVAL_STRINGL(z_args[2], "-1", 2, 0);
ZVAL_BOOL(z_args[3], 1);
call_user_function(&redis_ce->function_table, &z_from, &z_fun_zrange, &z_ret, 4, z_args TSRMLS_CC);
/* cleanup zrange args */
for(i = 0; i < 4; ++i) {
efree(z_args[i]); /* FIXME */
}
if(Z_TYPE(z_ret) != IS_ARRAY) { /* key not found or replaced */
/* TODO: report? */
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 = emalloc((1 + 2*count) * sizeof(zval*));
for(i = 1, zend_hash_internal_pointer_reset(h_zset_vals);
zend_hash_has_more_elements(h_zset_vals) == SUCCESS;
zend_hash_move_forward(h_zset_vals)) {
if(zend_hash_get_current_data(h_zset_vals, (void**)&z_score_pp) == FAILURE) {
continue;
}
/* add score */
convert_to_double(*z_score_pp);
MAKE_STD_ZVAL(z_zadd_args[i]);
ZVAL_DOUBLE(z_zadd_args[i], Z_DVAL_PP(z_score_pp));
/* add value */
MAKE_STD_ZVAL(z_zadd_args[i+1]);
switch (zend_hash_get_current_key_ex(h_zset_vals, &val, &val_len, &idx, 0, NULL)) {
case HASH_KEY_IS_STRING:
ZVAL_STRINGL(z_zadd_args[i+1], val, (int)val_len-1, 0); /* we have to remove 1 because it is an array key. */
break;
case HASH_KEY_IS_LONG:
ZVAL_LONG(z_zadd_args[i+1], (long)idx);
break;
default:
return -1; // Todo: log error
break;
}
i += 2;
}
/* run ZADD on target */
ZVAL_STRINGL(&z_fun_zadd, "ZADD", 4, 0);
MAKE_STD_ZVAL(z_zadd_args[0]);
ZVAL_STRINGL(z_zadd_args[0], key, key_len, 0);
call_user_function(&redis_ce->function_table, &z_to, &z_fun_zadd, &z_ret, 1 + 2 * count, z_zadd_args TSRMLS_CC);
/* Expire if needed */
ra_expire_key(key, key_len, z_to, ttl TSRMLS_CC);
/* cleanup */
for(i = 0; i < 1 + 2 * count; ++i) {
efree(z_zadd_args[i]);
}
return 1;
}
static zend_bool
ra_move_string(const char *key, int key_len, zval *z_from, zval *z_to, long ttl TSRMLS_DC) {
zval z_fun_get, z_fun_set, z_ret, *z_args[3];
/* run GET on source */
MAKE_STD_ZVAL(z_args[0]);
ZVAL_STRINGL(&z_fun_get, "GET", 3, 0);
ZVAL_STRINGL(z_args[0], key, key_len, 0);
call_user_function(&redis_ce->function_table, &z_from, &z_fun_get, &z_ret, 1, z_args TSRMLS_CC);
if(Z_TYPE(z_ret) != IS_STRING) { /* key not found or replaced */
/* TODO: report? */
efree(z_args[0]);
return 0;
}
/* run SET on target */
MAKE_STD_ZVAL(z_args[1]);
if (ttl > 0) {
MAKE_STD_ZVAL(z_args[2]);
ZVAL_STRINGL(&z_fun_set, "SETEX", 5, 0);
ZVAL_STRINGL(z_args[0], key, key_len, 0);
ZVAL_LONG(z_args[1], ttl);
ZVAL_STRINGL(z_args[2], Z_STRVAL(z_ret), Z_STRLEN(z_ret), 1); /* copy z_ret to arg 1 */
call_user_function(&redis_ce->function_table, &z_to, &z_fun_set, &z_ret, 3, z_args TSRMLS_CC);
/* cleanup */
efree(z_args[1]);
zval_dtor(z_args[2]);
efree(z_args[2]);
}
else {
ZVAL_STRINGL(&z_fun_set, "SET", 3, 0);
ZVAL_STRINGL(z_args[0], key, key_len, 0);
ZVAL_STRINGL(z_args[1], Z_STRVAL(z_ret), Z_STRLEN(z_ret), 1); /* copy z_ret to arg 1 */
call_user_function(&redis_ce->function_table, &z_to, &z_fun_set, &z_ret, 2, z_args TSRMLS_CC);
/* cleanup */
zval_dtor(z_args[1]);
efree(z_args[1]);
}
/* cleanup */
efree(z_args[0]);
return 1;
}
static zend_bool
ra_move_hash(const char *key, int key_len, zval *z_from, zval *z_to, long ttl TSRMLS_DC) {
zval z_fun_hgetall, z_fun_hmset, z_ret, *z_args[2];
/* run HGETALL on source */
MAKE_STD_ZVAL(z_args[0]);
ZVAL_STRINGL(&z_fun_hgetall, "HGETALL", 7, 0);
ZVAL_STRINGL(z_args[0], key, key_len, 0);
call_user_function(&redis_ce->function_table, &z_from, &z_fun_hgetall, &z_ret, 1, z_args TSRMLS_CC);
if(Z_TYPE(z_ret) != IS_ARRAY) { /* key not found or replaced */
/* TODO: report? */
efree(z_args[0]);
return 0;
}
/* run HMSET on target */
ZVAL_STRINGL(&z_fun_hmset, "HMSET", 5, 0);
ZVAL_STRINGL(z_args[0], key, key_len, 0);
z_args[1] = &z_ret; /* copy z_ret to arg 1 */
call_user_function(&redis_ce->function_table, &z_to, &z_fun_hmset, &z_ret, 2, z_args TSRMLS_CC);
/* Expire if needed */
ra_expire_key(key, key_len, z_to, ttl TSRMLS_CC);
/* cleanup */
efree(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 TSRMLS_DC) {
zval z_fun_retrieve, z_fun_sadd, z_ret, **z_retrieve_args, **z_sadd_args, **z_data_pp;
int count, i;
HashTable *h_set_vals;
/* run retrieval command on source */
z_retrieve_args = emalloc((1+list_count) * sizeof(zval*));
ZVAL_STRING(&z_fun_retrieve, cmd_list[0], 0); /* set the command */
/* set the key */
MAKE_STD_ZVAL(z_retrieve_args[0]);
ZVAL_STRINGL(z_retrieve_args[0], key, key_len, 0);
/* possibly add some other args if they were provided. */
for(i = 1; i < list_count; ++i) {
MAKE_STD_ZVAL(z_retrieve_args[i]);
ZVAL_STRING(z_retrieve_args[i], cmd_list[i], 0);
}
call_user_function(&redis_ce->function_table, &z_from, &z_fun_retrieve, &z_ret, list_count, z_retrieve_args TSRMLS_CC);
/* cleanup */
for(i = 0; i < list_count; ++i) {
efree(z_retrieve_args[i]);
}
efree(z_retrieve_args);
if(Z_TYPE(z_ret) != IS_ARRAY) { /* key not found or replaced */
/* TODO: report? */
return 0;
}
/* run SADD/RPUSH on target */
h_set_vals = Z_ARRVAL(z_ret);
count = zend_hash_num_elements(h_set_vals);
z_sadd_args = emalloc((1 + count) * sizeof(zval*));
ZVAL_STRING(&z_fun_sadd, cmd_add[0], 0);
MAKE_STD_ZVAL(z_sadd_args[0]); /* add key */
ZVAL_STRINGL(z_sadd_args[0], key, key_len, 0);
for(i = 0, zend_hash_internal_pointer_reset(h_set_vals);
zend_hash_has_more_elements(h_set_vals) == SUCCESS;
zend_hash_move_forward(h_set_vals), i++) {
if(zend_hash_get_current_data(h_set_vals, (void**)&z_data_pp) == FAILURE) {
continue;
}
/* add set elements */
MAKE_STD_ZVAL(z_sadd_args[i+1]);
*(z_sadd_args[i+1]) = **z_data_pp;
zval_copy_ctor(z_sadd_args[i+1]);
}
call_user_function(&redis_ce->function_table, &z_to, &z_fun_sadd, &z_ret, count+1, z_sadd_args TSRMLS_CC);
/* Expire if needed */
ra_expire_key(key, key_len, z_to, ttl TSRMLS_CC);
/* cleanup */
efree(z_sadd_args[0]); /* no dtor at [0] */
for(i = 0; i < count; ++i) {
zval_dtor(z_sadd_args[i + 1]);
efree(z_sadd_args[i + 1]);
}
efree(z_sadd_args);
return 1;
}
static zend_bool
ra_move_set(const char *key, int key_len, zval *z_from, zval *z_to, long ttl TSRMLS_DC) {
const char *cmd_list[] = {"SMEMBERS"};
const char *cmd_add[] = {"SADD"};
return ra_move_collection(key, key_len, z_from, z_to, 1, cmd_list, 1, cmd_add, ttl TSRMLS_CC);
}
static zend_bool
ra_move_list(const char *key, int key_len, zval *z_from, zval *z_to, long ttl TSRMLS_DC) {
const char *cmd_list[] = {"LRANGE", "0", "-1"};
const char *cmd_add[] = {"RPUSH"};
return ra_move_collection(key, key_len, z_from, z_to, 3, cmd_list, 1, cmd_add, ttl TSRMLS_CC);
}
void
ra_move_key(const char *key, int key_len, zval *z_from, zval *z_to TSRMLS_DC) {
long res[2], type, ttl;
zend_bool success = 0;
if (ra_get_key_type(z_from, key, key_len, z_from, res TSRMLS_CC)) {
type = res[0];
ttl = res[1];
/* open transaction on target server */
ra_index_multi(z_to, MULTI TSRMLS_CC);
switch(type) {
case REDIS_STRING:
success = ra_move_string(key, key_len, z_from, z_to, ttl TSRMLS_CC);
break;
case REDIS_SET:
success = ra_move_set(key, key_len, z_from, z_to, ttl TSRMLS_CC);
break;
case REDIS_LIST:
success = ra_move_list(key, key_len, z_from, z_to, ttl TSRMLS_CC);
break;
case REDIS_ZSET:
success = ra_move_zset(key, key_len, z_from, z_to, ttl TSRMLS_CC);
break;
case REDIS_HASH:
success = ra_move_hash(key, key_len, z_from, z_to, ttl TSRMLS_CC);
break;
default:
/* TODO: report? */
break;
}
}
if(success) {
ra_del_key(key, key_len, z_from TSRMLS_CC);
ra_index_key(key, key_len, z_to TSRMLS_CC);
}
/* close transaction */
ra_index_exec(z_to, NULL, 0 TSRMLS_CC);
}
/* callback with the current progress, with hostname and count */
static void zval_rehash_callback(zend_fcall_info *z_cb, zend_fcall_info_cache *z_cb_cache,
const char *hostname, long count TSRMLS_DC) {
zval *z_ret = NULL, **z_args[2];
zval *z_host, *z_count;
z_cb->retval_ptr_ptr = &z_ret;
z_cb->params = (struct _zval_struct ***)&z_args;
z_cb->param_count = 2;
z_cb->no_separation = 0;
/* run cb(hostname, count) */
MAKE_STD_ZVAL(z_host);
ZVAL_STRING(z_host, hostname, 0);
z_args[0] = &z_host;
MAKE_STD_ZVAL(z_count);
ZVAL_LONG(z_count, count);
z_args[1] = &z_count;
zend_call_function(z_cb, z_cb_cache TSRMLS_CC);
/* cleanup */
efree(z_host);
efree(z_count);
if(z_ret)
efree(z_ret);
}
static void
ra_rehash_server(RedisArray *ra, zval *z_redis, const char *hostname, zend_bool b_index,
zend_fcall_info *z_cb, zend_fcall_info_cache *z_cb_cache TSRMLS_DC) {
char **keys;
int *key_lens;
long count, i;
int target_pos;
zval *z_target;
/* list all keys */
if(b_index) {
count = ra_rehash_scan_index(z_redis, &keys, &key_lens TSRMLS_CC);
} else {
count = ra_rehash_scan_keys(z_redis, &keys, &key_lens TSRMLS_CC);
}
/* callback */
if(z_cb && z_cb_cache) {
zval_rehash_callback(z_cb, z_cb_cache, hostname, count TSRMLS_CC);
}
/* for each key, redistribute */
for(i = 0; i < count; ++i) {
/* check that we're not moving to the same node. */
z_target = ra_find_node(ra, keys[i], key_lens[i], &target_pos TSRMLS_CC);
if(strcmp(hostname, ra->hosts[target_pos])) { /* different host */
/* php_printf("move [%s] from [%s] to [%s]\n", keys[i], hostname, ra->hosts[target_pos]); */
ra_move_key(keys[i], key_lens[i], z_redis, z_target TSRMLS_CC);
}
}
/* cleanup */
for(i = 0; i < count; ++i) {
efree(keys[i]);
}
efree(keys);
efree(key_lens);
}
void
ra_rehash(RedisArray *ra, zend_fcall_info *z_cb, zend_fcall_info_cache *z_cb_cache TSRMLS_DC) {
int i;
/* redistribute the data, server by server. */
if(!ra->prev)
return; /* TODO: compare the two rings for equality */
for(i = 0; i < ra->prev->count; ++i) {
ra_rehash_server(ra, ra->prev->redis[i], ra->prev->hosts[i], ra->index, z_cb, z_cb_cache TSRMLS_CC);
}
}
Reference in New Issue View Git Blame Copy Permalink