#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>

extern zend_class_entry *redis_cluster_exception_ce;

/* Some debug methods that will go away when we're through with them */

static void cluster_dump_nodes(redisCluster *c) {

redisClusterNode **pp, *p;

for(zend_hash_internal_pointer_reset(c->nodes);

zend_hash_has_more_elements(c->nodes)==SUCCESS;

zend_hash_move_forward(c->nodes))

{

zend_hash_get_current_data(c->nodes, (void**)&pp);

p = *pp;

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");

}

}

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_str buf = {0};

int i;

switch(reply->type) {

case TYPE_ERR:

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

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

break;

case TYPE_LINE:

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

break;

case TYPE_INT:

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

smart_str_append_long(&buf, reply->integer);

break;

case TYPE_BULK:

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

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

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

break;

case TYPE_MULTIBULK:

if(reply->elements == (size_t)-1) {

smart_str_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_str_0(&buf);

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

php_printf("\n");

efree(buf.c);

}

}

/* 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) {

int 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:

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 void

cluster_multibulk_resp_recursive(RedisSock *sock, size_t elements,

clusterReply **element, int *err TSRMLS_DC)

{

size_t idx = 0;

clusterReply *r;

int len;

char buf[1024];

while(elements-- > 0) {

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

r = element[idx];

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

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

*err = 1;

return;

}

/* Set our reply len */

r->len = len;

switch(r->type) {

case TYPE_ERR:

case TYPE_LINE:

if(redis_sock_gets(sock,buf,sizeof(buf),&r->len TSRMLS_CC)<0) {

*err = 1;

return;

}

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 TSRMLS_CC);

if(!r->str) {

*err = 1;

return;

}

}

break;

case TYPE_MULTIBULK:

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

r->elements = r->len;

cluster_multibulk_resp_recursive(sock, r->elements, r->element,

err TSRMLS_CC);

if(*err) return;

break;

default:

*err = 1;

return;

}

idx++;

}

}

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

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

ulong slaveidx)

{

redisClusterNode **node;

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

if (slaveidx == 0) {

return SLOT_SOCK(c, slot);

}

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

if (!SLOT_SLAVES(c, slot) || zend_hash_index_find(SLOT_SLAVES(c,slot),

slaveidx, (void**)&node)==FAILURE) return NULL;

/* Success, return the slave */

return (*node)->sock;

}

/* Read the response from a cluster */

clusterReply *cluster_read_resp(redisCluster *c TSRMLS_DC) {

return cluster_read_sock_resp(c->cmd_sock,c->reply_type,c->reply_len TSRMLS_CC);

}

/* 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,

size_t len TSRMLS_DC)

{

clusterReply *r;

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

r->type = type;

// Error flag in case we go recursive

int err = 0;

switch(r->type) {

case TYPE_INT:

r->integer = len;

break;

case TYPE_LINE:

case TYPE_ERR:

return r;

case TYPE_BULK:

r->len = len;

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

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

cluster_free_reply(r, 1);

return NULL;

}

break;

case TYPE_MULTIBULK:

r->elements = len;

if(len != (size_t)-1) {

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

cluster_multibulk_resp_recursive(redis_sock, len, r->element,

&err TSRMLS_CC);

}

break;

default:

cluster_free_reply(r,1);

return NULL;

}

// Free/return null on communication error

if(err) {

cluster_free_reply(r,1);

return NULL;

}

// Success, return the reply

return r;

}

/*

* 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, char *cmd, int cmd_len,

REDIS_REPLY_TYPE type TSRMLS_DC)

{

char buf[1024];

/* Connect to the socket if we aren't yet */

CLUSTER_LAZY_CONNECT(redis_sock);

/* 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) ||

!php_stream_gets(redis_sock->stream, buf, sizeof(buf))) return -1;

/* Success! */

return 0;

}

