Возможное состояние гонки между сигналом Django post_save и задачей celery

В приложении django 2.0 у меня есть модель с именем Document, которая загружает и сохраняет изображение в файловую систему. Эта часть работает. Я выполняю распознавание лиц на изображении, используя https://github.com/ageitgey/face_recognition в задача сельдерея (v 4.2.1). Я передаю document_id изображения задаче celery, чтобы задача распознавания лиц могла найти изображение для работы. Все это хорошо работает, если я вызываю задачу распознавания лиц вручную из действия DocumentAdmin после сохранения изображения.

Я попытался вызвать задачу face_recognition из метода (models.signals.post_save, sender=Document) в моем models.py, и я получаю сообщение об ошибке из этой строки в задаче celery для face_recognition:

document = Document.objects.get(document_id=document_id)

и ошибка:

[2018-11-26 16:54:28,594: ERROR/ForkPoolWorker-1] Task biometric_identification.tasks.find_faces_task[428ca39b-aefb-4174-9906-ff2146fd6f14] raised unexpected: DoesNotExist('Document matching query does not exist.',)
Traceback (most recent call last):
  File "/home/mark/.virtualenvs/memorabilia-JSON/lib/python3.6/site-packages/celery/app/trace.py", line 382, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/home/mark/.virtualenvs/memorabilia-JSON/lib/python3.6/site-packages/celery/app/trace.py", line 641, in __protected_call__
    return self.run(*args, **kwargs)
  File "/home/mark/python-projects/memorabilia-JSON/biometric_identification/tasks.py", line 42, in find_faces_task
    document = Document.objects.get(document_id=document_id)
  File "/home/mark/.virtualenvs/memorabilia-JSON/lib/python3.6/site-packages/django/db/models/manager.py", line 82, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "/home/mark/.virtualenvs/memorabilia-JSON/lib/python3.6/site-packages/django/db/models/query.py", line 403, in get
    self.model._meta.object_name
memorabilia.models.DoesNotExist: Document matching query does not exist.

Кроме того, эта ошибка возникает не постоянно, а лишь изредка. В остальное время процесс работает; т.е. изображение сохраняется и лица идентифицируются.

Я переопределяю save_model в классе DocumentAdmin, но это просто сохраняет некоторые метаданные для изображения в другой модели. Последняя строка — это вызов super().save_model(request, obj, form, change), поэтому я предполагаю, что после этого вызывается сигнал post_save.

Мне кажется, что существует состояние гонки между сохранением модели в базе данных и задачей сельдерея, запрашивающей базу данных для вновь созданного document_id. Я думал, что сигнал post_save не был активирован, пока модель не была сохранена?

Должен ли я добавить некоторую искусственную задержку в задаче celery face_recognition, чтобы обойти это возможное состояние гонки, или я упустил что-то еще?

Спасибо!

Отметка


django celery django-models celery-task
person user1045680    schedule 27.11.2018    source источник

Ответы (2)


arrow_upward
3
arrow_downward

Проверьте свою функцию, в которой сохранена Document модель. Он упакован в atomic заблокировать где-нибудь или у вас ATOMIC_REQUESTS установлено значение True. Поэтому, когда вызывается post_save, транзакция еще не зафиксирована. Таким образом, ваша модель на самом деле не сохраняется в базе данных в этот момент времени.

person UnholyRaven    schedule 27.11.2018
comment
Я также нашел эту статью о переполнении стека, которую я попробую. stackoverflow.com/questions/45276828/ - person user1045680; 28.11.2018
comment
Эта ссылка хороша, я думаю, что transaction.on_commit + lambda + apply async — это то, что вам нужно. - person Bruno A.; 28.11.2018

arrow_upward
0
arrow_downward

Кажется, что иногда сигнал превосходит вашу скорость записи в БД! Что вы можете сделать в качестве вредного обходного пути, так это запустить задачу celery немного позже, всего на несколько секунд.

Вот как это делается:

your_task.apply_async(
            [document_id],
            countdown=5 # this is the delay in seconds - you can adapt it accordingly
        )

Дайте мне знать, если это работает для вашего случая!

person Kostas Livieratos    schedule 27.11.2018
comment
Мне трудно поверить, что новый Samsung 860 EVO sata SSD работает слишком медленно! Однако я добавил это в код, и теперь он работает: for x in range(0, 4): # try 4 times try: document = Document.objects.get(document_id=document_id) str_error = None except Exception as str_error: pass if str_error: sleep(2) # wait for 2 seconds before trying to fetch the data again else: break - person user1045680; 27.11.2018
comment
Извините, я не могу заставить теги кода работать. Мое решение основано на том, что сказал Костас, но я добавил цикл в задачу сельдерея, чтобы (1) попытаться получить document_id, если исключение, затем спать (2), а затем повторить попытку максимум из 4 попыток. - person user1045680; 27.11.2018