Get or create a Connection. Can raise ConnectionFailure.
(
self, checkout_started_time: float, handler: Optional[_MongoClientErrorHandler] = None
)
| 1195 | ) |
| 1196 | |
| 1197 | def _get_conn( |
| 1198 | self, checkout_started_time: float, handler: Optional[_MongoClientErrorHandler] = None |
| 1199 | ) -> Connection: |
| 1200 | """Get or create a Connection. Can raise ConnectionFailure.""" |
| 1201 | # We use the pid here to avoid issues with fork / multiprocessing. |
| 1202 | # See test.test_client:TestClient.test_fork for an example of |
| 1203 | # what could go wrong otherwise |
| 1204 | if self.pid != os.getpid(): |
| 1205 | self.reset_without_pause() |
| 1206 | |
| 1207 | if self.closed: |
| 1208 | duration = time.monotonic() - checkout_started_time |
| 1209 | if self.enabled_for_cmap: |
| 1210 | assert self.opts._event_listeners is not None |
| 1211 | self.opts._event_listeners.publish_connection_check_out_failed( |
| 1212 | self.address, ConnectionCheckOutFailedReason.POOL_CLOSED, duration |
| 1213 | ) |
| 1214 | if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG): |
| 1215 | _debug_log( |
| 1216 | _CONNECTION_LOGGER, |
| 1217 | message=_ConnectionStatusMessage.CHECKOUT_FAILED, |
| 1218 | clientId=self._client_id, |
| 1219 | serverHost=self.address[0], |
| 1220 | serverPort=self.address[1], |
| 1221 | reason="Connection pool was closed", |
| 1222 | error=ConnectionCheckOutFailedReason.POOL_CLOSED, |
| 1223 | durationMS=duration, |
| 1224 | ) |
| 1225 | raise _PoolClosedError( |
| 1226 | "Attempted to check out a connection from closed connection pool" |
| 1227 | ) |
| 1228 | |
| 1229 | with self.lock: |
| 1230 | self.operation_count += 1 |
| 1231 | |
| 1232 | # Get a free socket or create one. |
| 1233 | if _csot.get_timeout(): |
| 1234 | deadline = _csot.get_deadline() |
| 1235 | elif self.opts.wait_queue_timeout: |
| 1236 | deadline = time.monotonic() + self.opts.wait_queue_timeout |
| 1237 | else: |
| 1238 | deadline = None |
| 1239 | |
| 1240 | with self.size_cond: |
| 1241 | self._raise_if_not_ready(checkout_started_time, emit_event=True) |
| 1242 | while not (self.requests < self.max_pool_size): |
| 1243 | timeout = deadline - time.monotonic() if deadline else None |
| 1244 | if not _cond_wait(self.size_cond, timeout): |
| 1245 | # Timed out, notify the next thread to ensure a |
| 1246 | # timeout doesn't consume the condition. |
| 1247 | if self.requests < self.max_pool_size: |
| 1248 | self.size_cond.notify() |
| 1249 | self._raise_wait_queue_timeout(checkout_started_time) |
| 1250 | self._raise_if_not_ready(checkout_started_time, emit_event=True) |
| 1251 | self.requests += 1 |
| 1252 | |
| 1253 | # We've now acquired the semaphore and must release it on error. |
| 1254 | conn = None |
no test coverage detected