Make guard to have only one pending reconnect

ControlConnection.reconnect could be calld from many places in parallel.
When it happens you get to see streak for reconnects happning one by
one.
This commit adds guard that makes sure there is only one pending
reconnect.

Author: dkropachev

cassandra/cluster.py

@@ -3561,6 +3561,9 @@ def try_reconnect(self):
     def on_reconnection(self, connection):
         self.control_connection._set_new_connection(connection)
 
+    def on_done(self, success: bool):
+        self.control_connection._reconnection_pending = False
+
     def on_exception(self, exc, next_delay):
         # TODO only overridden to add logging, so add logging
         if isinstance(exc, AuthenticationFailed):
@@ -3662,6 +3665,7 @@ def __init__(self, cluster, timeout,
 
         self._reconnection_handler = None
         self._reconnection_lock = RLock()
+        self._reconnection_pending = False
 
         self._event_schedule_times = {}
 
@@ -3681,6 +3685,7 @@ def _set_new_connection(self, conn):
         with self._lock:
             old = self._connection
             self._connection = conn
+            self._reconnection_pending = False
 
         if old:
             log.debug("[control connection] Closing old connection %r, replacing with %r", old, conn)
@@ -3818,6 +3823,10 @@ def reconnect(self):
         if self._is_shutdown:
             return
 
+        if self._reconnection_pending:
+            return
+        self._reconnection_pending = True
+
         self._submit(self._reconnect)
 
     def _reconnect(self):
@@ -3855,6 +3864,7 @@ def _get_and_set_reconnection_handler(self, new_handler):
         with self._reconnection_lock:
             old = self._reconnection_handler
             self._reconnection_handler = new_handler
+            self._reconnection_pending = False
             return old
 
     def _submit(self, *args, **kwargs):
@@ -4216,6 +4226,9 @@ def wait_for_schema_agreement(self, target_connection=None, preloaded_results=No
         # from the response type and one from the pushed notification. Holding
         # a lock is just a simple way to cut down on the number of schema queries
         # we'll make.
+
+
+
         with self._schema_agreement_lock:
             if self._is_shutdown:
                 return

cassandra/pool.py

@@ -293,6 +293,7 @@ def run(self):
             # call on_exception for logging purposes even if next_delay is None
             if self.on_exception(exc, next_delay):
                 if next_delay is None:
+                    self.on_done(False)
                     log.warning(
                         "Will not continue to retry reconnection attempts "
                         "due to an exhausted retry schedule")
@@ -301,6 +302,7 @@ def run(self):
         else:
             if not self._cancelled:
                 self.on_reconnection(conn)
+                self.on_done(False)
                 self.callback(*(self.callback_args), **(self.callback_kwargs))
         finally:
             if conn:
@@ -324,6 +326,13 @@ def on_reconnection(self, connection):
         """
         pass
 
+    def on_done(self, success: bool):
+        """
+        Called when either new connection is successfully opened or
+        when failed to reconnect and no more reconnections scheduled.
+        """
+        pass
+
     def on_exception(self, exc, next_delay):
         """
         Called when an Exception is raised when trying to connect.

tests/integration/standard/test_cluster.py

@@ -326,6 +326,30 @@ def test_invalid_protocol_negotation(self):
                 cluster.connect()
             cluster.shutdown()
 
+    def test_control_connection_reconnect(self):
+        """
+        Ensure clusters that connect on a keyspace, do
+        """
+        cassandra.cluster.log.setLevel(logging.DEBUG)
+
+        cluster = TestCluster()
+        _ = cluster.connect()
+
+        cluster.control_connection._reconnect_internal = Mock(wraps=cluster.control_connection._reconnect_internal)
+
+        cluster.control_connection.reconnect()
+        cluster.control_connection.reconnect()
+        cluster.control_connection.reconnect()
+        cluster.control_connection.reconnect()
+
+        while cluster.control_connection._reconnection_pending:
+            time.sleep(0.1)
+
+        self.assertFalse(cluster.control_connection._connection.is_closed)
+        self.assertFalse(cluster.control_connection._connection.is_defunct)
+        self.assertTrue(cluster.control_connection.refresh_schema())
+        self.assertEqual(1, len(cluster.control_connection._reconnect_internal.mock_calls))
+
     def test_connect_on_keyspace(self):
         """
         Ensure clusters that connect on a keyspace, do