CPP-745 - Exponential reconnection policy with jitter (#254)

* CPP-745 - Exponential reconnection policy

Author: mikefero

cpp-driver/gtests/src/integration/objects/cluster.hpp

@@ -302,13 +302,13 @@ class Cluster : public Object<CassCluster, cass_cluster_free> {
   }
 
   /**
-   * Sets the amount of time to wait before attempting to reconnect.
+   * Sets the constant reconnection policy.
    *
-   * @param wait_time_ms Wait time in milliseconds (default: 2000)
+   * @param delay_ms Delay in milliseconds (default: 2000)
    * @return Cluster object
    */
-  Cluster& with_reconnect_wait_time(unsigned int wait_time_ms) {
-    cass_cluster_set_reconnect_wait_time(get(), wait_time_ms);
+  Cluster& with_constant_reconnect(unsigned int delay_ms) {
+    cass_cluster_set_constant_reconnect(get(), delay_ms);
     return *this;
   }
 

cpp-driver/gtests/src/integration/tests/test_cluster.cpp

@@ -28,3 +28,21 @@ TEST(ClusterTest, SetLoadBalanceDcAwareNullLocalDc) {
   EXPECT_EQ(CASS_ERROR_LIB_BAD_PARAMS,
             cass_cluster_set_load_balance_dc_aware(cluster.get(), NULL, 99, cass_false));
 }
+
+/**
+ * Set invalid parameters for exponential reconnection policy.
+ *
+ * @jira_ticket CPP-745
+ * @test_category configuration
+ * @expected_result CASS_ERROR_LIB_BAD_PARAMS.
+ */
+TEST(ClusterTest, ExponentialReconnectionPolicyBadParameters) {
+  test::driver::Cluster cluster;
+
+  // Base delay cannot be zero
+  EXPECT_EQ(CASS_ERROR_LIB_BAD_PARAMS, cass_cluster_set_exponential_reconnect(cluster.get(), 0, 1));
+  // Max delay cannot be zero
+  EXPECT_EQ(CASS_ERROR_LIB_BAD_PARAMS, cass_cluster_set_exponential_reconnect(cluster.get(), 1, 0));
+  // Base delay cannot be greater than max delay
+  EXPECT_EQ(CASS_ERROR_LIB_BAD_PARAMS, cass_cluster_set_exponential_reconnect(cluster.get(), 2, 1));
+}

cpp-driver/gtests/src/integration/tests/test_control_connection.cpp

@@ -355,7 +355,7 @@ CASSANDRA_INTEGRATION_TEST_F(ControlConnectionTwoNodeClusterTests, StatusChange)
    * Create a new session connection using the round robin load balancing policy
    * to ensure all nodes can be accessed during request execution
    */
-  Cluster cluster = default_cluster().with_load_balance_round_robin().with_reconnect_wait_time(
+  Cluster cluster = default_cluster().with_load_balance_round_robin().with_constant_reconnect(
       10); // Ensure reconnect timeout is quick
   Session session = cluster.connect();
 

cpp-driver/gtests/src/unit/mockssandra.cpp

@@ -350,7 +350,8 @@ ServerConnection::ServerConnection(const Address& address, const ClientConnectio
     , rc_(0)
     , address_(address)
     , factory_(factory)
-    , ssl_context_(NULL) {
+    , ssl_context_(NULL)
+    , connection_attempts_(0) {
   uv_mutex_init(&mutex_);
   uv_cond_init(&cond_);
 }
@@ -493,6 +494,7 @@ void ServerConnection::wait_close() {
   }
 }
 
+unsigned ServerConnection::connection_attempts() const { return connection_attempts_.load(); }
 void ServerConnection::run(const ServerConnectionTask::Ptr& task) {
   ScopedMutex l(&mutex_);
   if (state_ != STATE_LISTENING) return;
@@ -575,6 +577,8 @@ void ServerConnection::on_connection(uv_stream_t* server, int status) {
 }
 
 void ServerConnection::handle_connection(int status) {
+  connection_attempts_.fetch_add(1);
+
   if (status != 0) {
     fprintf(stderr, "Listen failure: %s\n", uv_strerror(status));
     return;
@@ -2184,6 +2188,14 @@ Hosts Cluster::hosts() const {
   return hosts;
 }
 
+unsigned Cluster::connection_attempts(size_t node) const {
+  if (node < 1 || node > servers_.size()) {
+    return 0;
+  }
+  const Server& server = servers_[node - 1];
+  return server.connection->connection_attempts();
+}
+
 int Cluster::create_and_add_server(AddressGenerator& generator, ClientConnectionFactory& factory,
                                    const String& dc) {
   Address address(generator.next());

cpp-driver/gtests/src/unit/mockssandra.hpp

@@ -176,6 +176,7 @@ class ServerConnection : public RefCounted<ServerConnection> {
   void close();
   void wait_close();
 
+  unsigned connection_attempts() const;
   void run(const ServerConnectionTask::Ptr& task);
 
 private:
@@ -215,6 +216,7 @@ class ServerConnection : public RefCounted<ServerConnection> {
   const Address address_;
   const ClientConnectionFactory& factory_;
   SSL_CTX* ssl_context_;
+  Atomic<unsigned> connection_attempts_;
 };
 
 } // namespace internal
@@ -1162,6 +1164,8 @@ class Cluster {
   const Host& host(const Address& address) const;
   Hosts hosts() const;
 
+  unsigned connection_attempts(size_t node) const;
+
   void event(const Event::Ptr& event);
 
 private:

cpp-driver/gtests/src/unit/tests/test_cluster.cpp

@@ -23,6 +23,8 @@
 using namespace datastax::internal;
 using namespace datastax::internal::core;
 
+#define FIFTEEN_PERCENT(value) static_cast<double>((value * 115) / 100)
+
 class ClusterUnitTest : public EventLoopTest {
 public:
   ClusterUnitTest()
@@ -239,6 +241,53 @@ class ClusterUnitTest : public EventLoopTest {
     mockssandra::Cluster& simple_cluster_;
   };
 
+  class ClusterUnitTestReconnectionPolicy : public ReconnectionPolicy {
+  public:
+    typedef SharedRefPtr<ClusterUnitTestReconnectionPolicy> Ptr;
+
+    ClusterUnitTestReconnectionPolicy()
+        : ReconnectionPolicy(ReconnectionPolicy::CONSTANT)
+        , reconnection_schedule_count_(0)
+        , destroyed_reconnection_schedule_count_(0)
+        , scheduled_delay_count_(0) {}
+
+    virtual const char* name() const { return "blah"; }
+    virtual ReconnectionSchedule* new_reconnection_schedule() {
+      ++reconnection_schedule_count_;
+      return new ClusterUnitTestReconnectionSchedule(&scheduled_delay_count_,
+                                                     &destroyed_reconnection_schedule_count_);
+    }
+
+    unsigned reconnection_schedule_count() const { return reconnection_schedule_count_; }
+    unsigned destroyed_reconnection_schedule_count() const {
+      return destroyed_reconnection_schedule_count_;
+    }
+    unsigned scheduled_delay_count() const { return scheduled_delay_count_; }
+
+  private:
+    unsigned reconnection_schedule_count_;
+    unsigned destroyed_reconnection_schedule_count_;
+    unsigned scheduled_delay_count_;
+
+    class ClusterUnitTestReconnectionSchedule : public ReconnectionSchedule {
+    public:
+      ClusterUnitTestReconnectionSchedule(unsigned* delay_count, unsigned* destroyed_count)
+          : delay_count_(delay_count)
+          , destroyed_count_(destroyed_count) {}
+
+      ~ClusterUnitTestReconnectionSchedule() { ++*destroyed_count_; }
+
+      virtual uint64_t next_delay_ms() {
+        ++*delay_count_;
+        return 1;
+      }
+
+    private:
+      unsigned* delay_count_;
+      unsigned* destroyed_count_;
+    };
+  };
+
   static void on_connection_connected(ClusterConnector* connector, Future* future) {
     if (connector->is_ok()) {
       future->set();
@@ -469,7 +518,7 @@ TEST_F(ClusterUnitTest, ReconnectToDiscoveredHosts) {
   ReconnectClusterListener::Ptr listener(new ReconnectClusterListener(close_future, &outage_plan));
 
   ClusterSettings settings;
-  settings.reconnect_timeout_ms = 1; // Reconnect immediately
+  settings.reconnection_policy.reset(new ConstantReconnectionPolicy(1)); // Reconnect immediately
   settings.control_connection_settings.connection_settings.connect_timeout_ms =
       200; // Give enough time for the connection to complete
 
@@ -512,7 +561,7 @@ TEST_F(ClusterUnitTest, ReconnectUpdateHosts) {
   ReconnectClusterListener::Ptr listener(new ReconnectClusterListener(close_future, &outage_plan));
 
   ClusterSettings settings;
-  settings.reconnect_timeout_ms = 1; // Reconnect immediately
+  settings.reconnection_policy.reset(new ConstantReconnectionPolicy(1)); // Reconnect immediately
   settings.control_connection_settings.connection_settings.connect_timeout_ms =
       200; // Give enough time for the connection to complete
 
@@ -553,7 +602,8 @@ TEST_F(ClusterUnitTest, CloseDuringReconnect) {
   Listener::Ptr listener(new Listener(close_future));
 
   ClusterSettings settings;
-  settings.reconnect_timeout_ms = 100000; // Make sure we're reconnecting when we close.
+  settings.reconnection_policy.reset(
+      new ConstantReconnectionPolicy(100000)); // Make sure we're reconnecting when we close.
 
   connector->with_settings(settings)->with_listener(listener.get())->connect(event_loop());
 
@@ -772,7 +822,7 @@ TEST_F(ClusterUnitTest, DCAwareRecoverOnRemoteHost) {
       new DCAwarePolicy("dc1", 1, false)); // Allow connection to a single remote host
   settings.load_balancing_policies.clear();
   settings.load_balancing_policies.push_back(settings.load_balancing_policy);
-  settings.reconnect_timeout_ms = 1; // Reconnect immediately
+  settings.reconnection_policy.reset(new ConstantReconnectionPolicy(1)); // Reconnect immediately
   settings.control_connection_settings.connection_settings.connect_timeout_ms =
       200; // Give enough time for the connection to complete
 
@@ -868,3 +918,42 @@ TEST_F(ClusterUnitTest, DisableEventsOnStartup) {
   connect_future->cluster()->close();
   ASSERT_TRUE(close_future->wait_for(WAIT_FOR_TIME));
 }
+
+TEST_F(ClusterUnitTest, ReconnectionPolicy) {
+  mockssandra::SimpleCluster mock_cluster(simple());
+  ASSERT_EQ(mock_cluster.start_all(), 0);
+
+  OutagePlan outage_plan(loop(), &mock_cluster);
+  outage_plan.stop_node(1);
+  outage_plan.start_node(1);
+  outage_plan.stop_node(1);
+  outage_plan.start_node(1);
+
+  ContactPointList contact_points;
+  contact_points.push_back("127.0.0.1");
+
+  Future::Ptr close_future(new Future());
+  Future::Ptr connect_future(new Future());
+  ClusterConnector::Ptr connector(
+      new ClusterConnector(contact_points, PROTOCOL_VERSION,
+                           bind_callback(on_connection_reconnect, connect_future.get())));
+  ReconnectClusterListener::Ptr listener(new ReconnectClusterListener(close_future, &outage_plan));
+
+  ClusterSettings settings;
+  settings.reconnection_policy.reset(new ClusterUnitTestReconnectionPolicy());
+  settings.control_connection_settings.connection_settings.connect_timeout_ms =
+      200; // Give enough time for the connection to complete
+  connector->with_settings(settings)->with_listener(listener.get())->connect(event_loop());
+
+  ASSERT_TRUE(connect_future->wait_for(WAIT_FOR_TIME));
+  EXPECT_FALSE(connect_future->error());
+
+  ASSERT_TRUE(close_future->wait_for(WAIT_FOR_TIME));
+
+  ClusterUnitTestReconnectionPolicy::Ptr policy(
+      static_cast<ClusterUnitTestReconnectionPolicy::Ptr>(settings.reconnection_policy));
+  EXPECT_EQ(2, policy->reconnection_schedule_count());
+  EXPECT_EQ(2, policy->destroyed_reconnection_schedule_count());
+  EXPECT_GE(policy->scheduled_delay_count(), 2u);
+  EXPECT_EQ(3, mock_cluster.connection_attempts(1)); // Includes initial connection attempt
+}