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