static int cluster_send_asking(RedisSock *redis_sock TSRMLS_DC) {

return cluster_send_direct(redis_sock, RESP_ASKING_CMD,

sizeof(RESP_ASKING_CMD)-1, TYPE_LINE TSRMLS_CC);

}

/* 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 TSRMLS_DC) {

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, RESP_READONLY_CMD,

sizeof(RESP_READONLY_CMD)-1, TYPE_LINE TSRMLS_CC);

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

redis_sock->readonly = !ret;

/* Return the result of our send */

return ret;

}

/* Send MULTI to a specific ReidsSock */

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

if (cluster_send_direct(SLOT_SOCK(c,slot), RESP_MULTI_CMD,

sizeof(RESP_MULTI_CMD)-1, TYPE_LINE TSRMLS_CC)==0)

{

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 TSRMLS_DC) {

int retval;

/* Send exec */

retval = cluster_send_slot(c, slot, RESP_EXEC_CMD, sizeof(RESP_EXEC_CMD)-1,

TYPE_MULTIBULK TSRMLS_CC);

/* 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 TSRMLS_DC) {

if (cluster_send_direct(SLOT_SOCK(c,slot), RESP_DISCARD_CMD,

sizeof(RESP_DISCARD_CMD)-1, TYPE_LINE TSRMLS_CC))

{

return 0;

}

return -1;

}

/*

* Cluster key distribution helpers. For a small handlful of commands, we want

* to distribute them across 1-N nodes. These methods provide simple containers

* for the purposes of splitting keys/values in this way

* */

/* Free cluster distribution list inside a HashTable */

static void cluster_dist_free_ht(void *p) {

clusterDistList *dl = *(clusterDistList**)p;

int i;

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

if(dl->entry[i].key_free)

efree(dl->entry[i].key);

if(dl->entry[i].val_free)

efree(dl->entry[i].val);

}

efree(dl->entry);

efree(dl);

}

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

HashTable *cluster_dist_create() {

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() {

clusterDistList *dl;

dl = emalloc(sizeof(clusterDistList));

dl->entry = emalloc(CLUSTER_KEYDIST_ALLOC * sizeof(clusterKeyVal));

dl->size = CLUSTER_KEYDIST_ALLOC;

dl->len = 0;

return dl;

}

/* Add a key to a dist list, returning the keval entry */

static clusterKeyVal *cluster_dl_add_key(clusterDistList *dl, char *key,

int key_len, int key_free)

{

// Reallocate if required

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

dl->entry = erealloc(dl->entry, sizeof(clusterKeyVal) * dl->size * 2);

dl->size *= 2;

}

// Set key info

dl->entry[dl->len].key = key;

dl->entry[dl->len].key_len = key_len;

dl->entry[dl->len].key_free = key_free;

// NULL out any values

dl->entry[dl->len].val = NULL;

dl->entry[dl->len].val_len = 0;

dl->entry[dl->len].val_free = 0;

return &(dl->entry[dl->len++]);

}

/* Add a key, returning a pointer to the entry where passed for easy adding

* of values to match this key */

int cluster_dist_add_key(redisCluster *c, HashTable *ht, char *key,

int key_len, clusterKeyVal **kv)

{

int key_free;

short slot;

clusterDistList **ppdl, *dl;

clusterKeyVal *retptr;

// Prefix our key and hash it

key_free = redis_key_prefix(c->flags, &key, &key_len);

slot = cluster_hash_key(key, key_len);

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

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

if(key_free) efree(key);

return FAILURE;

}

// Look for this slot

if(zend_hash_index_find(ht, (ulong)slot, (void**)&ppdl)==FAILURE) {

dl = cluster_dl_create();

zend_hash_index_update(ht, (ulong)slot, (void**)&dl,

sizeof(clusterDistList*), NULL);

} else {

dl = *ppdl;

}

// Now actually add this key

retptr = cluster_dl_add_key(dl, key, key_len, key_free);

