Called when the request associated with this ResponseFuture times out. This function may reschedule itself. The ``_attempts`` parameter tracks the number of times this has happened. This parameter should only be set in those cases, where ``_on_timeout`` reschedules
(self, _attempts=0)
| 4443 | self._timer.cancel() |
| 4444 | |
| 4445 | def _on_timeout(self, _attempts=0): |
| 4446 | """ |
| 4447 | Called when the request associated with this ResponseFuture times out. |
| 4448 | |
| 4449 | This function may reschedule itself. The ``_attempts`` parameter tracks |
| 4450 | the number of times this has happened. This parameter should only be |
| 4451 | set in those cases, where ``_on_timeout`` reschedules itself. |
| 4452 | """ |
| 4453 | # PYTHON-853: for short timeouts, we sometimes race with our __init__ |
| 4454 | if self._connection is None and _attempts < 3: |
| 4455 | self._timer = self.session.cluster.connection_class.create_timer( |
| 4456 | 0.01, |
| 4457 | partial(self._on_timeout, _attempts=_attempts + 1) |
| 4458 | ) |
| 4459 | return |
| 4460 | |
| 4461 | if self._connection is not None: |
| 4462 | try: |
| 4463 | self._connection._requests.pop(self._req_id) |
| 4464 | # PYTHON-1044 |
| 4465 | # This request might have been removed from the connection after the latter was defunct by heartbeat. |
| 4466 | # We should still raise OperationTimedOut to reject the future so that the main event thread will not |
| 4467 | # wait for it endlessly |
| 4468 | except KeyError: |
| 4469 | key = "Connection defunct by heartbeat" |
| 4470 | errors = {key: "Client request timeout. See Session.execute[_async](timeout)"} |
| 4471 | self._set_final_exception(OperationTimedOut(errors, self._current_host)) |
| 4472 | return |
| 4473 | |
| 4474 | pool = self.session._pools.get(self._current_host) |
| 4475 | if pool and not pool.is_shutdown: |
| 4476 | # Do not return the stream ID to the pool yet. We cannot reuse it |
| 4477 | # because the node might still be processing the query and will |
| 4478 | # return a late response to that query - if we used such stream |
| 4479 | # before the response to the previous query has arrived, the new |
| 4480 | # query could get a response from the old query |
| 4481 | with self._connection.lock: |
| 4482 | self._connection.orphaned_request_ids.add(self._req_id) |
| 4483 | if len(self._connection.orphaned_request_ids) >= self._connection.orphaned_threshold: |
| 4484 | self._connection.orphaned_threshold_reached = True |
| 4485 | |
| 4486 | pool.return_connection(self._connection, stream_was_orphaned=True) |
| 4487 | |
| 4488 | errors = self._errors |
| 4489 | if not errors: |
| 4490 | if self.is_schema_agreed: |
| 4491 | key = str(self._current_host.endpoint) if self._current_host else 'no host queried before timeout' |
| 4492 | errors = {key: "Client request timeout. See Session.execute[_async](timeout)"} |
| 4493 | else: |
| 4494 | connection = self.session.cluster.control_connection._connection |
| 4495 | host = str(connection.endpoint) if connection else 'unknown' |
| 4496 | errors = {host: "Request timed out while waiting for schema agreement. See Session.execute[_async](timeout) and Cluster.max_schema_agreement_wait."} |
| 4497 | |
| 4498 | self._set_final_exception(OperationTimedOut(errors, self._current_host)) |
| 4499 | |
| 4500 | def _on_speculative_execute(self): |
| 4501 | self._timer = None |