Method run

cassandra/connection.py:1704–1764

(self)

        pass

    def run(self):
        self._shutdown_event.wait(self._interval)
        while not self._shutdown_event.is_set():
            start_time = time.time()

            futures = []
            failed_connections = []
            try:
                for connections, owner in [(o.get_connections(), o) for o in self._get_connection_holders()]:
                    for connection in connections:
                        self._raise_if_stopped()
                        if not (connection.is_defunct or connection.is_closed):
                            if connection.is_idle:
                                try:
                                    futures.append(HeartbeatFuture(connection, owner))
                                except Exception as e:
                                    log.warning("Failed sending heartbeat message on connection (%s) to %s",
                                                id(connection), connection.endpoint)
                                    failed_connections.append((connection, owner, e))
                            else:
                                connection.reset_idle()
                        else:
                            log.debug("Cannot send heartbeat message on connection (%s) to %s",
                                      id(connection), connection.endpoint)
                            # make sure the owner sees this defunct/closed connection
                            owner.return_connection(connection)
                    self._raise_if_stopped()

                # Wait max `self._timeout` seconds for all HeartbeatFutures to complete
                timeout = self._timeout
                start_time = time.time()
                for f in futures:
                    self._raise_if_stopped()
                    connection = f.connection
                    try:
                        f.wait(timeout)
                        # TODO: move this, along with connection locks in pool, down into Connection
                        with connection.lock:
                            connection.in_flight -= 1
                        connection.reset_idle()
                    except Exception as e:
                        log.warning("Heartbeat failed for connection (%s) to %s",
                                    id(connection), connection.endpoint)
                        failed_connections.append((f.connection, f.owner, e))

                    timeout = self._timeout - (time.time() - start_time)

                for connection, owner, exc in failed_connections:
                    self._raise_if_stopped()
                    if not connection.is_control_connection:
                        # Only HostConnection supports shutdown_on_error
                        owner.shutdown_on_error = True
                    connection.defunct(exc)
                    owner.return_connection(connection)
            except self.ShutdownException:
                pass
            except Exception:
                log.error("Failed connection heartbeat", exc_info=True)