// Push our return pointer if requested

if(kv) *kv = retptr;

return SUCCESS;

}

/* Provided a clusterKeyVal, add a value */

void cluster_dist_add_val(redisCluster *c, clusterKeyVal *kv, zval *z_val

TSRMLS_DC)

{

char *val;

int val_len, val_free;

// Serialize our value

val_free = redis_serialize(c->flags, z_val, &val, &val_len TSRMLS_CC);

// Attach it to the provied keyval entry

kv->val = val;

kv->val_len = val_len;

kv->val_free = val_free;

}

/* Free allocated memory for a clusterMultiCmd */

void cluster_multi_free(clusterMultiCmd *mc) {

efree(mc->cmd.c);

efree(mc->args.c);

}

/* Add an argument to a clusterMultiCmd */

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

mc->argc++;

redis_cmd_append_sstr(&(mc->args), data, data_len);

}

/* Finalize a clusterMutliCmd by constructing the whole thing */

void cluster_multi_fini(clusterMultiCmd *mc) {

mc->cmd.len = 0;

redis_cmd_init_sstr(&(mc->cmd), mc->argc, mc->kw, mc->kw_len);

smart_str_appendl(&(mc->cmd), mc->args.c, mc->args.len);

}

/* Set our last error string encountered */

static void cluster_set_err(redisCluster *c, char *err, int err_len)

{

if(err && err_len>0) {

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

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

{

c->clusterdown = 1;

return;

}

if(c->err == NULL) {

c->err = emalloc(err_len+1);

} else if(err_len > c->err_len) {

c->err = erealloc(c->err, err_len + 1);

}

memcpy(c->err,err,err_len);

c->err[err_len]='\0';

c->err_len = err_len;

} else {

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

c->err = NULL;

c->err_len = 0;

}

}

/* Destructor for slaves */

static void ht_free_slave(void *data) {

if(*(redisClusterNode**)data) {

cluster_free_node(*(redisClusterNode**)data);

}

}

/* 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 betweeen { and }

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

}

/* 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),"%ld",Z_LVAL_P(z_key));

kptr = (const char *)buf;

break;

case IS_DOUBLE:

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

kptr = (const char *)buf;

break;

case IS_ARRAY:

kptr = "Array";

klen = sizeof("Array")-1;

break;

case IS_OBJECT:

kptr = "Object";

klen = sizeof("Object")-1;

break;

}

// Hash the string representation

return cluster_hash_key(kptr, klen);

}

static char **split_str_by_delim(char *str, char *delim, int *len) {

char **array, *tok, *tok_buf;

int size=16;

*len = 0;

// Initial storage

array = emalloc(size * sizeof(char*));

tok = php_strtok_r(str, delim, &tok_buf);

while(tok) {

if(size == *len) {

size *= 2;

array = erealloc(array, size * sizeof(char*));

}

array[*len] = tok;

(*len)++;

tok = php_strtok_r(NULL, delim, &tok_buf);

}

return array;

}

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

* reply or NULL on failure. */

clusterReply* cluster_get_slots(RedisSock *redis_sock TSRMLS_DC)

{

clusterReply *r;

REDIS_REPLY_TYPE type;

int 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 TSRMLS_CC)<0 ||

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

{

return NULL;

}

// Consume the rest of our response

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

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

