#include "php_redis.h"

#include "common.h"

#include "library.h"

#include "redis_commands.h"

#include "cluster_library.h"

#include "crc16.h"

#include <zend_exceptions.h>

#include <errno.h>

extern zend_class_entry *redis_cluster_exception_ce;

int le_cluster_slot_cache;

/* Debugging methods/

static void cluster_dump_nodes(redisCluster *c) {

redisClusterNode *p;

ZEND_HASH_FOREACH_PTR(c->nodes, p) {

if (p == NULL) {

continue;

}

const char *slave = (p->slave) ? "slave" : "master";

php_printf("%d %s %d %d", p->sock->port, slave,p->sock->prefix_len,

p->slot);

php_printf("\n");

} ZEND_HASH_FOREACH_END();

}

static void cluster_log(char *fmt, ...)

{

va_list args;

char buffer[1024];

va_start(args, fmt);

vsnprintf(buffer,sizeof(buffer),fmt,args);

va_end(args);

fprintf(stderr, "%s\n", buffer);

}

// Debug function to dump a clusterReply structure recursively

static void dump_reply(clusterReply *reply, int indent) {

smart_string buf = {0};

int i;

switch(reply->type) {

case TYPE_ERR:

smart_string_appendl(&buf, "(error) ", sizeof("(error) ")-1);

smart_string_appendl(&buf, reply->str, reply->len);

break;

case TYPE_LINE:

smart_string_appendl(&buf, reply->str, reply->len);

break;

case TYPE_INT:

smart_string_appendl(&buf, "(integer) ", sizeof("(integer) ")-1);

smart_string_append_long(&buf, reply->integer);

break;

case TYPE_BULK:

smart_string_appendl(&buf,"\"", 1);

smart_string_appendl(&buf, reply->str, reply->len);

smart_string_appendl(&buf, "\"", 1);

break;

case TYPE_MULTIBULK:

if (reply->elements < 0) {

smart_string_appendl(&buf, "(nil)", sizeof("(nil)")-1);

} else {

for (i = 0; i < reply->elements; i++) {

dump_reply(reply->element[i], indent+2);

}

}

break;

default:

break;

}

if (buf.len > 0) {

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

php_printf(" ");

}

smart_string_0(&buf);

php_printf("%s", buf.c);

php_printf("\n");

efree(buf.c);

}

}

*/

/* MULTI BULK processing callbacks */

static int mbulk_resp_loop(RedisSock *redis_sock, zval *z_result,

long long count, RedisCmdCtx ctx);

static int mbulk_resp_loop_raw(RedisSock *redis_sock, zval *z_result,

long long count, RedisCmdCtx ctx);

static int mbulk_resp_loop_zipstr(RedisSock *redis_sock, zval *z_result,

long long count, RedisCmdCtx ctx);

static int mbulk_resp_loop_dbl(RedisSock *redis_sock, zval *z_result,

long long count, RedisCmdCtx ctx);

static int mbulk_resp_loop_zipdbl(RedisSock *redis_sock, zval *z_result,

long long count, RedisCmdCtx ctx);

static int mbulk_resp_loop_assoc(RedisSock *redis_sock, zval *z_result,

long long count, RedisCmdCtx ctx);

static int

cluster_parse_reply_len(redisCluster *c, REDIS_REPLY_TYPE reply_type, size_t line_len)

{

char *endptr;

long long n;

if (reply_type == TYPE_LINE) {

c->reply_len = line_len;

return SUCCESS;

}

errno = 0;

n = strtoll(c->line_reply, &endptr, 10);

if (endptr == c->line_reply || *endptr != '\0' || errno == ERANGE) {

goto failure;

}

if ((reply_type == TYPE_BULK && (n < -1 || n > INT_MAX - 2)) ||

(reply_type == TYPE_MULTIBULK && (n < -1 || n > INT_MAX)))

{

goto failure;

}

c->reply_len = n;

return SUCCESS;

failure:

zend_throw_exception_ex(redis_cluster_exception_ce, 0,

"protocol error, invalid reply length");

return FAILURE;

}

/* Recursively free our reply object. If free_data is non-zero we'll also free

* the payload data (strings) themselves. If not, we just free the structs */

