Skip to content

Commit 9b81155

Browse files
patrick91patrick91
committed
Move ML recap analysis to heavy-processing Celery worker
Offload compute_similar_talks and compute_topic_clusters to the heavy_processing Celery queue to prevent OOM in Gunicorn web workers. The admin view now checks cache first, dispatches a Celery task on miss, and returns {"status": "processing"} for the frontend to poll.
1 parent 4883f6c commit 9b81155

4 files changed

Lines changed: 246 additions & 103 deletions

File tree

‎backend/reviews/admin.py‎

Lines changed: 19 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -448,49 +448,33 @@ def review_recap_compute_analysis_view(self, request, review_session_id):
448448
raise PermissionDenied()
449449

450450
conference = review_session.conference
451-
accepted_submissions = self._get_accepted_submissions(conference)
451+
accepted_submissions = list(self._get_accepted_submissions(conference))
452452
force_recompute = request.GET.get("recompute") == "1"
453453

454-
from reviews.similar_talks import compute_similar_talks, compute_topic_clusters
454+
from django.core.cache import cache
455455

456-
similar_talks = compute_similar_talks(
457-
accepted_submissions,
458-
top_n=5,
459-
conference_id=conference.id,
460-
force_recompute=force_recompute,
461-
)
456+
from pycon.tasks import check_pending_heavy_processing_work
457+
from reviews.similar_talks import _get_cache_key
458+
from reviews.tasks import compute_recap_analysis
462459

