Skip to content

Commit 8443769

Browse files
michael-grundermichael-grunder
committed
Merge remote-tracking branch 'mobli/develop' into feature/mobli_ra_changes
Conflicts: redis_array.c redis_array_impl.c redis_array_impl.h
2 parents a802461 + 0ff393b commit 8443769

10 files changed

Lines changed: 184 additions & 21 deletions

‎README.markdown‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,8 @@ _**Description**_: Connects to a Redis instance.
165165
*host*: string. can be a host, or the path to a unix domain socket
166166
*port*: int, optional
167167
*timeout*: float, value in seconds (optional, default is 0 meaning unlimited)
168+
*reserved*: should be NULL if retry_interval is specified
169+
*retry_interval*: int, value in milliseconds (optional)
168170

169171
##### *Return value*
170172

@@ -177,6 +179,7 @@ $redis->connect('127.0.0.1', 6379);
177179
$redis->connect('127.0.0.1'); // port 6379 by default
178180
$redis->connect('127.0.0.1', 6379, 2.5); // 2.5 sec timeout.
179181
$redis->connect('/tmp/redis.sock'); // unix domain socket.
182+
$redis->connect('127.0.0.1', 6379, 1, NULL, 100); // 1 sec timeout, 100ms delay between reconnection attempts.
180183
~~~
181184

182185
### pconnect, popen
@@ -199,6 +202,7 @@ persistent equivalents.
199202
*port*: int, optional
200203
*timeout*: float, value in seconds (optional, default is 0 meaning unlimited)
201204
*persistent_id*: string. identity for the requested persistent connection
205+
*retry_interval*: int, value in milliseconds (optional)
202206

203207
##### *Return value*
204208

‎arrays.markdown‎

Lines changed: 62 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ There are several ways of creating Redis arrays; they can be pre-defined in red
1717

