Stop ignoring task results

This is essentially a revert of 9cbff312a. The commit was at the advice
of the Celery docs for optimization, but I've since decided that the
downsides in terms of making things harder to debug (it makes Flower
nearly useless, for instance) are bigger than the upsides in performance
gain (which seem extremely small in practice, given how long our tasks
take, and the number of tasks we have).
This commit is contained in:
Wesley Aptekar-Cassels 2023-04-07 21:51:44 -04:00
parent 7272ca2564
commit 1048638e30
12 changed files with 34 additions and 34 deletions

View file

@ -241,7 +241,7 @@ class ActivityObject:
return data return data
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
@transaction.atomic @transaction.atomic
def set_related_field( def set_related_field(
model_name, origin_model_name, related_field_name, related_remote_id, data model_name, origin_model_name, related_field_name, related_remote_id, data

View file

@ -497,7 +497,7 @@ def remove_statuses_on_unshelve(sender, instance, *args, **kwargs):
# ---- TASKS # ---- TASKS
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def add_book_statuses_task(user_id, book_id): def add_book_statuses_task(user_id, book_id):
"""add statuses related to a book on shelve""" """add statuses related to a book on shelve"""
user = models.User.objects.get(id=user_id) user = models.User.objects.get(id=user_id)
@ -505,7 +505,7 @@ def add_book_statuses_task(user_id, book_id):
BooksStream().add_book_statuses(user, book) BooksStream().add_book_statuses(user, book)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def remove_book_statuses_task(user_id, book_id): def remove_book_statuses_task(user_id, book_id):
"""remove statuses about a book from a user's books feed""" """remove statuses about a book from a user's books feed"""
user = models.User.objects.get(id=user_id) user = models.User.objects.get(id=user_id)
@ -513,7 +513,7 @@ def remove_book_statuses_task(user_id, book_id):
BooksStream().remove_book_statuses(user, book) BooksStream().remove_book_statuses(user, book)
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def populate_stream_task(stream, user_id): def populate_stream_task(stream, user_id):
"""background task for populating an empty activitystream""" """background task for populating an empty activitystream"""
user = models.User.objects.get(id=user_id) user = models.User.objects.get(id=user_id)
@ -521,7 +521,7 @@ def populate_stream_task(stream, user_id):
stream.populate_streams(user) stream.populate_streams(user)
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def remove_status_task(status_ids): def remove_status_task(status_ids):
"""remove a status from any stream it might be in""" """remove a status from any stream it might be in"""
# this can take an id or a list of ids # this can take an id or a list of ids
@ -536,7 +536,7 @@ def remove_status_task(status_ids):
) )
@app.task(queue=HIGH, ignore_result=True) @app.task(queue=HIGH)
def add_status_task(status_id, increment_unread=False): def add_status_task(status_id, increment_unread=False):
"""add a status to any stream it should be in""" """add a status to any stream it should be in"""
status = models.Status.objects.select_subclasses().get(id=status_id) status = models.Status.objects.select_subclasses().get(id=status_id)
@ -548,7 +548,7 @@ def add_status_task(status_id, increment_unread=False):
stream.add_status(status, increment_unread=increment_unread) stream.add_status(status, increment_unread=increment_unread)
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def remove_user_statuses_task(viewer_id, user_id, stream_list=None): def remove_user_statuses_task(viewer_id, user_id, stream_list=None):
"""remove all statuses by a user from a viewer's stream""" """remove all statuses by a user from a viewer's stream"""
stream_list = [streams[s] for s in stream_list] if stream_list else streams.values() stream_list = [streams[s] for s in stream_list] if stream_list else streams.values()
@ -558,7 +558,7 @@ def remove_user_statuses_task(viewer_id, user_id, stream_list=None):
stream.remove_user_statuses(viewer, user) stream.remove_user_statuses(viewer, user)
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def add_user_statuses_task(viewer_id, user_id, stream_list=None): def add_user_statuses_task(viewer_id, user_id, stream_list=None):
"""add all statuses by a user to a viewer's stream""" """add all statuses by a user to a viewer's stream"""
stream_list = [streams[s] for s in stream_list] if stream_list else streams.values() stream_list = [streams[s] for s in stream_list] if stream_list else streams.values()
@ -568,7 +568,7 @@ def add_user_statuses_task(viewer_id, user_id, stream_list=None):
stream.add_user_statuses(viewer, user) stream.add_user_statuses(viewer, user)
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def handle_boost_task(boost_id): def handle_boost_task(boost_id):
"""remove the original post and other, earlier boosts""" """remove the original post and other, earlier boosts"""
instance = models.Status.objects.get(id=boost_id) instance = models.Status.objects.get(id=boost_id)