void cluster_free_reply(clusterReply *reply, int free_data) {

long long i;

switch(reply->type) {

case TYPE_ERR:

case TYPE_LINE:

case TYPE_BULK:

if (free_data && reply->str)

efree(reply->str);

break;

case TYPE_MULTIBULK:

if (reply->element) {

if (reply->elements > 0) {

for (i = 0; i < reply->elements && reply->element[i]; i++) {

cluster_free_reply(reply->element[i], free_data);

}

}

efree(reply->element);

}

break;

default:

break;

}

efree(reply);

}

static int

cluster_multibulk_resp_recursive(RedisSock *sock, size_t elements,

clusterReply **element, int status_strings)

{

int i;

size_t sz;

clusterReply *r;

long len;

char buf[1024];

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

r = element[i] = ecalloc(1, sizeof(clusterReply));

// Bomb out, flag error condition on a communication failure

if (redis_read_reply_type(sock, &r->type, &len) < 0) {

return FAILURE;

}

/* Set our reply len */

r->len = len;

switch(r->type) {

case TYPE_ERR:

case TYPE_LINE:

if (redis_sock_gets(sock,buf,sizeof(buf),&sz) < 0) {

return FAILURE;

}

r->len = (long long)sz;

if (status_strings) r->str = estrndup(buf, r->len);

break;

case TYPE_INT:

r->integer = len;

break;

case TYPE_BULK:

if (r->len >= 0) {

r->str = redis_sock_read_bulk_reply(sock,r->len);

if (!r->str) {

return FAILURE;

}

}

break;

case TYPE_MULTIBULK:

r->elements = r->len;

if (r->elements > 0) {

r->element = ecalloc(r->len, sizeof(*r->element));

if (cluster_multibulk_resp_recursive(sock, r->elements, r->element, status_strings) < 0) {

return FAILURE;

}

}

break;

default:

return FAILURE;

}

}

return SUCCESS;

}

/* Return the socket for a slot and slave index */

static RedisSock *cluster_slot_sock(redisCluster *c, unsigned short slot,

zend_ulong slaveidx)

{

redisClusterNode *node;

/* Return the master if we're not looking for a slave */

if (slaveidx == 0) {

return cluster_slot_master_sock(c, slot);

}

/* Abort if we can't find this slave */

if (!cluster_slot_slaves(c, slot) ||

(node = zend_hash_index_find_ptr(cluster_slot_slaves(c,slot), slaveidx)) == NULL

) {

return NULL;

}

/* Success, return the slave */

return node->sock;

}

/* Read the response from a cluster */

clusterReply *cluster_read_resp(redisCluster *c, int status_strings) {

return cluster_read_sock_resp(c->cmd_sock, c->reply_type,

status_strings ? c->line_reply : NULL,

c->reply_len);

}

/* Read any sort of response from the socket, having already issued the

* command and consumed the reply type and meta info (length) */

clusterReply*

cluster_read_sock_resp(RedisSock *redis_sock, REDIS_REPLY_TYPE type,

char *line_reply, long long len)

{

clusterReply *r;

r = ecalloc(1, sizeof(clusterReply));

r->type = type;

switch(r->type) {

case TYPE_INT:

r->integer = len;

break;

case TYPE_LINE:

if (line_reply) {

r->str = estrndup(line_reply, len);

r->len = len;

}

REDIS_FALLTHROUGH;

case TYPE_ERR:

return r;

case TYPE_BULK:

r->len = len;

r->str = redis_sock_read_bulk_reply(redis_sock, len);

if (r->len != -1 && !r->str) {

cluster_free_reply(r, 1);

return NULL;

}

break;

case TYPE_MULTIBULK:

r->elements = len;

if (r->elements > 0) {

r->element = ecalloc(len, sizeof(clusterReply*));

if (cluster_multibulk_resp_recursive(redis_sock, len, r->element, line_reply != NULL) < 0) {

cluster_free_reply(r, 1);

return NULL;

}

}

break;

default:

cluster_free_reply(r, 1);

return NULL;

}

// Success, return the reply

return r;

}

/* Connect the socket if needed and make sure it's still alive. Note that

* redis_check_eof can reconnect transparently, which drops any per-connection

* state we had sent (e.g. READONLY). */

static zend_always_inline zend_bool

cluster_sock_live(RedisSock *redis_sock)