{

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;

// Attach socket

node->sock = redis_sock_create(host, host_len, port, c->timeout,

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

return node;

}

/* Attach a slave to a master */

PHP_REDIS_API int

cluster_node_add_slave(redisClusterNode *master, redisClusterNode *slave)

{

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(master->slaves, index, (void*)&slave,

sizeof(redisClusterNode*), NULL) != SUCCESS;

}

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

#define VALIDATE_SLOTS_OUTER(r) \

(r->elements>=3 && r2->element[0]->type == TYPE_INT && \

r->element[1]->type==TYPE_INT)

#define VALIDATE_SLOTS_INNER(r) \

(r->type == TYPE_MULTIBULK && r->elements>=2 && \

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

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

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

int i,j, hlen, klen;

short low, high;

clusterReply *r2, *r3;

redisClusterNode **ppnode, *master, *slave;

unsigned short port;

char *host, key[1024];

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

// Inner response

r2 = r->element[i];

// Validate outer and master slot

if(!VALIDATE_SLOTS_OUTER(r2) || !VALIDATE_SLOTS_INNER(r2->element[2])) {

return -1;

}

// Master

r3 = r2->element[2];

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

low = (unsigned short)r2->element[0]->integer;

high = (unsigned short)r2->element[1]->integer;

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

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

port = (unsigned short)r3->element[1]->integer;

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

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

if(zend_hash_find(c->nodes,key,klen+1,(void**)&ppnode)==FAILURE) {

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

zend_hash_update(c->nodes, key, klen+1, (void*)&master,

sizeof(redisClusterNode*), NULL);

} else {

master = *ppnode;

}

// Attach slaves

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

r3 = r2->element[j];

if(!VALIDATE_SLOTS_INNER(r3)) {

return -1;

}

// Skip slaves where the host is ""

if(r3->element[0]->len == 0) continue;

// Attach this node to our slave

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

(int)r3->element[0]->len,

(unsigned short)r3->element[1]->integer, low, 1);

cluster_node_add_slave(master, slave);

}

// Attach this node to each slot in the range

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

c->master[j]=master;

}

}

// 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);

}

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 TSRMLS_DC) {

redisClusterNode **ppNode;

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 (zend_hash_find(c->nodes, key, key_len+1, (void**)&ppNode) == SUCCESS) {

return *ppNode;

}

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

*ppNode = cluster_node_create(c, c->redir_host, c->redir_host_len,

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

/* Return the node */

return *ppNode;

}

/* 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 TSRMLS_DC) {

return cluster_get_asking_node(c TSRMLS_CC)->sock;

}

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

static void ht_free_seed(void *data) {

RedisSock *redis_sock = *(RedisSock**)data;

if(redis_sock) redis_free_socket(redis_sock);

}

/* Free redisClusterNode objects we've stored */

static void ht_free_node(void *data) {

redisClusterNode *node = *(redisClusterNode**)data;

cluster_free_node(node);

}

/* 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->subscribed_slot = -1;

c->clusterdown = 0;

c->timeout = timeout;

c->read_timeout = read_timeout;

c->failover = failover;

c->persistent = persistent;

/* Set up our waitms based on timeout */

c->waitms = (long)(1000 * 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) {

/* Free any allocated prefix */

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

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) efree(c->err);

/* Free structure itself */

efree(c);

}

/* Initialize seeds */

PHP_REDIS_API int

cluster_init_seeds(redisCluster *cluster, HashTable *ht_seeds) {

RedisSock *redis_sock;

char *str, *psep, key[1024];

int key_len;

zval **z_seed;

// Iterate our seeds array

for(zend_hash_internal_pointer_reset(ht_seeds);

zend_hash_has_more_elements(ht_seeds)==SUCCESS;

zend_hash_move_forward(ht_seeds))

{

// Grab seed string

zend_hash_get_current_data(ht_seeds, (void**)&z_seed);

// Skip anything that isn't a string

if(Z_TYPE_PP(z_seed)!=IS_STRING)

continue;

// Grab a copy of the string

str = Z_STRVAL_PP(z_seed);

// Must be in host:port form

if(!(psep = strchr(str, ':')))

continue;

// Allocate a structure for this seed

redis_sock = redis_sock_create(str, psep-str,

(unsigned short)atoi(psep+1), cluster->timeout,

cluster->persistent, NULL, 0, 0);

// Index this seed by host/port

key_len = snprintf(key, sizeof(key), "%s:%u", redis_sock->host,

redis_sock->port);

// Add to our seed HashTable

zend_hash_update(cluster->seeds, key, key_len+1, (void*)&redis_sock,

sizeof(RedisSock*),NULL);

}

// Success if at least one seed seems valid

return zend_hash_num_elements(cluster->seeds) > 0 ? 0 : -1;

}

