| 4278 | log.debug("Ignoring scheduled task after shutdown: %r", task) |
| 4279 | |
| 4280 | def run(self): |
| 4281 | while True: |
| 4282 | if self.is_shutdown: |
| 4283 | return |
| 4284 | |
| 4285 | try: |
| 4286 | while True: |
| 4287 | run_at, i, task = self._queue.get(block=True, timeout=None) |
| 4288 | if self.is_shutdown: |
| 4289 | if task: |
| 4290 | log.debug("Not executing scheduled task due to Scheduler shutdown") |
| 4291 | return |
| 4292 | if run_at <= time.time(): |
| 4293 | self._scheduled_tasks.discard(task) |
| 4294 | fn, args, kwargs = task |
| 4295 | kwargs = dict(kwargs) |
| 4296 | future = self._executor.submit(fn, *args, **kwargs) |
| 4297 | future.add_done_callback(self._log_if_failed) |
| 4298 | else: |
| 4299 | self._queue.put_nowait((run_at, i, task)) |
| 4300 | break |
| 4301 | except queue.Empty: |
| 4302 | pass |
| 4303 | |
| 4304 | time.sleep(0.1) |
| 4305 | |
| 4306 | def _log_if_failed(self, future): |
| 4307 | exc = future.exception() |