Method execute

tasktiger/executor.py:186–381

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

    """

    def execute(
        self,
        queue: str,
        tasks: List[Task],
        log: BoundLogger,
        locks: Collection[Lock],
        queue_lock: Optional[Semaphore],
    ) -> bool:
        task_func = tasks[0].func
        serialized_task_func = tasks[0].serialized_func

        all_task_ids = {task.id for task in tasks}
        with g_fork_lock:
            child_pid = os.fork()

        if child_pid == 0:
            # Child process
            log = log.bind(child_pid=os.getpid())
            assert isinstance(log, BoundLogger)

            # Disconnect the Redis connection inherited from the main process.
            # Note that this doesn't disconnect the socket in the main process.
            self.connection.connection_pool.disconnect()

            random.seed()

            # Ignore Ctrl+C in the child so we don't abort the job -- the main
            # process already takes care of a graceful shutdown.
            signal.signal(signal.SIGINT, signal.SIG_IGN)

            # Run the tasks.
            success = self.execute_tasks(tasks, log)

            # Wait for any threads that might be running in the child, just
            # like sys.exit() would. Note we don't call sys.exit() directly
            # because it would perform additional cleanup (e.g. calling atexit
            # handlers twice). See also: https://bugs.python.org/issue18966
            threading._shutdown()  # type: ignore[attr-defined]

            os._exit(int(not success))
        else:
            # Main process
            log = log.bind(child_pid=child_pid)
            assert isinstance(log, BoundLogger)
            for task in tasks:
                log.info(
                    "processing",
                    func=serialized_task_func,
                    task_id=task.id,
                    params={"args": task.args, "kwargs": task.kwargs},
                )

            # Attach a signal handler to SIGCHLD (sent when the child process
            # exits) so we can capture it.
            signal.signal(signal.SIGCHLD, sigchld_handler)

            # Since newer Python versions retry interrupted system calls we can't
            # rely on the fact that select() is interrupted with EINTR. Instead,