Commit b18d4924 authored by Julien Palard's avatar Julien Palard
Browse files

POC with Celery.

parent 9e8b87ba
......@@ -3,5 +3,9 @@ from channels.routing import ProtocolTypeRouter, URLRouter
import website.routing
application = ProtocolTypeRouter(
{"websocket": AuthMiddlewareStack(URLRouter(website.routing.websocket_urlpatterns))}
{
"websocket": AuthMiddlewareStack(
URLRouter(website.routing.websocket_urlpatterns)
),
}
)
......@@ -33,6 +33,7 @@ SERVER_EMAIL = "team@example.com"
INSTALLED_APPS = [
"website",
"moulinette",
"registration",
"django.contrib.admin",
"django.contrib.auth",
......@@ -89,8 +90,8 @@ LOGGING = {
},
},
"loggers": {
"root": {"handlers": ["console"], "level": "INFO"},
"django": {"handlers": ["console", "mail_admins"], "level": "INFO"},
"website": {"handlers": ["console"], "level": "INFO"},
"django.server": {
"handlers": ["django.server"],
"level": "INFO",
......
import asyncio
from functools import partial
import os
import tempfile
from subprocess import Popen, PIPE
from subprocess import STDOUT, TimeoutExpired, DEVNULL
from server import FIREJAIL_OPTIONS
from celery import shared_task
FIREJAIL_OPTIONS = [
......@@ -64,3 +65,54 @@ def run_snippet(source_code):
except TimeoutExpired:
prof_proc.kill()
return 1, "Timed out after 20 seconds."
@shared_task
def check_answer_task(answer: dict):
"""Executed on Celery workers.
"""
with tempfile.TemporaryDirectory(prefix="hkis_snippets") as tmpdir:
with open(os.path.join(tmpdir, "snippet.py"), "w") as snippet_file:
snippet_file.write(answer["source_code"])
firejail_env = os.environ.copy()
firejail_env["PATH"] = (
os.path.expanduser("~/.local/bin") + ":" + firejail_env["PATH"]
)
prof_proc = Popen(
["firejail"]
+ FIREJAIL_OPTIONS
+ ["--private=" + tmpdir, "python3", os.path.expanduser("~/snippet.py"),],
stdin=DEVNULL,
stdout=PIPE,
stderr=STDOUT,
cwd=os.path.expanduser("~/"),
env=firejail_env,
)
try:
stdout = prof_proc.communicate(timeout=20)[0]
return {
"return_code": prof_proc.returncode,
"correction_message": stdout.decode("UTF-8", "backslashreplace")[
:500_000
],
}
except TimeoutExpired:
prof_proc.kill()
return {
"return_code": 1,
"correction_message": "Timed out after 20 seconds.",
}
async def check_answer(answer: dict):
"""Executed Django side (`runworker moulinette` precisely).
TODO with Celery 5: should no longer need run_in_executor.
"""
def sync_celery_check_answer(answer: dict):
return check_answer_task.apply_async((answer,), expires=60).get()
return await asyncio.get_running_loop().run_in_executor(
None, partial(sync_celery_check_answer, answer=answer)
)
from website.models import Answer, Exercise
from django.contrib.auth.models import User
from django.utils.timezone import now
from moulinette.tasks import check_answer
import logging
from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from channels.consumer import AsyncConsumer
logger = logging.getLogger(__name__)
# Channels reminders:
# - The following class is instanciated once per websocket connection
# (once per browser tab), it's the lifespan of a what channels call a scope.
# - Group can group together multiple scopes, usefull to send a
# message to all browser tabs of a given user at once for example.
@database_sync_to_async
def db_store_correction(answer, is_valid, message):
answer = Answer.get(answer["id"])
answer.is_valid = is_valid
answer.correction_message = message
answer.save()
@database_sync_to_async
def db_create_answer(exercise: Exercise, user: User, source_code):
return exercise.answers.create(source_code=source_code, user=user)
@database_sync_to_async
def db_get_exercise(exercise_id: int):
return Exercise.objects.get(id=exercise_id)
@database_sync_to_async
def db_update_answer(answer_id: int, correction_message: str):
answer = Answer.objects.get(id=answer_id)
answer.correction_message = correction_message
answer.is_corrected = True # TODO
answer.is_valid = True # TODO
answer.corrected_at = now()
answer.save()
return answer
class ExerciseConsumer(AsyncJsonWebsocketConsumer):
def log(self, message, *args):
logger.info(
f"WS ({self.scope['user'].username} on {self.exercise.id}) {message}",
*args,
)
async def connect(self):
self.exercise = await db_get_exercise(
self.scope["url_route"]["kwargs"]["exercise_id"]
)
self.log("connect")
await self.accept()
async def disconnect(self, close_code):
self.log("disconnect")
async def answer(self, answer):
"""A browser sent us a new answer.
"""
self.log("submit answer")
answer_obj = await db_create_answer(
self.exercise, self.scope["user"], answer["source_code"]
)
result = await check_answer(answer)
answer_obj = await db_update_answer(
answer_obj.id, correction_message=result["correction_message"]
)
self.log("got result from moulinette")
await self.send_json(
{
"type": "correction",
"exercise": self.exercise.id,
"correction_message": result["correction_message"],
"answer_id": answer_obj.id,
"is_corrected": answer_obj.is_corrected,
"is_valid": answer_obj.is_valid,
}
)
async def receive_json(self, content):
if content["type"] == "answer":
await self.answer(content)
if content["type"] == "snippet":
await self.snippet(content)
async def snippet(self, snippet):
self.log("run snippet")
await self.send_json(snippet)
class AnswerConsumer(AsyncJsonWebsocketConsumer):
async def connect(self):
......
......@@ -9,7 +9,7 @@ msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2020-06-29 22:52+0200\n"
"PO-Revision-Date: 2020-06-29 23:31+0200\n"
"PO-Revision-Date: 2020-07-14 23:55+0200\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"
"Language: \n"
......@@ -54,7 +54,7 @@ msgstr "Exécuter"
#: website/templates/hkis/exercise.html:183
msgid "Submit for validation"
msgstr "Envoi pour valiation"
msgstr "Envoi pour validation"
#: website/templates/hkis/exercise.html:187
msgid "Other students solutions"
......
......@@ -2,8 +2,5 @@ from django.conf.urls import url
from . import consumers
websocket_urlpatterns = [
url(
r"^ws/answers/(?P<user_id>[0-9]+)/(?P<exercise_id>[0-9]+)/$",
consumers.AnswerConsumer,
)
url(r"^ws/exercises/(?P<exercise_id>[0-9]+)/", consumers.ExerciseConsumer),
]
......@@ -117,26 +117,34 @@
}
function websocket_connect(ws_location) {
console.log("Trying to open WebSocket connection.");
var ws = new WebSocket(ws_location);
window.ws = ws;
ws.onmessage = function(e) {
var data = JSON.parse(e.data);
console.log("WebSocket recv:", data);
if (data.type == "snippet")
fill_snippet(data.id, data.executed_at, data.output)
if (data.type == "correction")
fill_answer(data.answer, data.is_corrected, data.is_valid, data.correction_message);
fill_answer(data.answer_id, data.is_corrected, data.is_valid, data.correction_message);
};
ws.onclose = function() {
ws = null;
// Try to reconnect in 5 seconds
console.log("WebSocket closed, will retry to connect in 5s");
setTimeout(function(){websocket_connect(ws_location)}, 5000);
};
}
function ws_submit_answer(form) {
document.getElementById('submit_answer').setAttribute('disabled', true);
window.ws.send(JSON.stringify({"type": "answer", "source_code": form["source_code"].value}));
}
var ws_protocol = window.location.protocol == "http:" ? "ws:" : "wss:";
websocket_connect(ws_protocol + "//" + window.location.host +
"/ws/answers/" + {{ user.id }} + "/" + {{object.id}} + "/");
"/ws/exercises/" + {{object.id}} + "/");
</script>
{% endautoescape %}
{% endblock %}
......@@ -156,7 +164,7 @@
<div class="row">
<div class="col-md-6">
<h2>{% trans "Your Code" %}</h2>
<form id="answer_form" action="/api/answers/" method="post" onsubmit="document.getElementById('submit_answer').setAttribute('disabled', true); AJAXSubmit(this); return false;">{% csrf_token %}
<form id="answer_form" action="#" method="post" onsubmit="ws_submit_answer(this); return false;">{% csrf_token %}
{% for field in answer_form %}
<div class="fieldWrapper">
{{ field }}
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment