Removes stale sockets then adds new ones if pool is too small and has not been reset. The `reference_generation` argument specifies the `generation` at the point in time this operation was requested on the pool.
(self, reference_generation: int)
| 919 | return self.gen.stale(gen, service_id) |
| 920 | |
| 921 | def remove_stale_sockets(self, reference_generation: int) -> None: |
| 922 | """Removes stale sockets then adds new ones if pool is too small and |
| 923 | has not been reset. The `reference_generation` argument specifies the |
| 924 | `generation` at the point in time this operation was requested on the |
| 925 | pool. |
| 926 | """ |
| 927 | # Take the lock to avoid the race condition described in PYTHON-2699. |
| 928 | with self.lock: |
| 929 | if self.state != PoolState.READY: |
| 930 | return |
| 931 | |
| 932 | if self.opts.max_idle_time_seconds is not None: |
| 933 | close_conns = [] |
| 934 | with self.lock: |
| 935 | while ( |
| 936 | self.conns |
| 937 | and self.conns[-1].idle_time_seconds() > self.opts.max_idle_time_seconds |
| 938 | ): |
| 939 | close_conns.append(self.conns.pop()) |
| 940 | if not _IS_SYNC: |
| 941 | asyncio.gather( |
| 942 | *[conn.close_conn(ConnectionClosedReason.IDLE) for conn in close_conns], # type: ignore[func-returns-value] |
| 943 | return_exceptions=True, |
| 944 | ) |
| 945 | else: |
| 946 | for conn in close_conns: |
| 947 | conn.close_conn(ConnectionClosedReason.IDLE) |
| 948 | |
| 949 | while True: |
| 950 | with self.size_cond: |
| 951 | # There are enough sockets in the pool. |
| 952 | if len(self.conns) + self.active_sockets >= self.opts.min_pool_size: |
| 953 | return |
| 954 | if self.requests >= self.opts.min_pool_size: |
| 955 | return |
| 956 | self.requests += 1 |
| 957 | incremented = False |
| 958 | try: |
| 959 | with self._max_connecting_cond: |
| 960 | # If maxConnecting connections are already being created |
| 961 | # by this pool then try again later instead of waiting. |
| 962 | if self._pending >= self._max_connecting: |
| 963 | return |
| 964 | self._pending += 1 |
| 965 | incremented = True |
| 966 | conn = self.connect() |
| 967 | close_conn = False |
| 968 | with self.lock: |
| 969 | # Close connection and return if the pool was reset during |
| 970 | # socket creation or while acquiring the pool lock. |
| 971 | if self.gen.get_overall() != reference_generation: |
| 972 | close_conn = True |
| 973 | if not close_conn: |
| 974 | self.conns.appendleft(conn) |
| 975 | self.active_contexts.discard(conn.cancel_context) |
| 976 | if close_conn: |
| 977 | conn.close_conn(ConnectionClosedReason.STALE) |
| 978 | return |
no test coverage detected