Add support for exponential backoff on retry

nbraun-amazon · nbraun-amazon · commit 18706d92711c · 2021-07-20T09:21:13.000Z
diff --git a/backoff.c b/backoff.c
@@ -0,0 +1,90 @@
+#include "common.h"
+
+#include <ext/standard/php_rand.h>
+
+#if PHP_VERSION_ID >= 70100
+#include <ext/standard/php_mt_rand.h>
+#else
+static zend_long php_mt_rand_range(zend_long min, zend_long max) {
+    zend_long number = php_rand();
+    RAND_RANGE(number, min, max, PHP_RAND_MAX);
+    return number;
+}
+#endif
+
+#include "backoff.h"
+
+static zend_ulong random_range(zend_ulong min, zend_ulong max) {
+    if (max < min) {
+        return php_mt_rand_range(max, min);
+    }
+
+    return php_mt_rand_range(min, max);
+}
+
+static zend_ulong redis_default_backoff(struct RedisBackoff *self, unsigned int retry_index) {
+    zend_ulong backoff = retry_index ? self->base : random_range(0, self->base);
+    return MIN(self->cap, backoff);
+}
+
+static zend_ulong redis_constant_backoff(struct RedisBackoff *self, unsigned int retry_index) {
+    zend_ulong backoff = self->base;
+    return MIN(self->cap, backoff);
+}
+
+static zend_ulong redis_uniform_backoff(struct RedisBackoff *self, unsigned int retry_index) {
+    zend_ulong backoff = random_range(0, self->base);
+    return MIN(self->cap, backoff);
+}
+
+static zend_ulong redis_exponential_backoff(struct RedisBackoff *self, unsigned int retry_index) {
+    zend_ulong pow = MIN(retry_index, 10);
+    zend_ulong backoff = self->base * (1 << pow);
+    return MIN(self->cap, backoff);
+}
+
+static zend_ulong redis_full_jitter_backoff(struct RedisBackoff *self, unsigned int retry_index) {
+    zend_ulong pow = MIN(retry_index, 10);
+    zend_ulong backoff = self->base * (1 << pow);
+    zend_ulong cap = MIN(self->cap, backoff);
+    return random_range(0, self->cap);
+}
+
+static zend_ulong redis_equal_jitter_backoff(struct RedisBackoff *self, unsigned int retry_index) {
+    zend_ulong pow = MIN(retry_index, 10);
+    zend_ulong backoff = self->base * (1 << pow);
+    zend_ulong temp = MIN(self->cap, backoff);
+    return temp / 2 + random_range(0, temp) / 2;
+}
+
+static zend_ulong redis_decorrelated_jitter_backoff(struct RedisBackoff *self, unsigned int retry_index) {
+    self->previous_backoff = random_range(self->base, self->previous_backoff * 3);
+    return MIN(self->cap, self->previous_backoff);
+}
+
+typedef zend_ulong (*redis_backoff_algorithm)(struct RedisBackoff *self, unsigned int retry_index);
+
+static redis_backoff_algorithm redis_backoff_algorithms[REDIS_BACKOFF_ALGORITHMS] = {
+    redis_default_backoff,
+    redis_decorrelated_jitter_backoff,
+    redis_full_jitter_backoff,
+    redis_equal_jitter_backoff,
+    redis_exponential_backoff,
+    redis_uniform_backoff,
+    redis_constant_backoff,
+};
+
+void redis_initialize_backoff(struct RedisBackoff *self, unsigned long retry_interval) {
+    self->algorithm = 0; // default backoff
+    self->base = retry_interval;
+    self->cap = retry_interval;
+    self->previous_backoff = 0;
+}
+
+void redis_backoff_reset(struct RedisBackoff *self) {
+    self->previous_backoff = 0;
+}
+
+zend_ulong redis_backoff_compute(struct RedisBackoff *self, unsigned int retry_index) {
+    return redis_backoff_algorithms[self->algorithm](self, retry_index);
+}
diff --git a/backoff.h b/backoff.h
@@ -0,0 +1,17 @@
+#ifndef REDIS_BACKOFF_H
+#define REDIS_BACKOFF_H
+
+/* {{{ struct RedisBackoff */
+struct RedisBackoff {
+    unsigned int algorithm;        /* index of algorithm function, returns backoff in microseconds*/
+    zend_ulong   base;             /* base backoff in microseconds */
+    zend_ulong   cap;              /* max backoff in microseconds */
+    zend_ulong   previous_backoff; /* previous backoff in microseconds */
+};
+/* }}} */
+
+void redis_initialize_backoff(struct RedisBackoff *self, unsigned long retry_interval);
+void redis_backoff_reset(struct RedisBackoff *self);
+zend_ulong redis_backoff_compute(struct RedisBackoff *self, unsigned int retry_index);
+
+#endif
diff --git a/common.h b/common.h
@@ -21,6 +21,8 @@
 #define NULL   ((void *) 0)
 #endif
 