/* Initial mapping of our cluster keyspace */

PHP_REDIS_API int

cluster_map_keyspace(redisCluster *c TSRMLS_DC) {

RedisSock **seed;

clusterReply *slots=NULL;

int mapped=0;

// Iterate over seeds until we can get slots

for(zend_hash_internal_pointer_reset(c->seeds);

!mapped && zend_hash_has_more_elements(c->seeds) == SUCCESS;

zend_hash_move_forward(c->seeds))

{

// Grab the redis_sock for this seed

zend_hash_get_current_data(c->seeds, (void**)&seed);

// Attempt to connect to this seed node

if(redis_sock_connect(*seed TSRMLS_CC)!=0) {

continue;

}

// Parse out cluster nodes. Flag mapped if we are valid

slots = cluster_get_slots(*seed TSRMLS_CC);

if(slots) mapped = !cluster_map_slots(c, slots);

// Bin anything mapped, if we failed somewhere

if(!mapped && slots) {

memset(c->master, 0, sizeof(redisClusterNode*)*REDIS_CLUSTER_SLOTS);

}

}

// Clean up slots reply if we got one

if(slots) cluster_free_reply(slots, 1);

// Throw an exception if we couldn't map

if(!mapped) {

zend_throw_exception(redis_cluster_exception_ce,

"Couldn't map cluster keyspace using any provided seed", 0

TSRMLS_CC);

return -1;

}

return 0;

}

/* Parse the MOVED OR ASK redirection payload when we get such a response

* and apply this information to our cluster. If we encounter a parse error

* nothing in the cluster will be modified, and -1 is returned. */

static int cluster_set_redirection(redisCluster* c, char *msg, int moved)

{

char *host, *port;

/* Move past "MOVED" or "ASK */

msg += moved ? MOVED_LEN : ASK_LEN;

// We need a slot seperator

if(!(host = strchr(msg, ' '))) return -1;

*host++ = '\0';

// We need a : that seperates host from port

if(!(port = strchr(host,':'))) return -1;

*port++ = '\0';

// Success, apply it

c->redir_type = moved ? REDIR_MOVED : REDIR_ASK;

strncpy(c->redir_host, host, sizeof(c->redir_host));

c->redir_host_len = port - host - 1;

c->redir_slot = (unsigned short)atoi(msg);

c->redir_port = (unsigned short)atoi(port);

return 0;

}

/* Once we write a command to a node in our cluster, this function will check

* the reply type and extract information from those that will specify a length

* bit. If we encounter an error condition, we'll check for MOVED or ASK

* redirection, parsing out slot host and port so the caller can take

* appropriate action.

*

* In the case of a non MOVED/ASK error, we wlll set our cluster error

* condition so GetLastError can be queried by the client.

*

* This function will return -1 on a critical error (e.g. parse/communication

* error, 0 if no redirection was encountered, and 1 if the data was moved. */

static int cluster_check_response(redisCluster *c, REDIS_REPLY_TYPE *reply_type

TSRMLS_DC)

{

size_t sz;

// Clear out any prior error state and our last line response

CLUSTER_CLEAR_ERROR(c);

CLUSTER_CLEAR_REPLY(c);

if(-1 == redis_check_eof(c->cmd_sock, 1 TSRMLS_CC) ||

EOF == (*reply_type = php_stream_getc(c->cmd_sock->stream)))

{

return -1;

}

// In the event of an ERROR, check if it's a MOVED/ASK error

if(*reply_type == TYPE_ERR) {

char inbuf[1024];

int moved;

// Attempt to read the error