| 1688 | |
| 1689 | |
| 1690 | class ConnectionHeartbeat(Thread): |
| 1691 | |
| 1692 | def __init__(self, interval_sec, get_connection_holders, timeout): |
| 1693 | Thread.__init__(self, name="Connection heartbeat") |
| 1694 | self._interval = interval_sec |
| 1695 | self._timeout = timeout |
| 1696 | self._get_connection_holders = get_connection_holders |
| 1697 | self._shutdown_event = Event() |
| 1698 | self.daemon = True |
| 1699 | self.start() |
| 1700 | |
| 1701 | class ShutdownException(Exception): |
| 1702 | pass |
| 1703 | |
| 1704 | def run(self): |
| 1705 | self._shutdown_event.wait(self._interval) |
| 1706 | while not self._shutdown_event.is_set(): |
| 1707 | start_time = time.time() |
| 1708 | |
| 1709 | futures = [] |
| 1710 | failed_connections = [] |
| 1711 | try: |
| 1712 | for connections, owner in [(o.get_connections(), o) for o in self._get_connection_holders()]: |
| 1713 | for connection in connections: |
| 1714 | self._raise_if_stopped() |
| 1715 | if not (connection.is_defunct or connection.is_closed): |
| 1716 | if connection.is_idle: |
| 1717 | try: |
| 1718 | futures.append(HeartbeatFuture(connection, owner)) |
| 1719 | except Exception as e: |
| 1720 | log.warning("Failed sending heartbeat message on connection (%s) to %s", |
| 1721 | id(connection), connection.endpoint) |
| 1722 | failed_connections.append((connection, owner, e)) |
| 1723 | else: |
| 1724 | connection.reset_idle() |
| 1725 | else: |
| 1726 | log.debug("Cannot send heartbeat message on connection (%s) to %s", |
| 1727 | id(connection), connection.endpoint) |
| 1728 | # make sure the owner sees this defunct/closed connection |
| 1729 | owner.return_connection(connection) |
| 1730 | self._raise_if_stopped() |
| 1731 | |
| 1732 | # Wait max `self._timeout` seconds for all HeartbeatFutures to complete |
| 1733 | timeout = self._timeout |
| 1734 | start_time = time.time() |
| 1735 | for f in futures: |
| 1736 | self._raise_if_stopped() |
| 1737 | connection = f.connection |
| 1738 | try: |
| 1739 | f.wait(timeout) |
| 1740 | # TODO: move this, along with connection locks in pool, down into Connection |
| 1741 | with connection.lock: |
| 1742 | connection.in_flight -= 1 |
| 1743 | connection.reset_idle() |
| 1744 | except Exception as e: |
| 1745 | log.warning("Heartbeat failed for connection (%s) to %s", |
| 1746 | id(connection), connection.endpoint) |
| 1747 | failed_connections.append((f.connection, f.owner, e)) |
no outgoing calls