+#include "backoff.h"
+
 typedef enum {
     REDIS_SOCK_STATUS_FAILED = -1,
     REDIS_SOCK_STATUS_DISCONNECTED,
@@ -83,6 +85,10 @@ typedef enum _PUBSUB_TYPE {
 #define REDIS_OPT_REPLY_LITERAL      8
 #define REDIS_OPT_COMPRESSION_LEVEL  9
 #define REDIS_OPT_NULL_MBULK_AS_NULL 10
+#define REDIS_OPT_MAX_RETRIES        11
+#define REDIS_OPT_BACKOFF_ALGORITHM  12
+#define REDIS_OPT_BACKOFF_BASE       13
+#define REDIS_OPT_BACKOFF_CAP        14
 
 /* cluster options */
 #define REDIS_FAILOVER_NONE              0
@@ -109,6 +115,16 @@ typedef enum {
 #define REDIS_SCAN_PREFIX  2
 #define REDIS_SCAN_NOPREFIX 3
 
+/* BACKOFF_ALGORITHM options */
+#define REDIS_BACKOFF_ALGORITHMS                    7
+#define REDIS_BACKOFF_ALGORITHM_DEFAULT             0
+#define REDIS_BACKOFF_ALGORITHM_DECORRELATED_JITTER 1
+#define REDIS_BACKOFF_ALGORITHM_FULL_JITTER         2
+#define REDIS_BACKOFF_ALGORITHM_EQUAL_JITTER        3
+#define REDIS_BACKOFF_ALGORITHM_EXPONENTIAL         4
+#define REDIS_BACKOFF_ALGORITHM_UNIFORM             5
+#define REDIS_BACKOFF_ALGORITHM_CONSTANT            6
+
 /* GETBIT/SETBIT offset range limits */
 #define BITOP_MIN_OFFSET 0
 #define BITOP_MAX_OFFSET 4294967295U
@@ -258,41 +274,43 @@ typedef enum {
 
 /* {{{ struct RedisSock */
 typedef struct {
-    php_stream         *stream;
-    php_stream_context *stream_ctx;
-    zend_string        *host;
-    int                port;
-    zend_string        *user;
-    zend_string        *pass;
-    double             timeout;
-    double             read_timeout;
-    long               retry_interval;
-    redis_sock_status  status;
-    int                persistent;
-    int                watching;
-    zend_string        *persistent_id;
-
-    redis_serializer   serializer;
-    int                compression;
-    int                compression_level;
-    long               dbNumber;
-
-    zend_string        *prefix;
-
-    short              mode;
-    struct fold_item   *head;
-    struct fold_item   *current;
-
-    zend_string        *pipeline_cmd;
-
-    zend_string        *err;
-
-    int                scan;
-
-    int                readonly;
-    int                reply_literal;
-    int                null_mbulk_as_null;
-    int                tcp_keepalive;
+    php_stream          *stream;
+    php_stream_context  *stream_ctx;
+    zend_string         *host;
+    int                 port;
+    zend_string         *user;
+    zend_string         *pass;
+    double              timeout;
+    double              read_timeout;
+    long                retry_interval;
+    int                 max_retries;
+    struct RedisBackoff backoff;
+    redis_sock_status   status;
+    int                 persistent;
+    int                 watching;
+    zend_string         *persistent_id;
+
+    redis_serializer    serializer;
+    int                 compression;
+    int                 compression_level;
+    long                dbNumber;
+
+    zend_string         *prefix;
+
+    short               mode;
+    struct fold_item    *head;
+    struct fold_item    *current;
+
+    zend_string         *pipeline_cmd;
+
+    zend_string         *err;
+
+    int                 scan;
+
+    int                 readonly;
+    int                 reply_literal;
+    int                 null_mbulk_as_null;
+    int                 tcp_keepalive;
 } RedisSock;
 /* }}} */
 
diff --git a/config.m4 b/config.m4
@@ -323,5 +323,5 @@ if test "$PHP_REDIS" != "no"; then
   fi
 
   PHP_SUBST(REDIS_SHARED_LIBADD)
-  PHP_NEW_EXTENSION(redis, redis.c redis_commands.c library.c redis_session.c redis_array.c redis_array_impl.c redis_cluster.c cluster_library.c redis_sentinel.c sentinel_library.c $lzf_sources, $ext_shared)
+  PHP_NEW_EXTENSION(redis, redis.c redis_commands.c library.c redis_session.c redis_array.c redis_array_impl.c redis_cluster.c cluster_library.c redis_sentinel.c sentinel_library.c backoff.c $lzf_sources, $ext_shared)
 fi
diff --git a/config.w32 b/config.w32
@@ -5,7 +5,7 @@ ARG_ENABLE("redis-session", "whether to enable sessions", "yes");
 ARG_ENABLE("redis-igbinary", "whether to enable igbinary serializer support", "no");
 
 if (PHP_REDIS != "no") {
-	var sources = "redis.c redis_commands.c library.c redis_session.c redis_array.c redis_array_impl.c redis_cluster.c cluster_library.c redis_sentinel.c sentinel_library.c";
+	var sources = "redis.c redis_commands.c library.c redis_session.c redis_array.c redis_array_impl.c redis_cluster.c cluster_library.c redis_sentinel.c sentinel_library.c backoff.c";
 	if (PHP_REDIS_SESSION != "no") {
 		ADD_EXTENSION_DEP("redis", "session");
 		ADD_FLAG("CFLAGS_REDIS", ' /D PHP_SESSION=1 ');
diff --git a/library.c b/library.c
@@ -301,7 +301,7 @@ redis_error_throw(RedisSock *redis_sock)
 PHP_REDIS_API int
 redis_check_eof(RedisSock *redis_sock, int no_throw)
 {
-    int count;
+    unsigned int retry_index;
     char *errmsg;
 
     if (!redis_sock || !redis_sock->stream || redis_sock->status == REDIS_SOCK_STATUS_FAILED) {
@@ -333,18 +333,17 @@ redis_check_eof(RedisSock *redis_sock, int no_throw)
         errmsg = "Connection lost and socket is in MULTI/watching mode";
     } else {
         errmsg = "Connection lost";
-        /* TODO: configurable max retry count */
-        for (count = 0; count < 10; ++count) {
+        redis_backoff_reset(&redis_sock->backoff);
+        for (retry_index = 0; retry_index < redis_sock->max_retries; ++retry_index) {
             /* close existing stream before reconnecting */
             if (redis_sock->stream) {
                 redis_sock_disconnect(redis_sock, 1);
             }
-            // Wait for a while before trying to reconnect
-            if (redis_sock->retry_interval) {
-                // Random factor to avoid having several (or many) concurrent connections trying to reconnect at the same time
-                long retry_interval = (count ? redis_sock->retry_interval : (php_rand() % redis_sock->retry_interval));
-                usleep(retry_interval);
-            }
+            /* Sleep based on our backoff algorithm */
+            zend_ulong delay = redis_backoff_compute(&redis_sock->backoff, retry_index);
+            if (delay != 0)
+                usleep(delay);
+
             /* reconnect */
             if (redis_sock_connect(redis_sock) == 0) {
                 /* check for EOF again. */
@@ -2150,6 +2149,8 @@ redis_sock_create(char *host, int host_len, int port,
     redis_sock->host = zend_string_init(host, host_len, 0);
     redis_sock->status = REDIS_SOCK_STATUS_DISCONNECTED;
     redis_sock->retry_interval = retry_interval * 1000;
+    redis_sock->max_retries = 10;
+    redis_initialize_backoff(&redis_sock->backoff, retry_interval);
     redis_sock->persistent = persistent;
 
     if (persistent && persistent_id != NULL) {
diff --git a/package.xml b/package.xml
@@ -88,6 +88,8 @@ http://pear.php.net/dtd/package-2.0.xsd">
    <file role='doc' name='arrays.markdown'/>
    <file role='doc' name='cluster.markdown'/>
    <file role='doc' name='sentinel.markdown'/>
+   <file role='src' name='backoff.c'/>
+   <file role='src' name='backoff.h'/>
    <file role='src' name='cluster_library.c'/>
    <file role='src' name='cluster_library.h'/>
    <file role='src' name='common.h'/>
diff --git a/redis.c b/redis.c
@@ -775,6 +775,22 @@ static void add_class_constants(zend_class_entry *ce, int is_cluster) {
         zend_declare_class_constant_stringl(ce, "LEFT", 4, "left", 4);
         zend_declare_class_constant_stringl(ce, "RIGHT", 5, "right", 5);
     }
+
+    /* retry/backoff options*/
+    zend_declare_class_constant_long(ce, ZEND_STRL("OPT_MAX_RETRIES"), REDIS_OPT_MAX_RETRIES);
+
+    zend_declare_class_constant_long(ce, ZEND_STRL("OPT_BACKOFF_ALGORITHM"), REDIS_OPT_BACKOFF_ALGORITHM);
+    zend_declare_class_constant_long(ce, ZEND_STRL("BACKOFF_ALGORITHM_DEFAULT"), REDIS_BACKOFF_ALGORITHM_DEFAULT);
+    zend_declare_class_constant_long(ce, ZEND_STRL("BACKOFF_ALGORITHM_CONSTANT"), REDIS_BACKOFF_ALGORITHM_CONSTANT);
+    zend_declare_class_constant_long(ce, ZEND_STRL("BACKOFF_ALGORITHM_UNIFORM"), REDIS_BACKOFF_ALGORITHM_UNIFORM);
+    zend_declare_class_constant_long(ce, ZEND_STRL("BACKOFF_ALGORITHM_EXPONENTIAL"), REDIS_BACKOFF_ALGORITHM_EXPONENTIAL);
+    zend_declare_class_constant_long(ce, ZEND_STRL("BACKOFF_ALGORITHM_FULL_JITTER"), REDIS_BACKOFF_ALGORITHM_FULL_JITTER);
+    zend_declare_class_constant_long(ce, ZEND_STRL("BACKOFF_ALGORITHM_EQUAL_JITTER"), REDIS_BACKOFF_ALGORITHM_EQUAL_JITTER);
+    zend_declare_class_constant_long(ce, ZEND_STRL("BACKOFF_ALGORITHM_DECORRELATED_JITTER"), REDIS_BACKOFF_ALGORITHM_DECORRELATED_JITTER);
+
+    zend_declare_class_constant_long(ce, ZEND_STRL("OPT_BACKOFF_BASE"), REDIS_OPT_BACKOFF_BASE);
+
+    zend_declare_class_constant_long(ce, ZEND_STRL("OPT_BACKOFF_CAP"), REDIS_OPT_BACKOFF_CAP);
 }
 
 static ZEND_RSRC_DTOR_FUNC(redis_connections_pool_dtor)
diff --git a/redis_commands.c b/redis_commands.c
@@ -4308,6 +4308,14 @@ void redis_getoption_handler(INTERNAL_FUNCTION_PARAMETERS,
             RETURN_LONG(redis_sock->null_mbulk_as_null);
         case REDIS_OPT_FAILOVER:
             RETURN_LONG(c->failover);
+        case REDIS_OPT_MAX_RETRIES:
+            RETURN_LONG(redis_sock->max_retries);
+        case REDIS_OPT_BACKOFF_ALGORITHM:
+            RETURN_LONG(redis_sock->backoff.algorithm);
+        case REDIS_OPT_BACKOFF_BASE:
+            RETURN_LONG(redis_sock->backoff.base / 1000);
+        case REDIS_OPT_BACKOFF_CAP:
+            RETURN_LONG(redis_sock->backoff.cap / 1000);
         default:
             RETURN_FALSE;
     }
@@ -4441,6 +4449,35 @@ void redis_setoption_handler(INTERNAL_FUNCTION_PARAMETERS,
                 RETURN_TRUE;
             }
             break;
+        case REDIS_OPT_MAX_RETRIES:
+            val_long = zval_get_long(val);
+            if(val_long >= 0) {
+                redis_sock->max_retries = val_long;
+                RETURN_TRUE;
+            }
+            break;
+        case REDIS_OPT_BACKOFF_ALGORITHM:
+            val_long = zval_get_long(val);
+            if(val_long >= 0 &&
+               val_long < REDIS_BACKOFF_ALGORITHMS) {
+                redis_sock->backoff.algorithm = val_long;
+                RETURN_TRUE;
+            }
+            break;
+        case REDIS_OPT_BACKOFF_BASE:
+            val_long = zval_get_long(val);
+            if(val_long >= 0) {
+                redis_sock->backoff.base = val_long * 1000;
+                RETURN_TRUE;
+            }
+            break;
+        case REDIS_OPT_BACKOFF_CAP:
+            val_long = zval_get_long(val);
+            if(val_long >= 0) {
+                redis_sock->backoff.cap = val_long * 1000;
+                RETURN_TRUE;
+            }
+            break;
         EMPTY_SWITCH_DEFAULT_CASE()
     }
     RETURN_FALSE;
diff --git a/tests/RedisTest.php b/tests/RedisTest.php
