memcache.get返回错误的对象(Celery,Django)

Question

这是我们目前的情况:

1. 我们试图获得缓存的django模型实例,缓存键包括模型名称和实例ID.使用Django的标准memcached后端.该程序是非常广泛使用的常用程序的一部分,不仅在芹菜中.
2. 有时(随机和/或很少)cache.get(key)返回错误的对象:int或不同的模型实例,甚至出现了same-model-different-id情况.通过检查型号名称和id以及缓存键的对应关系来捕捉这个.
3. bug只出现在我们的三个celery任务的上下文中,永远不会在python shell或其他芹菜任务中重现.UPD:仅在长时间运行的CPU-RAM密集型任务下出现
4. 缓存存储正确的值(我们在错误出现时手动检查)
5. 使用相同的参数再次调用相同的任务可能不会重现该问题,尽管概率要高得多,因此错误出现倾向于在同一时间段内"分组"
6. 重启芹菜解决了随机时间段(分钟 - 周)的问题
7. *NEW*这与内存溢出无关.发生这种情况时,我们总是至少有2Gb可用内存.
8. *新*我们有cache_instance = cache.get_cache("cache_entry")静态代码.在调查过程中,我发现错误发生的那一刻会cache_instance.get(key)返回错误的值,尽管get_cache("cache_entry").get(key)在下一行返回正确的值.这意味着错误消失太快或由于某种原因cache_instance对象被破坏.django的缓存线程返回的缓存实例对象不安全吗？
9. *NEW*我们记录了非常奇怪的情况:作为缓存中的另一个错误对象,我们得到了模型实例w/o id set.这意味着,实例从未保存到DB,因此无法缓存.(我希望)
10. *新*MemoryError这些天至少记录了一个

我知道,所有这些听起来都像是某种魔法......而且,真的,任何想法如何可能或如何调试这将是非常感激的.

PS:我目前的假设是这与多处理有关:只要在静态代码中创建缓存实例,并且在工作进程分叉之前,这将导致所有工作者共享相同的套接字(这听起来似乎合理吗？)

Answer 1

终于解决了:

1. Celery具有动态扩展功能 - 它能够根据负载添加/杀死工作人员
2. 它是通过分叉现有的来实现的
3. 打开的套接字和文件被复制到分叉进程,因此当一个进程读取另一个进程的响应时,两个进程共享它们,这会导致竞争条件.简单地说,一个进程可能会读取针对第二个进程的响应,反之亦然.
4. from django.core.cache import cache此对象存储预连接的memcached套接字.当您的进程可以动态分叉时,请不要使用它.并且不要使用存储的连接,池和其他.
5. 或者将它们存储在当前PID下,并在每次访问缓存时进行检查

Answer 2

这一直困扰着我一段时间，直到我找到了这个问题和答案。我只是想补充一些我学到的东西。

您可以使用本地 memcached 实例轻松重现此问题：

from django.core.cache import cache
import os

def write_read_test():
    pid = os.getpid()
    cache.set(pid, pid)
    for x in range(5):
        value = cache.get(pid)
        if value != pid:
            print "Unexpected response {} in process {}. Attempt {}/5".format(
                    value, pid, x+1)
    os._exit(0)

cache.set("access cache", "before fork")
for x in range(5):
    if os.fork() == 0:
        write_read_test()

您可以做的是关闭缓存客户端，就像 Django 在request_finished信号中所做的那样：

https://github.com/django/django/blob/master/django/core/cache/初始化的.py＃L128

如果你cache.close()在叉子后面放一个，一切都会按预期进行。

对于 celery，您可以连接到在工作人员分叉并执行后触发的信号cache.close()。

当预加载处于活动状态并且在分叉工人之前初始化缓存时，这也会影响 gunicorn。

对于gunicorn，你可以用post_fork你的gunicorn配置：

def post_fork(server, worker):
    from django.core.cache import cache
    cache.close()