1818
#### Declaring a new array with a list of nodes
1919
<pre>
20-
$ra = new RedisArray(array("host1", "host2:63792, "host2:6380"));
20+
$ra = new RedisArray(array("host1", "host2:63792", "host2:6380"));
2121
</pre>
2222

2323

@@ -26,15 +26,27 @@ $ra = new RedisArray(array("host1", "host2:63792, "host2:6380"));
2626
function extract_key_part($k) {
2727
return substr($k, 0, 3); // hash only on first 3 characters.
2828
}
29-
$ra = new RedisArray(array("host1", "host2:63792, "host2:6380"), array("function" => "extract_key_part"));
29+
$ra = new RedisArray(array("host1", "host2:63792", "host2:6380"), array("function" => "extract_key_part"));
3030
</pre>
3131

3232
#### Defining a "previous" array when nodes are added or removed.
3333
When a new node is added to an array, phpredis needs to know about it. The old list of nodes becomes the “previous” array, and the new list of nodes is used as a main ring. Right after a node has been added, some read commands will point to the wrong nodes and will need to look up the keys in the previous ring.
3434

3535
<pre>
3636
// adding host3 to a ring containing host1 and host2. Read commands will look in the previous ring if the data is not found in the main ring.
37-
$ra = new RedisArray(array('host1', 'host2', 'host3'), array('previous' => array('host1', 'host2')));
37+
$ra = new RedisArray(array("host1", "host2", "host3"), array("previous" => array("host1", "host2")));
38+
</pre>
39+
40+
#### Specifying the "retry_interval" parameter
41+
The retry_interval is used to specify a delay in milliseconds between reconnection attempts in case the client loses connection with a server
42+
<pre>
43+
$ra = new RedisArray(array("host1", "host2:63792", "host2:6380"), array("retry_timeout" => 100)));
44+
</pre>
45+
46+
#### Specifying the "lazy_connect" parameter
47+
This option is useful when a cluster has many shards but not of them are necessarily used at one time.
48+
<pre>
49+
$ra = new RedisArray(array("host1", "host2:63792", "host2:6380"), array("lazy_connect" => true)));
3850
</pre>
3951

4052
#### Defining arrays in Redis.ini
@@ -76,6 +88,53 @@ In order to control the distribution of keys by hand, you can provide a custom f
7688

7789
For instance, instanciate a RedisArray object with `new RedisArray(array("us-host", "uk-host", "de-host"), array("distributor" => "dist"));` and write a function called "dist" that will return `2` for all the keys that should end up on the "de-host" server.
7890

91+
You may also provide an array of 2 values that will be used as follows:
92+
- The first value is the initial amount of shards in use before the resharding (the x first shards specified in the constructor)
93+
- The second value is the resharding level, or number of resharding iterations.
94+
95+
For instance, suppose you started with 4 shards as follows:
96+
<pre>
97+
0 => 0 1 2 3
98+
</pre>
99+
100+
After 1 iteration of resharding, keys will be assigned to the following servers:
101+
<pre>
102+
1 => 0 4 1 5 2 6 3 7
103+
</pre>
104+
105+
After 2 iterations, keys will be assigned to the following servers:
106+
<pre>
107+
2 => 0 8 4 12 1 9 5 13 2 10 6 14 3 11 7 15
108+
</pre>
109+
110+
After 3 iterations, keys will be assigned to the following servers:
111+
<pre>
112+
3 => 0 16 8 24 4 20 12 28 1 17 9 25 5 21 13 29 2 18 10 26 6 22 14 30 3 19 11 27 7 23 15 31
113+
</pre>
114+
115+
And so on...
116+
117+
The idea here is to be able to reshard the keys easily, without moving keys from 1 server to another.
118+
119+
The procedure to adopt is simple:
120+
121+
For each initial shard, setup a slave. For instance, for shard 1 we setup slave 5.
122+
123+
Keys will now be assigned to either shard 1 or shard 5. Once the application sees the new settings, just setup shard 5 as a master. Then, in order to reclaim memory, just cleanup keys from shard 1 that belong to shard 5 and vice-versa.
124+
125+
On the next iteration, setup a new slave 9 for shard 1 and a new slave 13 for shard 5.
126+
127+
Update the application settings, disconnect the new slaves and clean up the shards from keys that don't belong there anymore.
128+
129+
Apply the same procedure for each resharding iteration.
130+
131+
### Example
132+
<pre>
133+
$ra = new RedisArray(array("host1", "host2", "host3", "host4", "host5", "host6", "host7", "host8"), array("distributor" => array(2, 2)));
134+
</pre>
135+
136+
This declares that we started with 2 shards and moved to 4 then 8 shards. The number of initial shards is 2 and the resharding level (or number of iterations) is 2.
137+
79138
## Migrating keys
80139

81140
When a node is added or removed from a ring, RedisArray instances must be instanciated with a “previous” list of nodes. A single call to `$ra->_rehash()` causes all the keys to be redistributed according to the new list of nodes. Passing a callback function to `_rehash()` makes it possible to track the progress of that operation: the function is called with a node name and a number of keys that will be examined, e.g. `_rehash(function ($host, $count){ ... });`.

‎common.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,7 @@ typedef struct {
182182

183183
char *err;
184184
int err_len;
185+
zend_bool lazy_connect;
185186
} RedisSock;
186187
/* }}} */
187188

‎library.c‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -970,7 +970,8 @@ PHPAPI void redis_ping_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_s
970970
*/
971971
PHPAPI RedisSock* redis_sock_create(char *host, int host_len, unsigned short port,
972972
double timeout, int persistent, char *persistent_id,
973-
long retry_interval)
973+
long retry_interval,
974+
zend_bool lazy_connect)
974975
{
975976
RedisSock *redis_sock;
976977

@@ -982,6 +983,7 @@ PHPAPI RedisSock* redis_sock_create(char *host, int host_len, unsigned short por
982983
redis_sock->dbNumber = 0;
983984
redis_sock->retry_interval = retry_interval * 1000;
984985
redis_sock->persistent = persistent;
986+
redis_sock->lazy_connect = lazy_connect;
985987

986988
if(persistent_id) {
987989
size_t persistent_id_len = strlen(persistent_id);

‎library.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ PHPAPI void redis_string_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis
2020
PHPAPI void redis_ping_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock, zval *z_tab, void *ctx);
2121
PHPAPI void redis_info_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock, zval *z_tab, void *ctx);
2222
PHPAPI void redis_type_response(INTERNAL_FUNCTION_PARAMETERS, RedisSock *redis_sock, zval *z_tab, void *ctx);
23-
PHPAPI RedisSock* redis_sock_create(char *host, int host_len, unsigned short port, double timeout, int persistent, char *persistent_id, long retry_interval);
23+
PHPAPI RedisSock* redis_sock_create(char *host, int host_len, unsigned short port, double timeout, int persistent, char *persistent_id, long retry_interval, zend_bool lazy_connect);
2424
PHPAPI int redis_sock_connect(RedisSock *redis_sock TSRMLS_DC);
2525
PHPAPI int redis_sock_server_open(RedisSock *redis_sock, int force_connect TSRMLS_DC);
2626
PHPAPI int redis_sock_disconnect(RedisSock *redis_sock TSRMLS_DC);

‎redis.c‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -394,6 +394,13 @@ PHPAPI int redis_sock_get(zval *id, RedisSock **redis_sock TSRMLS_DC, int no_thr
394394
}
395395
return -1;
396396
}
397+
if ((*redis_sock)->lazy_connect)
398+
{
399+
(*redis_sock)->lazy_connect = 0;
400+
if (redis_sock_server_open(*redis_sock, 1 TSRMLS_CC) < 0) {
401+
return -1;
402+
}
403+
}
397404

398405
return Z_LVAL_PP(socket);
399406
}
@@ -640,7 +647,7 @@ PHPAPI int redis_connect(INTERNAL_FUNCTION_PARAMETERS, int persistent) {
640647
zend_clear_exception(TSRMLS_C); /* clear exception triggered by non-existent socket during connect(). */
641648
}
642649

643-
redis_sock = redis_sock_create(host, host_len, port, timeout, persistent, persistent_id, retry_interval);
650+
redis_sock = redis_sock_create(host, host_len, port, timeout, persistent, persistent_id, retry_interval, 0);
644651

645652
if (redis_sock_server_open(redis_sock, 1 TSRMLS_CC) < 0) {
646653
redis_free_socket(redis_sock);

‎redis_array.c‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -197,6 +197,7 @@ PHP_METHOD(RedisArray, __construct)
197197
zend_bool b_index = 0, b_autorehash = 0, b_pconnect = 0;
198198
HashTable *hPrev = NULL, *hOpts = NULL;
199199
long l_retry_interval = 0;
200+
zend_bool b_lazy_connect = 0;
200201

201202
if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "z|a", &z0, &z_opts) == FAILURE) {
202203
RETURN_FALSE;
@@ -255,6 +256,11 @@ PHP_METHOD(RedisArray, __construct)
255256
}
256257
}
257258
}
259+
260+
/* extract lazy connect option. */
261+
if(FAILURE != zend_hash_find(hOpts, "lazy_connect", sizeof("lazy_connect"), (void**)&zpData) && Z_TYPE_PP(zpData) == IS_BOOL) {
262+
b_lazy_connect = Z_BVAL_PP(zpData);
263+
}
258264
}
259265

