Connection Error Tracking & Auto-Reconnect

Author: fcostaoliveira

.gitignore

@@ -33,9 +33,18 @@ tests/tls/*
 *.txt
 !/tests/test_requirements.txt
 __pycache__
+*.csv
+*.json
 
 # Code coverage with lcov/gcov
 *.gcno
 *.gcov
 *.gcda
 *.info
+
+# redis related
+*.rdb
+*.aof
+appendonlydir/
+*.conf
+

client.cpp

@@ -622,6 +622,18 @@ int client_group::create_clients(int num)
         }
 
         m_clients.push_back(c);
+
+        // Add jitter between connection creation (except for the last connection)
+        if (i < num - 1 && m_config->thread_conn_start_max_jitter_micros > 0) {
+            unsigned int jitter_range = m_config->thread_conn_start_max_jitter_micros - m_config->thread_conn_start_min_jitter_micros;
+            unsigned int jitter_micros = m_config->thread_conn_start_min_jitter_micros;
+
+            if (jitter_range > 0) {
+                jitter_micros += rand() % (jitter_range + 1);
+            }
+
+            usleep(jitter_micros);
+        }
     }
 
     return num;
@@ -646,6 +658,32 @@ void client_group::run(void)
     event_base_dispatch(m_base);
 }
 
+void client_group::interrupt(void)
+{
+    // Mark all clients as interrupted
+    set_all_clients_interrupted();
+    // Break the event loop to stop processing
+    event_base_loopbreak(m_base);
+    // Set end time for all clients as close as possible to the loop break
+    finalize_all_clients();
+}
+
+void client_group::finalize_all_clients(void)
+{
+    for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
+        client* c = *i;
+        c->set_end_time();
+    }
+}
+
+void client_group::set_all_clients_interrupted(void)
+{
+    for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
+        client* c = *i;
+        c->get_stats()->set_interrupted(true);
+    }
+}
+
 unsigned long int client_group::get_total_bytes(void)
 {
     unsigned long int total_bytes = 0;
@@ -688,6 +726,16 @@ unsigned long int client_group::get_duration_usec(void)
     return duration;
 }
 
+unsigned long int client_group::get_total_connection_errors(void)
+{
+    unsigned long int total_errors = 0;
+    for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
+        total_errors += (*i)->get_stats()->get_total_connection_errors();
+    }
+
+    return total_errors;
+}
+
 void client_group::merge_run_stats(run_stats* target)
 {
     assert(target != NULL);

client.h

@@ -210,18 +210,22 @@ class client_group {
     int create_clients(int count);
     int prepare(void);
     void run(void);
+    void interrupt(void);
+    void finalize_all_clients(void);
+    void set_all_clients_interrupted(void);
 
     void write_client_stats(const char *prefix);
 
     struct event_base *get_event_base(void) { return m_base; }
     benchmark_config *get_config(void) { return m_config; }
     abstract_protocol* get_protocol(void) { return m_protocol; }
-    object_generator* get_obj_gen(void) { return m_obj_gen; }    
+    object_generator* get_obj_gen(void) { return m_obj_gen; }
 
     unsigned long int get_total_bytes(void);
     unsigned long int get_total_ops(void);
     unsigned long int get_total_latency(void);
     unsigned long int get_duration_usec(void);
+    unsigned long int get_total_connection_errors(void);
 
     void merge_run_stats(run_stats* target);
 };

cluster_client.cpp

@@ -209,6 +209,7 @@ bool cluster_client::connect_shard_connection(shard_connection* sc, char* addres
     memcpy(ci.addr_buf, addr_info->ai_addr, addr_info->ai_addrlen);
     ci.ci_addr = (struct sockaddr *) ci.addr_buf;
     ci.ci_addrlen = addr_info->ai_addrlen;
+
     freeaddrinfo(addr_info);
 
     // call connect
@@ -497,4 +498,3 @@ void cluster_client::handle_response(unsigned int conn_id, struct timeval timest
     // continue with base class
     client::handle_response(conn_id, timestamp, request, response);
 }
-

memtier_benchmark.1

@@ -128,6 +128,24 @@ Number of concurrent pipelined requests (default: 1)
 \fB\-\-reconnect\-interval\fR=\fI\,NUM\/\fR
 Number of requests after which re\-connection is performed
 .TP
+\fB\-\-reconnect\-on\-error\fR
+Enable automatic reconnection on connection errors (default: disabled)
+.TP
+\fB\-\-max\-reconnect\-attempts\fR=\fI\,NUM\/\fR
+Maximum number of reconnection attempts, 0 for unlimited (default: 0)
+.TP
+\fB\-\-reconnect\-backoff\-factor\fR=\fI\,NUM\/\fR
+Backoff factor for reconnection delays, 0 for no backoff (default: 0)
+.TP
+\fB\-\-connection\-timeout\fR=\fI\,SECS\/\fR
+Connection timeout in seconds, 0 to disable (default: 0)
+.TP
+\fB\-\-thread\-conn\-start\-min\-jitter\-micros\fR=\fI\,NUM\/\fR
+Minimum jitter in microseconds between connection creation (default: 0)
+.TP
+\fB\-\-thread\-conn\-start\-max\-jitter\-micros\fR=\fI\,NUM\/\fR
+Maximum jitter in microseconds between connection creation (default: 0)
+.TP
 \fB\-\-multi\-key\-get\fR=\fI\,NUM\/\fR
 Enable multi\-key get commands, up to NUM keys (default: 0)
 .TP