{

if (redis_sock == NULL)

return 0;

if (redis_sock_server_open(redis_sock) != 0 || redis_sock->stream == NULL)

return 0;

return redis_check_eof(redis_sock, 0, 1) == 0;

}

/* Require a complete write; leave retry decisions to the caller. */

static zend_always_inline zend_bool

cluster_write_payload(RedisSock *redis_sock, const char *buf, size_t len)

{

ssize_t nwritten = redis_sock_write_raw(redis_sock, buf, len);

return nwritten >= 0 && nwritten == len;

}

static zend_always_inline zend_bool

cluster_send_payload(RedisSock *redis_sock, const char *buf, size_t len)

{

return cluster_sock_live(redis_sock) &&

cluster_write_payload(redis_sock, buf, len);

}

static zend_always_inline zend_bool

cluster_validate_reply_type(RedisSock *redis_sock, REDIS_REPLY_TYPE type)

{

if (redis_check_eof(redis_sock, 1, 1) != 0)

return 0;

return redis_sock_getc(redis_sock) == type;

}

/*

* Helpers to send various 'control type commands to a specific node, e.g.

* MULTI, ASKING, READONLY, READWRITE, etc

*/

/* Send a command to the specific socket and validate reply type */

static int cluster_send_direct(RedisSock *redis_sock, const char *cmd, size_t cmd_len,

REDIS_REPLY_TYPE type)

{

char buf[1024];

/* Connect to the socket if we aren't yet and send our command, validate the reply type, and consume the first line */

if (!cluster_send_payload(redis_sock, cmd, cmd_len) ||

!cluster_validate_reply_type(redis_sock, type) ||

!redis_sock_gets_raw(redis_sock, buf, sizeof(buf))) return -1;

/* Success! */

return 0;

}

static int cluster_send_asking(RedisSock *redis_sock) {

return cluster_send_direct(redis_sock, ZEND_STRL(RESP_ASKING_CMD), TYPE_LINE);

}

/* Send READONLY to a specific RedisSock unless it's already flagged as being

* in READONLY mode. If we can send the command, we flag the socket as being

* in that mode. */

static int cluster_send_readonly(RedisSock *redis_sock) {

int ret;

/* We don't have to do anything if we're already in readonly mode */

if (redis_sock->readonly) return 0;

/* Return success if we can send it */

ret = cluster_send_direct(redis_sock, ZEND_STRL(RESP_READONLY_CMD), TYPE_LINE);

/* Flag this socket as READONLY if our command worked */

redis_sock->readonly = !ret;

/* Return the result of our send */

return ret;

}

/* Send a command to a replica, issuing READONLY first if this connection

* doesn't have it. Liveness is checked once, before READONLY, because a

* transparent reconnect after that point would drop it again. */

static zend_bool

cluster_send_replica_payload(RedisSock *redis_sock, const char *buf, size_t len)

{

return cluster_sock_live(redis_sock) &&

cluster_send_readonly(redis_sock) == 0 &&

cluster_write_payload(redis_sock, buf, len);

}

/* Send MULTI to a specific ReidsSock */

static int cluster_send_multi(redisCluster *c, short slot) {

if (cluster_send_direct(cluster_slot_master_sock(c,slot), ZEND_STRL(RESP_MULTI_CMD), TYPE_LINE) == 0) {

c->flags->txBytes += sizeof(RESP_MULTI_CMD) - 1;

c->cmd_sock->mode = MULTI;

return 0;

}

return -1;

}

/* Send EXEC to a given slot. We can use the normal command processing mechanism

* here because we know we'll only have sent MULTI to the master nodes. We can't

* failover inside a transaction, as we don't know if the transaction will only

* be readonly commands, or contain write commands as well */

PHP_REDIS_API int cluster_send_exec(redisCluster *c, short slot) {

int retval;

/* Send exec */

retval = cluster_send_slot(c, slot, ZEND_STRL(RESP_EXEC_CMD), TYPE_MULTIBULK);

/* We'll either get a length corresponding to the number of commands sent to

* this node, or -1 in the case of EXECABORT or WATCH failure. */

c->multi_len[slot] = c->reply_len > 0 ? 1 : -1;

/* Return our retval */

return retval;

}

