Used to pool connections to a host for v1 and v2 native protocol.
| 584 | |
| 585 | |
| 586 | class HostConnectionPool(object): |
| 587 | """ |
| 588 | Used to pool connections to a host for v1 and v2 native protocol. |
| 589 | """ |
| 590 | |
| 591 | host = None |
| 592 | host_distance = None |
| 593 | |
| 594 | is_shutdown = False |
| 595 | open_count = 0 |
| 596 | _scheduled_for_creation = 0 |
| 597 | _next_trash_allowed_at = 0 |
| 598 | _keyspace = None |
| 599 | |
| 600 | def __init__(self, host, host_distance, session): |
| 601 | self.host = host |
| 602 | self.host_distance = host_distance |
| 603 | |
| 604 | self._session = weakref.proxy(session) |
| 605 | self._lock = RLock() |
| 606 | self._conn_available_condition = Condition() |
| 607 | |
| 608 | log.debug("Initializing new connection pool for host %s", self.host) |
| 609 | core_conns = session.cluster.get_core_connections_per_host(host_distance) |
| 610 | self._connections = [session.cluster.connection_factory(host.endpoint, on_orphaned_stream_released=self.on_orphaned_stream_released) |
| 611 | for i in range(core_conns)] |
| 612 | |
| 613 | self._keyspace = session.keyspace |
| 614 | if self._keyspace: |
| 615 | for conn in self._connections: |
| 616 | conn.set_keyspace_blocking(self._keyspace) |
| 617 | |
| 618 | self._trash = set() |
| 619 | self._next_trash_allowed_at = time.time() |
| 620 | self.open_count = core_conns |
| 621 | log.debug("Finished initializing new connection pool for host %s", self.host) |
| 622 | |
| 623 | def borrow_connection(self, timeout): |
| 624 | if self.is_shutdown: |
| 625 | raise ConnectionException( |
| 626 | "Pool for %s is shutdown" % (self.host,), self.host) |
| 627 | |
| 628 | conns = self._connections |
| 629 | if not conns: |
| 630 | # handled specially just for simpler code |
| 631 | log.debug("Detected empty pool, opening core conns to %s", self.host) |
| 632 | core_conns = self._session.cluster.get_core_connections_per_host(self.host_distance) |
| 633 | with self._lock: |
| 634 | # we check the length of self._connections again |
| 635 | # along with self._scheduled_for_creation while holding the lock |
| 636 | # in case multiple threads hit this condition at the same time |
| 637 | to_create = core_conns - (len(self._connections) + self._scheduled_for_creation) |
| 638 | for i in range(to_create): |
| 639 | self._scheduled_for_creation += 1 |
| 640 | self._session.submit(self._create_new_connection) |
| 641 | |
| 642 | # in_flight is incremented by wait_for_conn |
| 643 | conn = self._wait_for_conn(timeout) |
no outgoing calls
no test coverage detected