(self, timeout)
| 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) |
| 644 | return conn |
| 645 | else: |
| 646 | # note: it would be nice to push changes to these config settings |
| 647 | # to pools instead of doing a new lookup on every |
| 648 | # borrow_connection() call |
| 649 | max_reqs = self._session.cluster.get_max_requests_per_connection(self.host_distance) |
| 650 | max_conns = self._session.cluster.get_max_connections_per_host(self.host_distance) |
| 651 | |
| 652 | least_busy = min(conns, key=lambda c: c.in_flight) |
| 653 | request_id = None |
| 654 | # to avoid another thread closing this connection while |
| 655 | # trashing it (through the return_connection process), hold |
| 656 | # the connection lock from this point until we've incremented |
| 657 | # its in_flight count |
| 658 | need_to_wait = False |
| 659 | with least_busy.lock: |
| 660 | if least_busy.in_flight < least_busy.max_request_id: |
| 661 | least_busy.in_flight += 1 |
| 662 | request_id = least_busy.get_request_id() |
| 663 | else: |
| 664 | # once we release the lock, wait for another connection |
| 665 | need_to_wait = True |
| 666 | |
| 667 | if need_to_wait: |
| 668 | # wait_for_conn will increment in_flight on the conn |
| 669 | least_busy, request_id = self._wait_for_conn(timeout) |
| 670 | |
| 671 | # if we have too many requests on this connection, but we still |
| 672 | # have space to open a new connection against this host, go ahead |
| 673 | # and schedule the creation of a new connection |
| 674 | if least_busy.in_flight >= max_reqs and len(self._connections) < max_conns: |
| 675 | self._maybe_spawn_new_connection() |
| 676 | |
| 677 | return least_busy, request_id |
| 678 | |
| 679 | def _maybe_spawn_new_connection(self): |
| 680 | with self._lock: |
nothing calls this directly
no test coverage detected