260266
/* extract either name of list of hosts from z0 */

‎redis_array_impl.c‎

Lines changed: 95 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ extern int le_redis_sock;
3030
extern zend_class_entry *redis_ce;
3131

3232
RedisArray*
33-
ra_load_hosts(RedisArray *ra, HashTable *hosts, long retry_interval TSRMLS_DC)
33+
ra_load_hosts(RedisArray *ra, HashTable *hosts, long retry_interval, zend_bool lazy_connect TSRMLS_DC)
3434
{
3535
int i, host_len, id;
3636
int count = zend_hash_num_elements(hosts);
@@ -70,10 +70,13 @@ ra_load_hosts(RedisArray *ra, HashTable *hosts, long retry_interval TSRMLS_DC)
7070
call_user_function(&redis_ce->function_table, &ra->redis[i], &z_cons, &z_ret, 0, NULL TSRMLS_CC);
7171

7272
/* create socket */
73-
redis_sock = redis_sock_create(host, host_len, port, 0, ra->pconnect, NULL, retry_interval);
73+
redis_sock = redis_sock_create(host, host_len, port, 0, ra->pconnect, NULL, retry_interval, lazy_connect);
7474

75-
/* connect */
76-
redis_sock_server_open(redis_sock, 1 TSRMLS_CC);
75+
if (!lazy_connect)
76+
{
77+
/* connect */
78+
redis_sock_server_open(redis_sock, 1 TSRMLS_CC);
79+
}
7780

7881
/* attach */
7982
#if PHP_VERSION_ID >= 50400
@@ -163,10 +166,12 @@ RedisArray *ra_load_array(const char *name TSRMLS_DC) {
163166
zval *z_params_autorehash;
164167
zval *z_params_retry_interval;
165168
zval *z_params_pconnect;
169+
zval *z_params_lazy_connect;
166170
RedisArray *ra = NULL;
167171

168172
zend_bool b_index = 0, b_autorehash = 0, b_pconnect = 0;
169173
long l_retry_interval = 0;
174+
zend_bool b_lazy_connect = 0;
170175
HashTable *hHosts = NULL, *hPrev = NULL;
171176

172177
/* find entry */
@@ -253,9 +258,18 @@ RedisArray *ra_load_array(const char *name TSRMLS_DC) {
253258
b_pconnect = 1;
254259
}
255260
}
261+
/* find retry interval option */
262+
MAKE_STD_ZVAL(z_params_lazy_connect);
263+
array_init(z_params_lazy_connect);
264+
sapi_module.treat_data(PARSE_STRING, estrdup(INI_STR("redis.arrays.lazyconnect")), z_params_lazy_connect TSRMLS_CC);
265+
if (zend_hash_find(Z_ARRVAL_P(z_params_lazy_connect), name, strlen(name) + 1, (void **) &z_data_pp) != FAILURE) {
266+
if(Z_TYPE_PP(z_data_pp) == IS_STRING && strncmp(Z_STRVAL_PP(z_data_pp), "1", 1) == 0) {
267+
b_lazy_connect = 1;
268+
}
269+
}
256270

257271
/* create RedisArray object */
258-
ra = ra_make_array(hHosts, z_fun, z_dist, hPrev, b_index, l_retry_interval, b_pconnect TSRMLS_CC);
272+
ra = ra_make_array(hHosts, z_fun, z_dist, hPrev, b_index, l_retry_interval, b_pconnect, b_lazy_connect TSRMLS_CC);
259273
ra->auto_rehash = b_autorehash;
260274

261275
/* cleanup */
@@ -273,12 +287,14 @@ RedisArray *ra_load_array(const char *name TSRMLS_DC) {
273287
efree(z_params_retry_interval);
274288
zval_dtor(z_params_pconnect);
275289
efree(z_params_pconnect);
290+
zval_dtor(z_params_lazy_connect);
291+
efree(z_params_lazy_connect);
276292

277293
return ra;
278294
}
279295

280296
RedisArray *
281-
ra_make_array(HashTable *hosts, zval *z_fun, zval *z_dist, HashTable *hosts_prev, zend_bool b_index, long retry_interval, zend_bool b_pconnect TSRMLS_DC) {
297+
ra_make_array(HashTable *hosts, zval *z_fun, zval *z_dist, HashTable *hosts_prev, zend_bool b_index, long retry_interval, zend_bool b_pconnect, zend_bool b_lazy_connect TSRMLS_DC) {
282298

283299
int count = zend_hash_num_elements(hosts);
284300

@@ -296,10 +312,10 @@ ra_make_array(HashTable *hosts, zval *z_fun, zval *z_dist, HashTable *hosts_prev
296312
/* init array data structures */
297313
ra_init_function_table(ra);
298314

299-
if(NULL == ra_load_hosts(ra, hosts, retry_interval TSRMLS_CC)) {
315+
if(NULL == ra_load_hosts(ra, hosts, retry_interval, lazy_connect TSRMLS_CC)) {
300316
return NULL;
301317
}
302-
ra->prev = hosts_prev ? ra_make_array(hosts_prev, z_fun, z_dist, NULL, b_index, retry_interval, b_pconnect TSRMLS_CC) : NULL;
318+
ra->prev = hosts_prev ? ra_make_array(hosts_prev, z_fun, z_dist, NULL, b_index, retry_interval, b_pconnect, b_lazy_connect TSRMLS_CC) : NULL;
303319

304320
/* copy function if provided */
305321
if(z_fun) {
@@ -410,20 +426,89 @@ ra_call_distributor(RedisArray *ra, const char *key, int key_len, int *pos TSRML
410426
return 1;
411427
}
412428

429+
zend_bool
430+
ra_check_distributor(RedisArray *ra, int *num_original, int *num_reshards TSRMLS_DC) {
431+
if(Z_TYPE_P(ra->z_dist) == IS_ARRAY) {
432+
zval **z_original_pp;
433+
zval **z_reshards_pp;
434+
HashTable *shards = Z_ARRVAL_P(ra->z_dist);
435+
if (zend_hash_num_elements(shards) != 2) {
436+
return 0;
437+
}
438+
if (zend_hash_index_find(shards, 0, (void **)&z_original_pp) != SUCCESS ||
439+
zend_hash_index_find(shards, 1, (void **)&z_reshards_pp) != SUCCESS) {
440+
return 0;
441+
}
442+
if (Z_TYPE_PP(z_original_pp) == IS_LONG) {
443+
if ((*num_original = Z_LVAL_PP(z_original_pp)) == 0) {
444+
return 0;
445+
}
446+
}
447+
else if (Z_TYPE_PP(z_original_pp) == IS_STRING) {
448+
if ((*num_original = atol(Z_STRVAL_PP(z_original_pp))) == 0) {
449+
return 0;
450+
}
451+
}
452+
else {
453+
return 0;
454+
}
455+
if (Z_TYPE_PP(z_reshards_pp) == IS_LONG) {
456+
if ((*num_reshards = Z_LVAL_PP(z_reshards_pp)) == 0) {
457+
return 0;
458+
}
459+
}
460+
else if (Z_TYPE_PP(z_reshards_pp) == IS_STRING) {
461+
if ((*num_reshards = atol(Z_STRVAL_PP(z_reshards_pp))) == 0) {
462+
return 0;
463+
}
464+
}
465+
else {
466+
return 0;
467+
}
468+
return 1;
469+
}
470+
return 0;
471+
}
472+
413473
zval *
414474
ra_find_node(RedisArray *ra, const char *key, int key_len, int *out_pos TSRMLS_DC) {
415475

416476
uint32_t hash;
417477
char *out;
418-
int pos, out_len;
478+
int pos = 0, out_len;
419479

420480
/* extract relevant part of the key */
421481
out = ra_extract_key(ra, key, key_len, &out_len TSRMLS_CC);
422482
if(!out)
423483
return NULL;
424484

425485
if(ra->z_dist) {
426-
if (!ra_call_distributor(ra, key, key_len, &pos TSRMLS_CC)) {
486+
char *error = NULL;
487+
int num_original, num_reshards;
488+
if (ra_check_distributor(ra, &num_original, &num_reshards TSRMLS_CC)) {
489+
if (num_reshards < 1 || ra->count != (num_original * (1 << num_reshards))) {
490+
return NULL;
491+
}
492+
/* Calculate original hash */
493+
hash = rcrc32(out, out_len);
494+
efree(out);
495+
uint64_t h64 = hash;
496+
h64 *= num_original;
497+
h64 /= 0xffffffff;
498+
pos = (int)h64;
499+
int i;
500+
/* Infer the new position */
501+
for(i = 0; i < num_reshards; i++) {
502+
int total = num_original * 2;
503+
h64 = hash;
504+
h64 *= total;
505+
h64 /= 0xffffffff;
506+
h64 %= 2;
507+
pos = pos + h64 * num_original;
508+
num_original = total;
509+
}
510+
}
511+
else if (!ra_call_distributor(ra, key, key_len, &pos TSRMLS_CC)) {
427512
return NULL;
428513
}
429514
}
@@ -439,7 +524,6 @@ ra_find_node(RedisArray *ra, const char *key, int key_len, int *out_pos TSRMLS_D
439524
pos = (int)h64;
440525
}
441526
if(out_pos) *out_pos = pos;
442-
443527
return ra->redis[pos];
444528
}
445529

‎redis_array_impl.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,9 @@
55
#include "common.h"
66
#include "redis_array.h"
77

8-
RedisArray *ra_load_hosts(RedisArray *ra, HashTable *hosts, long retry_interval TSRMLS_DC);
8+
RedisArray *ra_load_hosts(RedisArray *ra, HashTable *hosts, long retry_interval, zend_bool lazy_connect TSRMLS_DC);
99
RedisArray *ra_load_array(const char *name TSRMLS_DC);
10-
RedisArray *ra_make_array(HashTable *hosts, zval *z_fun, zval *z_dist, HashTable *hosts_prev, zend_bool b_index, long retry_interval, zend_bool b_pconnect TSRMLS_DC);
10+
RedisArray *ra_make_array(HashTable *hosts, zval *z_fun, zval *z_dist, HashTable *hosts_prev, zend_bool b_index, long retry_interval, zend_bool b_pconnect, zend_bool b_lazy_connect TSRMLS_DC);
1111
zval *ra_find_node_by_name(RedisArray *ra, const char *host, int host_len TSRMLS_DC);
1212
zval *ra_find_node(RedisArray *ra, const char *key, int key_len, int *out_pos TSRMLS_DC);
1313
void ra_init_function_table(RedisArray *ra);

0 commit comments

Comments
 (0)