Method execute

tasktiger/executor.py:406–451

(
        self,
        queue: str,
        tasks: List[Task],
        log: BoundLogger,
        locks: Collection[Lock],
        queue_lock: Optional[Semaphore],
    )

                log.exception("task heartbeat failed")

    def execute(
        self,
        queue: str,
        tasks: List[Task],
        log: BoundLogger,
        locks: Collection[Lock],
        queue_lock: Optional[Semaphore],
    ) -> bool:
        assert tasks

        # Run heartbeat thread.
        all_task_ids = {task.id for task in tasks}
        stop_event = threading.Event()
        heartbeat_thread = threading.Thread(
            target=self._periodic_heartbeat,
            kwargs={
                "queue": queue,
                "task_ids": all_task_ids,
                "log": log,
                "locks": locks,
                "queue_lock": queue_lock,
                "stop_event": stop_event,
            },
        )
        heartbeat_thread.start()

        serialized_task_func = tasks[0].serialized_func
        for task in tasks:
            log.info(
                "processing",
                func=serialized_task_func,
                task_id=task.id,
                params={"args": task.args, "kwargs": task.kwargs},
            )

        # Run the tasks.
        try:
            result = self.execute_tasks(tasks, log)
        # Always stop the heartbeat thread -- even in case of an unhandled
        # exception after running the task code, or when an unhandled
        # BaseException is raised from within the task.
        finally:
            stop_event.set()
            heartbeat_thread.join()

        return result