PHP_REDIS_API int cluster_send_discard(redisCluster *c, short slot) {

if (cluster_send_direct(cluster_slot_master_sock(c,slot), ZEND_STRL(RESP_DISCARD_CMD), TYPE_LINE))

{

return 0;

}

return -1;

}

/* Group owned WATCH keys by hash slot. */

/* Free cluster distribution list inside a HashTable */

static void cluster_dist_free_ht(zval *p) {

clusterDistList *dl = Z_PTR_P(p);

size_t i;

for (i = 0; i < dl->len; i++) {

zend_string_release(dl->keys[i]);

}

efree(dl->keys);

efree(dl);

}

/* Spin up a HashTable that will contain distribution lists */

HashTable *cluster_dist_create(void) {

HashTable *ret;

ALLOC_HASHTABLE(ret);

zend_hash_init(ret, 0, NULL, cluster_dist_free_ht, 0);

return ret;

}

/* Free distribution list */

void cluster_dist_free(HashTable *ht) {

zend_hash_destroy(ht);

efree(ht);

}

/* Create a clusterDistList object */

static clusterDistList *cluster_dl_create(void) {

clusterDistList *dl;

dl = emalloc(sizeof(clusterDistList));

dl->keys = emalloc(CLUSTER_KEYDIST_ALLOC * sizeof(*dl->keys));

dl->size = CLUSTER_KEYDIST_ALLOC;

dl->len = 0;

return dl;

}

/* Transfer ownership of a prefixed key to the list. */

static void cluster_dl_add_key(clusterDistList *dl, zend_string *key)

{

// Reallocate if required

if (dl->len == dl->size) {

dl->keys = safe_erealloc(dl->keys, dl->size, 2 * sizeof(*dl->keys), 0);

dl->size *= 2;

}

dl->keys[dl->len++] = key;

}

/* Borrow key, retaining an owned prefixed string on success. */

int cluster_dist_add_key(redisCluster *c, HashTable *ht, zend_string *key)

{

short slot;

clusterDistList *dl;

// Prefix our key and hash it

key = redis_key_prefix_zstr(c->flags, key);

slot = cluster_hash_key_zstr(key);

// We can't do this if we don't fully understand the keyspace

if (c->master[slot] == NULL) {

zend_string_release(key);

return FAILURE;

}

// Look for this slot

if ((dl = zend_hash_index_find_ptr(ht, (zend_ulong)slot)) == NULL) {

dl = cluster_dl_create();

zend_hash_index_update_ptr(ht, (zend_ulong)slot, dl);

}

// Now actually add this key

cluster_dl_add_key(dl, key);

return SUCCESS;

}

/* Free allocated memory for a clusterMultiCmd */

void cluster_multi_free(clusterMultiCmd *mc) {

redis_cmd_free(mc->cmd);

}

/* Add an argument to a clusterMultiCmd */

void cluster_multi_add(clusterMultiCmd *mc, const char *data, int data_len) {

if (mc->cmd == NULL) {

mc->cmd = redis_cmd_create(NULL, mc->kw, mc->kw_len);

}

mc->argc++;

redis_cmd_cat_str(mc->cmd, data, data_len);

}

/* Finalize a clusterMultiCmd */

void cluster_multi_fini(clusterMultiCmd *mc) {

if (mc->cmd == NULL) {

mc->cmd = redis_cmd_create(NULL, mc->kw, mc->kw_len);

}

}

/* Set our last error string encountered */

static void

cluster_set_err(redisCluster *c, const char *err, int err_len)

{

// Free our last error

if (c->err != NULL) {

zend_string_release(c->err);

c->err = NULL;

}

if (err != NULL && err_len > 0) {

c->err = zend_string_init(err, err_len, 0);

if (err_len >= sizeof("CLUSTERDOWN") - 1 &&

!memcmp(err, "CLUSTERDOWN", sizeof("CLUSTERDOWN") - 1)

) {

c->clusterdown = 1;

}

}

}

/* Destructor for slaves */

static void ht_free_slave(zval *data) {

redisClusterNode *node = Z_PTR_P(data);

if (node) {

cluster_free_node(node);

}

}

/* Get the hash slot for a given key */

