| 1008 | self._continuous_paging_sessions[stream_id].on_error(exc) |
| 1009 | |
| 1010 | def error_all_requests(self, exc): |
| 1011 | with self.lock: |
| 1012 | requests = self._requests |
| 1013 | self._requests = {} |
| 1014 | |
| 1015 | if not requests: |
| 1016 | return |
| 1017 | |
| 1018 | new_exc = ConnectionShutdown(str(exc)) |
| 1019 | |
| 1020 | def try_callback(cb): |
| 1021 | try: |
| 1022 | cb(new_exc) |
| 1023 | except Exception: |
| 1024 | log.warning("Ignoring unhandled exception while erroring requests for a " |
| 1025 | "failed connection (%s) to host %s:", |
| 1026 | id(self), self.endpoint, exc_info=True) |
| 1027 | |
| 1028 | # run first callback from this thread to ensure pool state before leaving |
| 1029 | cb, _, _ = requests.popitem()[1] |
| 1030 | try_callback(cb) |
| 1031 | |
| 1032 | if not requests: |
| 1033 | return |
| 1034 | |
| 1035 | # additional requests are optionally errored from a separate thread |
| 1036 | # The default callback and retry logic is fairly expensive -- we don't |
| 1037 | # want to tie up the event thread when there are many requests |
| 1038 | def err_all_callbacks(): |
| 1039 | for cb, _, _ in requests.values(): |
| 1040 | try_callback(cb) |
| 1041 | if len(requests) < Connection.CALLBACK_ERR_THREAD_THRESHOLD: |
| 1042 | err_all_callbacks() |
| 1043 | else: |
| 1044 | # daemon thread here because we want to stay decoupled from the cluster TPE |
| 1045 | # TODO: would it make sense to just have a driver-global TPE? |
| 1046 | t = Thread(target=err_all_callbacks) |
| 1047 | t.daemon = True |
| 1048 | t.start() |
| 1049 | |
| 1050 | def get_request_id(self): |
| 1051 | """ |