cluster.py 文件源码

def run(self):
        while True:
            if self.is_shutdown:
                return

            try:
                while True:
                    run_at, i, task = self._queue.get(block=True, timeout=None)
                    if self.is_shutdown:
                        if task:
                            log.debug("Not executing scheduled task due to Scheduler shutdown")
                        return
                    if run_at <= time.time():
                        self._scheduled_tasks.discard(task)
                        fn, args, kwargs = task
                        kwargs = dict(kwargs)
                        future = self._executor.submit(fn, *args, **kwargs)
                        future.add_done_callback(self._log_if_failed)
                    else:
                        self._queue.put_nowait((run_at, i, task))
                        break
            except Queue.Empty:
                pass

            time.sleep(0.1)