463-
topic_clusters = compute_topic_clusters(
464-
accepted_submissions,
465-
min_topic_size=3,
466-
conference_id=conference.id,
467-
force_recompute=force_recompute,
460+
combined_cache_key = _get_cache_key(
461+
"recap_analysis", conference.id, accepted_submissions
468462
)
469463

470-
# Build submissions list with similar talks, sorted by highest similarity
471-
submissions_list = sorted(
472-
[
473-
{
474-
"id": s.id,
475-
"title": str(s.title),
476-
"type": s.type.name,
477-
"speaker": s.speaker.display_name if s.speaker else "Unknown",
478-
"similar": similar_talks.get(s.id, []),
479-
}
480-
for s in accepted_submissions
481-
],
482-
key=lambda x: max(
483-
(item["similarity"] for item in x["similar"]), default=0
484-
),
485-
reverse=True,
486-
)
464+
if not force_recompute:
465+
cached_result = cache.get(combined_cache_key)
466+
if cached_result is not None:
467+
return JsonResponse(cached_result)
487468

488-
return JsonResponse(
489-
{
490-
"submissions_list": submissions_list,
491-
"topic_clusters": topic_clusters,
492-
}
469+
# Dispatch the Celery task to the heavy_processing queue
470+
compute_recap_analysis.apply_async(
471+
args=[conference.id],
472+
kwargs={"force_recompute": force_recompute},
473+
queue="heavy_processing",
493474
)
475+
check_pending_heavy_processing_work.delay()
476+
477+
return JsonResponse({"status": "processing"})
494478

495479
def review_view(self, request, review_session_id, review_item_id):
496480
review_session = ReviewSession.objects.get(id=review_session_id)

‎backend/reviews/tasks.py‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
import logging
2+
3+
from pycon.celery import app
4+
5+
logger = logging.getLogger(__name__)
6+
7+
8+
@app.task
9+
def compute_recap_analysis(conference_id, force_recompute=False):
10+
from django.core.cache import cache
11+
from django.db.models import Q
12+
13+
from reviews.similar_talks import (
14+
_get_cache_key,
15+
compute_similar_talks,
16+
compute_topic_clusters,
17+
)
18+
from submissions.models import Submission
19+
20+
accepted_submissions = list(
21+
Submission.objects.filter(conference_id=conference_id)
22+
.filter(
23+
Q(pending_status=Submission.STATUS.accepted)
24+
| Q(pending_status__isnull=True, status=Submission.STATUS.accepted)
25+
| Q(pending_status="", status=Submission.STATUS.accepted)
26+
)
27+
.select_related("speaker", "type", "audience_level")
28+
.prefetch_related("languages")
29+
)
30+
31+
similar_talks = compute_similar_talks(
32+
accepted_submissions,
33+
top_n=5,
34+
conference_id=conference_id,
35+
force_recompute=force_recompute,
36+
)
37+
38+
topic_clusters = compute_topic_clusters(
39+
accepted_submissions,
40+
min_topic_size=3,
41+
conference_id=conference_id,
42+
force_recompute=force_recompute,
43+
)
44+
45+
submissions_list = sorted(
46+
[
47+
{
48+
"id": s.id,
49+
"title": str(s.title),
50+
"type": s.type.name,
51+
"speaker": s.speaker.display_name if s.speaker else "Unknown",
52+
"similar": similar_talks.get(s.id, []),
53+
}
54+
for s in accepted_submissions
55+
],
56+
key=lambda x: max(
57+
(item["similarity"] for item in x["similar"]), default=0
58+
),
59+
reverse=True,
60+
)
61+
62+
result = {
63+
"submissions_list": submissions_list,
64+
"topic_clusters": topic_clusters,
65+
}
66+
67+
combined_cache_key = _get_cache_key(
68+
"recap_analysis", conference_id, accepted_submissions
69+
)
70+
cache.set(combined_cache_key, result, 60 * 60 * 24)
71+
72+
return result

‎backend/reviews/templates/reviews-recap.html‎

Lines changed: 70 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -570,6 +570,70 @@ <h2 class="recap-section-title">🔗 Similar Talks</h2>
570570
section.style.display = '';
571571
}
572572

573+
var pollTimer = null;
574+
var pollStartTime = null;
575+
var POLL_INTERVAL = 3000;
576+
var POLL_TIMEOUT = 120000;
577+
578+
function stopPolling() {
579+
if (pollTimer) {
580+
clearTimeout(pollTimer);
581+
pollTimer = null;
582+
}
583+
pollStartTime = null;
584+
}
585+
586+
function handleResult(data, recompute) {
587+
loading.style.display = 'none';
588+
btn.style.display = 'none';
589+
recomputeBtn.style.display = '';
590+
recomputeBtn.disabled = false;
591+
recomputeBtn.textContent = 'Recompute (ignore cache)';
592+
593+
renderTopicClusters(data.topic_clusters);
594+
renderSimilarTalks(data.submissions_list);
595+
}
596+
597+
function pollForResults(recompute) {
598+
if (pollStartTime && (Date.now() - pollStartTime) > POLL_TIMEOUT) {
599+
stopPolling();
600+
loading.style.display = 'none';
601+
var activeBtn = recompute ? recomputeBtn : btn;
602+
activeBtn.disabled = false;
603+
activeBtn.textContent = recompute ? 'Recompute (ignore cache)' : 'Compute Topic Clusters & Similar Talks';
604+
errorDiv.textContent = 'Analysis is taking longer than expected. Please try again later.';
605+
errorDiv.style.display = '';
606+
return;
607+
}
608+
609+
var url = recompute ? computeUrl + '?recompute=1' : computeUrl;
610+
611+
fetch(url, {
612+
headers: { 'X-Requested-With': 'XMLHttpRequest' }
613+
})
614+
.then(function(response) {
615+
if (!response.ok) throw new Error('Server error: ' + response.status);
616+
return response.json();
617+
})
618+
.then(function(data) {
619+
if (data.status === 'processing') {
620+
pollTimer = setTimeout(function() { pollForResults(false); }, POLL_INTERVAL);
621+
return;
622+
}
623+
stopPolling();
624+
handleResult(data, recompute);
625+
})
626+
.catch(function(err) {
627+
stopPolling();
628+
loading.style.display = 'none';
629+
var activeBtn = recompute ? recomputeBtn : btn;
630+
activeBtn.disabled = false;
631+
activeBtn.textContent = recompute ? 'Recompute (ignore cache)' : 'Compute Topic Clusters & Similar Talks';
632+
errorDiv.textContent = 'Failed to compute analysis: ' + err.message;
633+
errorDiv.style.display = '';
634+
});
635+
}
636+
573637
function fetchAnalysis(recompute) {
574638
var url = recompute ? computeUrl + '?recompute=1' : computeUrl;
575639
var activeBtn = recompute ? recomputeBtn : btn;
@@ -587,14 +651,12 @@ <h2 class="recap-section-title">🔗 Similar Talks</h2>
587651
return response.json();
588652
})
589653
.then(function(data) {
590-
loading.style.display = 'none';
591-
btn.style.display = 'none';
592-
recomputeBtn.style.display = '';
593-
recomputeBtn.disabled = false;
594-
recomputeBtn.textContent = 'Recompute (ignore cache)';
595-
596-
renderTopicClusters(data.topic_clusters);
597-
renderSimilarTalks(data.submissions_list);
654+
if (data.status === 'processing') {
655+
pollStartTime = Date.now();
656+
pollTimer = setTimeout(function() { pollForResults(false); }, POLL_INTERVAL);
657+
return;
658+
}
659+
handleResult(data, recompute);
598660
})
599661
.catch(function(err) {
600662
loading.style.display = 'none';

0 commit comments

Comments
 (0)