View file

@ -143,7 +143,7 @@ def get_or_create_connector(remote_id):
return load_connector(connector_info) return load_connector(connector_info)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def load_more_data(connector_id, book_id): def load_more_data(connector_id, book_id):
"""background the work of getting all 10,000 editions of LoTR""" """background the work of getting all 10,000 editions of LoTR"""
connector_info = models.Connector.objects.get(id=connector_id) connector_info = models.Connector.objects.get(id=connector_id)
@ -152,7 +152,7 @@ def load_more_data(connector_id, book_id):
connector.expand_book_data(book) connector.expand_book_data(book)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def create_edition_task(connector_id, work_id, data): def create_edition_task(connector_id, work_id, data):
"""separate task for each of the 10,000 editions of LoTR""" """separate task for each of the 10,000 editions of LoTR"""
connector_info = models.Connector.objects.get(id=connector_id) connector_info = models.Connector.objects.get(id=connector_id)

View file

@ -75,7 +75,7 @@ def format_email(email_name, data):
return (subject, html_content, text_content) return (subject, html_content, text_content)
@app.task(queue=HIGH, ignore_result=True) @app.task(queue=HIGH)
def send_email(recipient, subject, html_content, text_content): def send_email(recipient, subject, html_content, text_content):
"""use a task to send the email""" """use a task to send the email"""
email = EmailMultiAlternatives( email = EmailMultiAlternatives(

View file

@ -217,14 +217,14 @@ def add_list_on_account_create_command(user_id):
# ---- TASKS # ---- TASKS
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def populate_lists_task(user_id): def populate_lists_task(user_id):
"""background task for populating an empty list stream""" """background task for populating an empty list stream"""
user = models.User.objects.get(id=user_id) user = models.User.objects.get(id=user_id)
ListsStream().populate_lists(user) ListsStream().populate_lists(user)
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def remove_list_task(list_id, re_add=False): def remove_list_task(list_id, re_add=False):
"""remove a list from any stream it might be in""" """remove a list from any stream it might be in"""
stores = models.User.objects.filter(local=True, is_active=True).values_list( stores = models.User.objects.filter(local=True, is_active=True).values_list(
@ -239,14 +239,14 @@ def remove_list_task(list_id, re_add=False):
add_list_task.delay(list_id) add_list_task.delay(list_id)
@app.task(queue=HIGH, ignore_result=True) @app.task(queue=HIGH)
def add_list_task(list_id): def add_list_task(list_id):
"""add a list to any stream it should be in""" """add a list to any stream it should be in"""
book_list = models.List.objects.get(id=list_id) book_list = models.List.objects.get(id=list_id)
ListsStream().add_list(book_list) ListsStream().add_list(book_list)
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def remove_user_lists_task(viewer_id, user_id, exclude_privacy=None): def remove_user_lists_task(viewer_id, user_id, exclude_privacy=None):
"""remove all lists by a user from a viewer's stream""" """remove all lists by a user from a viewer's stream"""
viewer = models.User.objects.get(id=viewer_id) viewer = models.User.objects.get(id=viewer_id)
@ -254,7 +254,7 @@ def remove_user_lists_task(viewer_id, user_id, exclude_privacy=None):
ListsStream().remove_user_lists(viewer, user, exclude_privacy=exclude_privacy) ListsStream().remove_user_lists(viewer, user, exclude_privacy=exclude_privacy)
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def add_user_lists_task(viewer_id, user_id): def add_user_lists_task(viewer_id, user_id):
"""add all lists by a user to a viewer's stream""" """add all lists by a user to a viewer's stream"""
viewer = models.User.objects.get(id=viewer_id) viewer = models.User.objects.get(id=viewer_id)

View file

@ -506,7 +506,7 @@ def unfurl_related_field(related_field, sort_field=None):
return related_field.remote_id return related_field.remote_id
@app.task(queue=BROADCAST, ignore_result=True) @app.task(queue=BROADCAST)
def broadcast_task(sender_id: int, activity: str, recipients: List[str]): def broadcast_task(sender_id: int, activity: str, recipients: List[str]):
"""the celery task for broadcast""" """the celery task for broadcast"""
user_model = apps.get_model("bookwyrm.User", require_ready=True) user_model = apps.get_model("bookwyrm.User", require_ready=True)

View file

@ -65,7 +65,7 @@ class AutoMod(AdminModel):
created_by = models.ForeignKey("User", on_delete=models.PROTECT) created_by = models.ForeignKey("User", on_delete=models.PROTECT)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def automod_task(): def automod_task():
"""Create reports""" """Create reports"""
if not AutoMod.objects.exists(): if not AutoMod.objects.exists():

View file

@ -327,7 +327,7 @@ class ImportItem(models.Model):
) )
@app.task(queue=IMPORTS, ignore_result=True) @app.task(queue=IMPORTS)
def start_import_task(job_id): def start_import_task(job_id):
"""trigger the child tasks for each row""" """trigger the child tasks for each row"""
job = ImportJob.objects.get(id=job_id) job = ImportJob.objects.get(id=job_id)
@ -346,7 +346,7 @@ def start_import_task(job_id):
job.save() job.save()
@app.task(queue=IMPORTS, ignore_result=True) @app.task(queue=IMPORTS)
def import_item_task(item_id): def import_item_task(item_id):
"""resolve a row into a book""" """resolve a row into a book"""
item = ImportItem.objects.get(id=item_id) item = ImportItem.objects.get(id=item_id)

View file

@ -469,7 +469,7 @@ class KeyPair(ActivitypubMixin, BookWyrmModel):
return super().save(*args, **kwargs) return super().save(*args, **kwargs)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def set_remote_server(user_id): def set_remote_server(user_id):
"""figure out the user's remote server in the background""" """figure out the user's remote server in the background"""
user = User.objects.get(id=user_id) user = User.objects.get(id=user_id)
@ -513,7 +513,7 @@ def get_or_create_remote_server(domain, refresh=False):
return server return server
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def get_remote_reviews(outbox): def get_remote_reviews(outbox):
"""ingest reviews by a new remote bookwyrm user""" """ingest reviews by a new remote bookwyrm user"""
outbox_page = outbox + "?page=true&type=Review" outbox_page = outbox + "?page=true&type=Review"

View file

@ -420,7 +420,7 @@ def save_and_cleanup(image, instance=None):
# pylint: disable=invalid-name # pylint: disable=invalid-name
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def generate_site_preview_image_task(): def generate_site_preview_image_task():
"""generate preview_image for the website""" """generate preview_image for the website"""
if not settings.ENABLE_PREVIEW_IMAGES: if not settings.ENABLE_PREVIEW_IMAGES:
@ -445,7 +445,7 @@ def generate_site_preview_image_task():
# pylint: disable=invalid-name # pylint: disable=invalid-name
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def generate_edition_preview_image_task(book_id): def generate_edition_preview_image_task(book_id):
"""generate preview_image for a book""" """generate preview_image for a book"""
if not settings.ENABLE_PREVIEW_IMAGES: if not settings.ENABLE_PREVIEW_IMAGES:
@ -470,7 +470,7 @@ def generate_edition_preview_image_task(book_id):
save_and_cleanup(image, instance=book) save_and_cleanup(image, instance=book)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def generate_user_preview_image_task(user_id): def generate_user_preview_image_task(user_id):
"""generate preview_image for a user""" """generate preview_image for a user"""
if not settings.ENABLE_PREVIEW_IMAGES: if not settings.ENABLE_PREVIEW_IMAGES:
@ -496,7 +496,7 @@ def generate_user_preview_image_task(user_id):
save_and_cleanup(image, instance=user) save_and_cleanup(image, instance=user)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def remove_user_preview_image_task(user_id): def remove_user_preview_image_task(user_id):
"""remove preview_image for a user""" """remove preview_image for a user"""
if not settings.ENABLE_PREVIEW_IMAGES: if not settings.ENABLE_PREVIEW_IMAGES:

View file

@ -244,20 +244,20 @@ def domain_level_update(sender, instance, created, update_fields=None, **kwargs)
# ------------------- TASKS # ------------------- TASKS
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def rerank_suggestions_task(user_id): def rerank_suggestions_task(user_id):
"""do the hard work in celery""" """do the hard work in celery"""
suggested_users.rerank_user_suggestions(user_id) suggested_users.rerank_user_suggestions(user_id)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def rerank_user_task(user_id, update_only=False): def rerank_user_task(user_id, update_only=False):
"""do the hard work in celery""" """do the hard work in celery"""
user = models.User.objects.get(id=user_id) user = models.User.objects.get(id=user_id)
suggested_users.rerank_obj(user, update_only=update_only) suggested_users.rerank_obj(user, update_only=update_only)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def remove_user_task(user_id): def remove_user_task(user_id):
"""do the hard work in celery""" """do the hard work in celery"""
user = models.User.objects.get(id=user_id) user = models.User.objects.get(id=user_id)
@ -266,14 +266,14 @@ def remove_user_task(user_id):
) )
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def remove_suggestion_task(user_id, suggested_user_id): def remove_suggestion_task(user_id, suggested_user_id):
"""remove a specific user from a specific user's suggestions""" """remove a specific user from a specific user's suggestions"""
suggested_user = models.User.objects.get(id=suggested_user_id) suggested_user = models.User.objects.get(id=suggested_user_id)
suggested_users.remove_suggestion(user_id, suggested_user) suggested_users.remove_suggestion(user_id, suggested_user)
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def bulk_remove_instance_task(instance_id): def bulk_remove_instance_task(instance_id):
"""remove a bunch of users from recs""" """remove a bunch of users from recs"""
for user in models.User.objects.filter(federated_server__id=instance_id): for user in models.User.objects.filter(federated_server__id=instance_id):
@ -282,7 +282,7 @@ def bulk_remove_instance_task(instance_id):
) )
@app.task(queue=LOW, ignore_result=True) @app.task(queue=LOW)
def bulk_add_instance_task(instance_id): def bulk_add_instance_task(instance_id):
"""remove a bunch of users from recs""" """remove a bunch of users from recs"""
for user in models.User.objects.filter(federated_server__id=instance_id): for user in models.User.objects.filter(federated_server__id=instance_id):

View file

@ -115,7 +115,7 @@ def sometimes_async_activity_task(activity_json, queue=MEDIUM):
activity_task.apply_async(args=(activity_json,), queue=queue) activity_task.apply_async(args=(activity_json,), queue=queue)
@app.task(queue=MEDIUM, ignore_result=True) @app.task(queue=MEDIUM)
def activity_task(activity_json): def activity_task(activity_json):
"""do something with this json we think is legit""" """do something with this json we think is legit"""
# lets see if the activitypub module can make sense of this json # lets see if the activitypub module can make sense of this json