unsigned short cluster_hash_key(const char *key, int len) {

int s, e;

// Find first occurrence of {, if any

for (s = 0; s < len; s++) {

if (key[s]=='{') break;

}

// There is no '{', hash everything

if (s == len) return crc16(key, len) & REDIS_CLUSTER_MOD;

// Found it, look for a tailing '}'

for (e =s + 1; e < len; e++) {

if (key[e] == '}') break;

}

// Hash the whole key if we don't find a tailing } or if {} is empty

if (e == len || e == s+1) return crc16(key, len) & REDIS_CLUSTER_MOD;

// Hash just the bit between { and }

return crc16(key + s + 1, e - s - 1) & REDIS_CLUSTER_MOD;

}

unsigned short cluster_hash_key_zstr(zend_string *key) {

return cluster_hash_key(ZSTR_VAL(key), ZSTR_LEN(key));

}

/* Grab the current time in milliseconds */

long long mstime(void) {

struct timeval tv;

long long mst;

gettimeofday(&tv, NULL);

mst = ((long long)tv.tv_sec)*1000;

mst += tv.tv_usec/1000;

return mst;

}

/* Hash a key from a ZVAL */

unsigned short cluster_hash_key_zval(zval *z_key) {

const char *kptr;

char buf[255];

int klen;

// Switch based on ZVAL type

switch(Z_TYPE_P(z_key)) {

case IS_STRING:

kptr = Z_STRVAL_P(z_key);

klen = Z_STRLEN_P(z_key);

break;

case IS_LONG:

klen = snprintf(buf,sizeof(buf),ZEND_LONG_FMT,Z_LVAL_P(z_key));

kptr = buf;

break;

case IS_DOUBLE:

klen = snprintf(buf,sizeof(buf),"%f",Z_DVAL_P(z_key));

kptr = buf;

break;

case IS_ARRAY:

kptr = "Array";

klen = sizeof("Array")-1;

break;

case IS_OBJECT:

kptr = "Object";

klen = sizeof("Object")-1;

break;

default:

kptr = "";

klen = 0;

}

// Hash the string representation

return cluster_hash_key(kptr, klen);

}

/* Fisher-Yates shuffle for integer array */

static void fyshuffle(int *array, size_t len) {

int temp, n = len;

size_t r;

/* Randomize */

while (n > 1) {

r = ((int)((double)n-- * (rand() / (RAND_MAX+1.0))));

temp = array[n];

array[n] = array[r];

array[r] = temp;

};

}

/* Execute a CLUSTER SLOTS command against the seed socket, and return the

* reply or NULL on failure. */

clusterReply* cluster_get_slots(RedisSock *redis_sock)

{

clusterReply *r;

REDIS_REPLY_TYPE type;

long len;

// Send the command to the socket and consume reply type

if (redis_sock_write(redis_sock, RESP_CLUSTER_SLOTS_CMD,

sizeof(RESP_CLUSTER_SLOTS_CMD)-1) < 0 ||

redis_read_reply_type(redis_sock, &type, &len) < 0)

{

return NULL;

}

// Consume the rest of our response

if ((r = cluster_read_sock_resp(redis_sock, type, NULL, len)) == NULL ||

r->type != TYPE_MULTIBULK || r->elements < 1)

{

if (r) cluster_free_reply(r, 1);

return NULL;

}

// Return our reply

return r;

}

/* Create a cluster node */

static redisClusterNode*

cluster_node_create(redisCluster *c, char *host, size_t host_len,

unsigned short port, unsigned short slot, short slave)

{

redisClusterNode *node = emalloc(sizeof(redisClusterNode));

// It lives in at least this slot, flag slave status

node->slot = slot;

node->slave = slave;

node->slaves = NULL;

/* Initialize our list of slot ranges */

zend_llist_init(&node->slots, sizeof(redisSlotRange), NULL, 0);

// Attach socket

node->sock = redis_sock_create(REDIS_SOCK_CLUSTER, host, host_len, port,

c->flags->timeout, c->flags->read_timeout,

c->flags->persistent, NULL, 0);

redis_sock_set_context(node->sock, c->flags->context);

redis_sock_set_auth(node->sock, c->flags->user, c->flags->pass);

return node;

}

/* Attach a slave to a master */

PHP_REDIS_API int

cluster_node_add_slave(redisClusterNode *master, redisClusterNode *slave)

