使用上下文管理器从celery的SoftTimeLimitExceeded中恢复

Question

我正在尝试为我的芹菜工作设置最长运行时间.

我目前正在使用上下文管理器从异常中恢复.我最终得到的代码非常类似于此代码段:

from celery.exceptions import SoftTimeLimitExceeded

class Manager:

    def __enter__(self):
        return self

    def __exit__(self, error_type, error, tb):
        if error_type == SoftTimeLimitExceeded:
            logger.info('job killed.')
            # swallow the exception
            return True


@task
def do_foo():
    with Manager():
        run_task1()
        run_task2()
        run_task3()

我的期望:

如果do_foo超时run_task1,则记录器记录,吞下SoftTimeLimitExceeded异常,跳过管理器的主体,作业结束而不运行run_task2和run_task3.

我观察什么: do_foo超时中run_task1,SoftTimeLimitExceeded上升时,记录日志,则SoftTimeLimitExceeded异常吞咽,但 run_task2并run_task3正在运行仍.

我正在寻找以下两个问题的答案:

1. 在此设置中run_task2引发SoftTimeLimitExceeded时为什么仍然执行run_task1？

2. 是否有一种简单的方法来转换我的代码,以便它可以按预期执行？

Answer 1

清理代码

这段代码非常好；没有太多清理工作要做。

• self如果__enter__上下文管理器未设计为与关键字一起使用，则不应返回as。
• is检查类时应该使用，因为它们是单例......
• 但您应该更喜欢issubclass正确模拟异常处理。

实施这些更改可以：

from celery.exceptions import SoftTimeLimitExceeded

class Manager:
    def __enter__(self):
        pass

    def __exit__(self, error_type, error, tb):
        if issubclass(error_type, SoftTimeLimitExceeded):
            logger.info('job killed.')
            # swallow the exception
            return True

@task
def do_foo():
    with Manager():
        run_task1()
        run_task2()
        run_task3()

调试

我创建了一个用于调试的模拟环境：

class SoftTimeLimitExceeded(Exception):
    pass

class Logger:
    info = print
logger = Logger()
del Logger

def task(f):
    return f

def run_task1():
    print("running task 1")
    raise SoftTimeLimitExceeded

def run_task2():
    print("running task 2")

def run_task_3():
    print("running task 3")

执行此命令，然后您的程序给出：

>>> do_foo()
running task 1
job killed.

这是预期的行为。

假设

我能想到两种可能性：

1. 链中的某些东西可能run_task1是异步的。
2. celery正在做一些奇怪的事情。

我将采用第二个假设，因为我无法测试前一个假设。

我以前曾被上下文管理器、异常和协程之间的组合的晦涩行为所困扰，所以我知道它会导致什么样的问题。这似乎是其中之一，但我必须先查看celery的代码，然后才能进一步进行。

编辑：我无法弄清楚 的celery代码的头部或尾部，并且搜索没有找到引发的代码SoftTimeLimitExceeded以允许我向后追踪它。我会将其传递给更有经验的人，celery看看他们是否可以弄清楚它是如何工作的。