{

zend_ulong index;

// Allocate our slaves hash table if we haven't yet

if (!master->slaves) {

ALLOC_HASHTABLE(master->slaves);

zend_hash_init(master->slaves, 0, NULL, ht_free_slave, 0);

index = 1;

} else {

index = master->slaves->nNextFreeElement;

}

return zend_hash_index_update_ptr(master->slaves, index, slave) != NULL;

}

/* Sanity check/validation for CLUSTER SLOTS command */

static zend_always_inline zend_bool

cluster_validate_slots_outer(const clusterReply *r)

{

return r->type == TYPE_MULTIBULK && r->elements >= 3 &&

r->element[0]->type == TYPE_INT && r->element[1]->type == TYPE_INT;

}

static zend_always_inline zend_bool

cluster_validate_slots_inner(const clusterReply *r)

{

return r->type == TYPE_MULTIBULK && r->elements >= 2 &&

r->element[0]->type == TYPE_BULK && r->element[0]->str != NULL &&

r->element[0]->len > 0 && r->element[1]->type == TYPE_INT;

}

static zend_always_inline int

cluster_validate_slot_range(size_t low, size_t high) {

return low <= high && low < REDIS_CLUSTER_SLOTS && high < REDIS_CLUSTER_SLOTS;

}

static zend_always_inline int

cluster_validate_port(size_t port) {

return port <= 65535;

}

static zend_always_inline int

cluster_validate_host(const char *host, size_t len) {

return len > 0 && len < sizeof(((redisCluster *)0)->redir_host) &&

memchr(host, '\0', len) == NULL;

}

/* Use the output of CLUSTER SLOTS to map our nodes */

static int cluster_map_slots(redisCluster *c, clusterReply *r) {

redisClusterNode *pnode, *master, *slave;

redisSlotRange range;

int i,j, klen;

size_t low, high, hlen, port;

clusterReply *r2, *r3;

char *host, key[1024];

zend_hash_clean(c->nodes);

for (i = 0; i < r->elements; i++) {

// Inner response

r2 = r->element[i];

// Validate outer and master slot structure

if (!cluster_validate_slots_outer(r2) || !cluster_validate_slots_inner(r2->element[2])) {

return -1;

}

// Master

r3 = r2->element[2];

// Grab our slot range, as well as master host/port

low = r2->element[0]->integer;

high = r2->element[1]->integer;

host = r3->element[0]->str;

hlen = r3->element[0]->len;

port = r3->element[1]->integer;

/* Ensure slot range and port are sane and within bounds */

if (!cluster_validate_slot_range(low, high) ||

!cluster_validate_port(port) ||

!cluster_validate_host(host, hlen)

) {

return -1;

}

// If the node is new, create and add to nodes. Otherwise use it.

klen = snprintf(key, sizeof(key), "%s:%zu", host, port);

if (klen < 0 || klen >= sizeof(key)) {

return -1;

}

if ((pnode = zend_hash_str_find_ptr(c->nodes, key, klen)) == NULL) {

master = cluster_node_create(c, host, hlen, port, low, 0);

zend_hash_str_update_ptr(c->nodes, key, klen, master);

// Attach slaves first time we encounter a given master in order to

// avoid registering the slaves multiple times

for (j = 3; j< r2->elements; j++) {

r3 = r2->element[j];

// Skip slaves whose host bulk is missing or empty

if (!cluster_validate_slots_inner(r3)) {

continue;

}

if (!cluster_validate_port(r3->element[1]->integer) ||

!cluster_validate_host(r3->element[0]->str, r3->element[0]->len)

) {

continue;

}

// Attach this node to our slave

slave = cluster_node_create(c, r3->element[0]->str,

r3->element[0]->len, r3->element[1]->integer, low, 1);

cluster_node_add_slave(master, slave);

}

} else {

master = pnode;

}

// Attach this node to each slot in the range

for (j = low; j<= high; j++) {

c->master[j] = master;

}

/* Append to our list of slot ranges */

range = (redisSlotRange){.low = low, .high = high};

zend_llist_add_element(&master->slots, &range);

}

// Success

return 0;

}

/* Free a redisClusterNode structure */

PHP_REDIS_API void cluster_free_node(redisClusterNode *node) {

if (node->slaves) {

zend_hash_destroy(node->slaves);

efree(node->slaves);

}

zend_llist_destroy(&node->slots);

redis_free_socket(node->sock);

efree(node);

}

/* Get or create a redisClusterNode that corresponds to the asking redirection */

static redisClusterNode *cluster_get_asking_node(redisCluster *c) {

redisClusterNode *pNode;

char key[1024];

int key_len;

/* Hashed by host:port */

key_len = snprintf(key, sizeof(key), "%s:%u", c->redir_host, c->redir_port);

/* See if we've already attached to it */

if ((pNode = zend_hash_str_find_ptr(c->nodes, key, key_len)) != NULL) {

return pNode;

}

/* This host:port is unknown to us, so add it */

pNode = cluster_node_create(c, c->redir_host, c->redir_host_len,

c->redir_port, c->redir_slot, 0);

zend_hash_str_update_ptr(c->nodes, key, key_len, pNode);

/* Return the node */

return pNode;

}

/* Get or create a node at the host:port we were asked to check, and return the

* redis_sock for it. */

static RedisSock *cluster_get_asking_sock(redisCluster *c) {

return cluster_get_asking_node(c)->sock;

}

/* Our context seeds will be a hash table with RedisSock* pointers */

static void ht_free_seed(zval *data) {

RedisSock *redis_sock = Z_PTR_P(data);

if (redis_sock) redis_free_socket(redis_sock);

}

/* Free redisClusterNode objects we've stored */

static void ht_free_node(zval *data) {

redisClusterNode *node = Z_PTR_P(data);

cluster_free_node(node);

}

/* zend_llist of slot ranges -> persistent array */

static redisSlotRange *slot_range_list_clone(zend_llist *src, size_t *count) {

redisSlotRange *dst, *range;

size_t i = 0;

*count = zend_llist_count(src);

dst = pemalloc(*count * sizeof(*dst), 1);

range = zend_llist_get_first(src);

while (range) {

memcpy(&dst[i++], range, sizeof(*range));

range = zend_llist_get_next(src);

}

return dst;

}

/* Construct a redisCluster object */

PHP_REDIS_API redisCluster *cluster_create(double timeout, double read_timeout,

int failover, int persistent)

{

redisCluster *c;

/* Actual our actual cluster structure */

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

/* Initialize flags and settings */

c->flags = ecalloc(1, sizeof(RedisSock));

c->flags->type = REDIS_SOCK_CLUSTER;

c->flags->timeout = timeout;

c->flags->read_timeout = read_timeout;

c->flags->persistent = persistent;

c->subscribed_slot = -1;

c->clusterdown = 0;

c->failover = failover;

c->err = NULL;

/* Set up our waitms based on timeout */

c->waitms = (long)(1000 * (timeout + read_timeout));

/* Allocate our seeds hash table */

ALLOC_HASHTABLE(c->seeds);

zend_hash_init(c->seeds, 0, NULL, ht_free_seed, 0);

/* Allocate our nodes HashTable */

ALLOC_HASHTABLE(c->nodes);

zend_hash_init(c->nodes, 0, NULL, ht_free_node, 0);

return c;

}

PHP_REDIS_API void

cluster_free(redisCluster *c, int free_ctx)

{

/* Disconnect from each node we're connected to */

cluster_disconnect(c, 0);

/* Free any allocated prefix */

if (c->flags->prefix) zend_string_release(c->flags->prefix);

redis_sock_free_auth(c->flags);

redis_sock_free_context(c->flags);

efree(c->flags);

/* Call hash table destructors */

zend_hash_destroy(c->seeds);

zend_hash_destroy(c->nodes);

/* Free hash tables themselves */

efree(c->seeds);

efree(c->nodes);

/* Free any error we've got */

if (c->err) zend_string_release(c->err);

if (c->cache_key) {

/* Invalidate persistent cache if the cluster has changed */

if (c->redirections) {

zend_hash_del(&EG(persistent_list), c->cache_key);

}

/* Release our hold on the cache key */

zend_string_release(c->cache_key);

}

/* Free structure itself */

if (free_ctx) efree(c);

}

/* Create a cluster slot cache structure */

PHP_REDIS_API

redisCachedCluster *cluster_cache_create(zend_string *hash, HashTable *nodes) {

redisCachedCluster *cc;

redisCachedMaster *cm;

redisClusterNode *node, *slave;

cc = pecalloc(1, sizeof(*cc), 1);