import hmac
import json
import logging
import threading
import time
from collections import deque, namedtuple
from operator import itemgetter
from django import db
from django.conf import settings
from django.utils import timezone
from django.db.models import F
from judge import event_poster as event
from judge.bridge.base_handler import ZlibPacketHandler, proxy_list
from judge.caching import finished_submission
from judge.models import (
Judge,
Language,
LanguageLimit,
Problem,
RuntimeVersion,
Submission,
SubmissionTestCase,
)
logger = logging.getLogger("judge.bridge")
json_log = logging.getLogger("judge.json.bridge")
UPDATE_RATE_LIMIT = 5
UPDATE_RATE_TIME = 0.5
SubmissionData = namedtuple(
"SubmissionData",
"time memory short_circuit pretests_only contest_no attempt_no user_id",
)
def _ensure_connection():
db.connection.close_if_unusable_or_obsolete()
class JudgeHandler(ZlibPacketHandler):
proxies = proxy_list(settings.BRIDGED_JUDGE_PROXIES or [])
def __init__(self, request, client_address, server, judges):
super().__init__(request, client_address, server)
self.judges = judges
self.handlers = {
"grading-begin": self.on_grading_begin,
"grading-end": self.on_grading_end,
"compile-error": self.on_compile_error,
"compile-message": self.on_compile_message,
"batch-begin": self.on_batch_begin,
"batch-end": self.on_batch_end,
"test-case-status": self.on_test_case,
"internal-error": self.on_internal_error,
"submission-terminated": self.on_submission_terminated,
"submission-acknowledged": self.on_submission_acknowledged,
"ping-response": self.on_ping_response,
"supported-problems": self.on_supported_problems,
"handshake": self.on_handshake,
}
self._working = False
self._no_response_job = None
self._problems = []
self.executors = {}
self.problems = {}
self.latency = None
self.time_delta = None
self.load = 1e100
self.name = None
self.batch_id = None
self.in_batch = False
self._stop_ping = threading.Event()
self._ping_average = deque(maxlen=6) # 1 minute average, just like load
self._time_delta = deque(maxlen=6)
# each value is (updates, last reset)
self.update_counter = {}
self.judge = None
self.judge_address = None
self._submission_cache_id = None
self._submission_cache = {}
def on_connect(self):
self.timeout = 15
logger.info("Judge connected from: %s", self.client_address)
json_log.info(self._make_json_log(action="connect"))
def on_disconnect(self):
self._stop_ping.set()
if self._working:
logger.error(
"Judge %s disconnected while handling submission %s",
self.name,
self._working,
)
self.judges.remove(self)
if self.name is not None:
self._disconnected()
logger.info(
"Judge disconnected from: %s with name %s", self.client_address, self.name
)
json_log.info(
self._make_json_log(action="disconnect", info="judge disconnected")
)
if self._working:
Submission.objects.filter(id=self._working).update(
status="IE", result="IE", error=""
)
json_log.error(
self._make_json_log(
sub=self._working,
action="close",
info="IE due to shutdown on grading",
)
)
def _authenticate(self, id, key):
try:
judge = Judge.objects.get(name=id, is_blocked=False)
except Judge.DoesNotExist:
result = False
else:
result = hmac.compare_digest(judge.auth_key, key)
if not result:
json_log.warning(
self._make_json_log(
action="auth", judge=id, info="judge failed authentication"
)
)
return result
def _connected(self):
judge = self.judge = Judge.objects.get(name=self.name)
judge.start_time = timezone.now()
judge.online = True
judge.problems.set(Problem.objects.filter(code__in=list(self.problems.keys())))
judge.runtimes.set(Language.objects.filter(key__in=list(self.executors.keys())))
# Delete now in case we somehow crashed and left some over from the last connection
RuntimeVersion.objects.filter(judge=judge).delete()
versions = []
for lang in judge.runtimes.all():
versions += [
RuntimeVersion(
language=lang,
name=name,
version=".".join(map(str, version)),
priority=idx,
judge=judge,
)
for idx, (name, version) in enumerate(self.executors[lang.key])
]
RuntimeVersion.objects.bulk_create(versions)
judge.last_ip = self.client_address[0]
judge.save()
self.judge_address = "[%s]:%s" % (
self.client_address[0],
self.client_address[1],
)
json_log.info(
self._make_json_log(
action="auth",
info="judge successfully authenticated",
executors=list(self.executors.keys()),
)
)
def _disconnected(self):
Judge.objects.filter(id=self.judge.id).update(online=False)
RuntimeVersion.objects.filter(judge=self.judge).delete()
def _update_ping(self):
try:
Judge.objects.filter(name=self.name).update(
ping=self.latency, load=self.load
)
except Exception as e:
# What can I do? I don't want to tie this to MySQL.
if (
e.__class__.__name__ == "OperationalError"
and e.__module__ == "_mysql_exceptions"
and e.args[0] == 2006
):
db.connection.close()
def send(self, data):
super().send(json.dumps(data, separators=(",", ":")))
def on_handshake(self, packet):
if "id" not in packet or "key" not in packet:
logger.warning("Malformed handshake: %s", self.client_address)
self.close()
return
if not self._authenticate(packet["id"], packet["key"]):
logger.warning("Authentication failure: %s", self.client_address)
self.close()
return
self.timeout = 60
self._problems = packet["problems"]
self.problems = dict(self._problems)
self.executors = packet["executors"]
self.name = packet["id"]
self.send({"name": "handshake-success"})
logger.info("Judge authenticated: %s (%s)", self.client_address, packet["id"])
self.judges.register(self)
threading.Thread(target=self._ping_thread).start()
self._connected()
def can_judge(self, problem, executor, judge_id=None):
return (
problem in self.problems
and executor in self.executors
and (not judge_id or self.name == judge_id)
)
@property
def working(self):
return bool(self._working)
def get_related_submission_data(self, submission):
_ensure_connection()
try:
(
pid,
time,
memory,
short_circuit,
lid,
is_pretested,
sub_date,
uid,
part_virtual,
part_id,
) = (
Submission.objects.filter(id=submission).values_list(
"problem__id",
"problem__time_limit",
"problem__memory_limit",
"problem__short_circuit",
"language__id",
"is_pretested",
"date",
"user__id",
"contest__participation__virtual",
"contest__participation__id",
)
).get()
except Submission.DoesNotExist:
logger.error("Submission vanished: %s", submission)
json_log.error(
self._make_json_log(
sub=self._working,
action="request",
info="submission vanished when fetching info",
)
)
return
attempt_no = (
Submission.objects.filter(
problem__id=pid,
contest__participation__id=part_id,
user__id=uid,
date__lt=sub_date,
)
.exclude(status__in=("CE", "IE"))
.count()
+ 1
)
try:
time, memory = (
LanguageLimit.objects.filter(problem__id=pid, language__id=lid)
.values_list("time_limit", "memory_limit")
.get()
)
except LanguageLimit.DoesNotExist:
pass
return SubmissionData(
time=time,
memory=memory,
short_circuit=short_circuit,
pretests_only=is_pretested,
contest_no=part_virtual,
attempt_no=attempt_no,
user_id=uid,
)
def disconnect(self, force=False):
if force:
# Yank the power out.
self.close()
else:
self.send({"name": "disconnect"})
def submit(self, id, problem, language, source):
data = self.get_related_submission_data(id)
self._working = id
self._no_response_job = threading.Timer(20, self._kill_if_no_response)
self.send(
{
"name": "submission-request",
"submission-id": id,
"problem-id": problem,
"language": language,
"source": source,
"time-limit": data.time,
"memory-limit": data.memory,
"short-circuit": data.short_circuit,
"meta": {
"pretests-only": data.pretests_only,
"in-contest": data.contest_no,
"attempt-no": data.attempt_no,
"user": data.user_id,
},
}
)
def _kill_if_no_response(self):
logger.error(
"Judge failed to acknowledge submission: %s: %s", self.name, self._working
)
self.close()
def on_timeout(self):
if self.name:
logger.warning("Judge seems dead: %s: %s", self.name, self._working)
def malformed_packet(self, exception):
logger.exception("Judge sent malformed packet: %s", self.name)
super(JudgeHandler, self).malformed_packet(exception)
def on_submission_processing(self, packet):
_ensure_connection()
id = packet["submission-id"]
if Submission.objects.filter(id=id).update(status="P", judged_on=self.judge):
event.post("sub_%s" % Submission.get_id_secret(id), {"type": "processing"})
self._post_update_submission(id, "processing")
json_log.info(self._make_json_log(packet, action="processing"))
else:
logger.warning("Unknown submission: %s", id)
json_log.error(
self._make_json_log(
packet, action="processing", info="unknown submission"
)
)
def on_submission_wrong_acknowledge(self, packet, expected, got):
json_log.error(
self._make_json_log(
packet, action="processing", info="wrong-acknowledge", expected=expected
)
)
Submission.objects.filter(id=expected).update(
status="IE", result="IE", error=None
)
Submission.objects.filter(id=got, status="QU").update(
status="IE", result="IE", error=None
)
def on_submission_acknowledged(self, packet):
if not packet.get("submission-id", None) == self._working:
logger.error(
"Wrong acknowledgement: %s: %s, expected: %s",
self.name,
packet.get("submission-id", None),
self._working,
)
self.on_submission_wrong_acknowledge(
packet, self._working, packet.get("submission-id", None)
)
self.close()
logger.info("Submission acknowledged: %d", self._working)
if self._no_response_job:
self._no_response_job.cancel()
self._no_response_job = None
self.on_submission_processing(packet)
def abort(self):
self.send({"name": "terminate-submission"})
def get_current_submission(self):
return self._working or None
def ping(self):
self.send({"name": "ping", "when": time.time()})
def on_packet(self, data):
try:
try:
data = json.loads(data)
if "name" not in data:
raise ValueError
except ValueError:
self.on_malformed(data)
else:
handler = self.handlers.get(data["name"], self.on_malformed)
handler(data)
except Exception:
logger.exception("Error in packet handling (Judge-side): %s", self.name)
self._packet_exception()
# You can't crash here because you aren't so sure about the judges
# not being malicious or simply malforms. THIS IS A SERVER!
def _packet_exception(self):
json_log.exception(
self._make_json_log(sub=self._working, info="packet processing exception")
)
def _submission_is_batch(self, id):
if not Submission.objects.filter(id=id).update(batch=True):
logger.warning("Unknown submission: %s", id)
def on_supported_problems(self, packet):
logger.info("%s: Updated problem list", self.name)
self._problems = packet["problems"]
self.problems = dict(self._problems)
if not self.working:
self.judges.update_problems(self)
self.judge.problems.set(
Problem.objects.filter(code__in=list(self.problems.keys()))
)
json_log.info(
self._make_json_log(action="update-problems", count=len(self.problems))
)
def on_grading_begin(self, packet):
logger.info("%s: Grading has begun on: %s", self.name, packet["submission-id"])
self.batch_id = None
if Submission.objects.filter(id=packet["submission-id"]).update(
status="G",
is_pretested=packet["pretested"],
current_testcase=1,
points=0,
batch=False,
judged_date=timezone.now(),
):
SubmissionTestCase.objects.filter(
submission_id=packet["submission-id"]
).delete()
event.post(
"sub_%s" % Submission.get_id_secret(packet["submission-id"]),
{"type": "grading-begin"},
)
self._post_update_submission(packet["submission-id"], "grading-begin")
json_log.info(self._make_json_log(packet, action="grading-begin"))
else:
logger.warning("Unknown submission: %s", packet["submission-id"])
json_log.error(
self._make_json_log(
packet, action="grading-begin", info="unknown submission"
)
)
def on_grading_end(self, packet):
logger.info("%s: Grading has ended on: %s", self.name, packet["submission-id"])
self._free_self(packet)
self.batch_id = None
try:
submission = Submission.objects.get(id=packet["submission-id"])
except Submission.DoesNotExist:
logger.warning("Unknown submission: %s", packet["submission-id"])
json_log.error(
self._make_json_log(
packet, action="grading-end", info="unknown submission"
)
)
return
time = 0
memory = 0
points = 0.0
total = 0
status = 0
status_codes = ["SC", "AC", "WA", "MLE", "TLE", "IR", "RTE", "OLE"]
batches = {} # batch number: (points, total)
for case in SubmissionTestCase.objects.filter(submission=submission):
time += case.time
if not case.batch:
points += case.points
total += case.total
else:
if case.batch in batches:
batches[case.batch][0] += case.points
batches[case.batch][1] += case.total
else:
batches[case.batch] = [case.points, case.total]
memory = max(memory, case.memory)
i = status_codes.index(case.status)
if i > status:
status = i
for i in batches:
points += batches[i][0]
total += batches[i][1]
points = points
total = total
submission.case_points = points
submission.case_total = total
problem = submission.problem
sub_points = round(points / total * problem.points if total > 0 else 0, 3)
if not problem.partial and sub_points != problem.points:
sub_points = 0
submission.status = "D"
submission.time = time
submission.memory = memory
submission.points = sub_points
submission.result = status_codes[status]
submission.save()
json_log.info(
self._make_json_log(
packet,
action="grading-end",
time=time,
memory=memory,
points=sub_points,
total=problem.points,
result=submission.result,
case_points=points,
case_total=total,
user=submission.user_id,
problem=problem.code,
finish=True,
)
)
submission.user._updating_stats_only = True
submission.user.calculate_points()
problem._updating_stats_only = True
problem.update_stats()
submission.update_contest()
finished_submission(submission)
event.post(
"sub_%s" % submission.id_secret,
{
"type": "grading-end",
"time": time,
"memory": memory,
"points": float(points),
"total": float(problem.points),
"result": submission.result,
},
)
if hasattr(submission, "contest"):
participation = submission.contest.participation
event.post("contest_%d" % participation.contest_id, {"type": "update"})
self._post_update_submission(submission.id, "grading-end", done=True)
def on_compile_error(self, packet):
logger.info(
"%s: Submission failed to compile: %s", self.name, packet["submission-id"]
)
self._free_self(packet)
if Submission.objects.filter(id=packet["submission-id"]).update(
status="CE", result="CE", error=packet["log"]
):
event.post(
"sub_%s" % Submission.get_id_secret(packet["submission-id"]),
{
"type": "compile-error",
"log": packet["log"],
},
)
self._post_update_submission(
packet["submission-id"], "compile-error", done=True
)
json_log.info(
self._make_json_log(
packet,
action="compile-error",
log=packet["log"],
finish=True,
result="CE",
)
)
else:
logger.warning("Unknown submission: %s", packet["submission-id"])
json_log.error(
self._make_json_log(
packet,
action="compile-error",
info="unknown submission",
log=packet["log"],
finish=True,
result="CE",
)
)
def on_compile_message(self, packet):
logger.info(
"%s: Submission generated compiler messages: %s",
self.name,
packet["submission-id"],
)
if Submission.objects.filter(id=packet["submission-id"]).update(
error=packet["log"]
):
event.post(
"sub_%s" % Submission.get_id_secret(packet["submission-id"]),
{"type": "compile-message"},
)
json_log.info(
self._make_json_log(packet, action="compile-message", log=packet["log"])
)
else:
logger.warning("Unknown submission: %s", packet["submission-id"])
json_log.error(
self._make_json_log(
packet,
action="compile-message",
info="unknown submission",
log=packet["log"],
)
)
def on_internal_error(self, packet):
try:
raise ValueError("\n\n" + packet["message"])
except ValueError:
logger.exception(
"Judge %s failed while handling submission %s",
self.name,
packet["submission-id"],
)
self._free_self(packet)
id = packet["submission-id"]
if Submission.objects.filter(id=id).update(
status="IE", result="IE", error=packet["message"]
):
event.post(
"sub_%s" % Submission.get_id_secret(id), {"type": "internal-error"}
)
self._post_update_submission(id, "internal-error", done=True)
json_log.info(
self._make_json_log(
packet,
action="internal-error",
message=packet["message"],
finish=True,
result="IE",
)
)
else:
logger.warning("Unknown submission: %s", id)
json_log.error(
self._make_json_log(
packet,
action="internal-error",
info="unknown submission",
message=packet["message"],
finish=True,
result="IE",
)
)
def on_submission_terminated(self, packet):
logger.info("%s: Submission aborted: %s", self.name, packet["submission-id"])
self._free_self(packet)
if Submission.objects.filter(id=packet["submission-id"]).update(
status="AB", result="AB"
):
event.post(
"sub_%s" % Submission.get_id_secret(packet["submission-id"]),
{"type": "aborted-submission"},
)
self._post_update_submission(
packet["submission-id"], "terminated", done=True
)
json_log.info(
self._make_json_log(packet, action="aborted", finish=True, result="AB")
)
else:
logger.warning("Unknown submission: %s", packet["submission-id"])
json_log.error(
self._make_json_log(
packet,
action="aborted",
info="unknown submission",
finish=True,
result="AB",
)
)
def on_batch_begin(self, packet):
logger.info("%s: Batch began on: %s", self.name, packet["submission-id"])
self.in_batch = True
if self.batch_id is None:
self.batch_id = 0
self._submission_is_batch(packet["submission-id"])
self.batch_id += 1
json_log.info(
self._make_json_log(packet, action="batch-begin", batch=self.batch_id)
)
def on_batch_end(self, packet):
self.in_batch = False
logger.info("%s: Batch ended on: %s", self.name, packet["submission-id"])
json_log.info(
self._make_json_log(packet, action="batch-end", batch=self.batch_id)
)
def on_test_case(
self,
packet,
max_feedback=SubmissionTestCase._meta.get_field("feedback").max_length,
):
logger.info(
"%s: %d test case(s) completed on: %s",
self.name,
len(packet["cases"]),
packet["submission-id"],
)
id = packet["submission-id"]
updates = packet["cases"]
max_position = max(map(itemgetter("position"), updates))
sum_points = sum(map(itemgetter("points"), updates))
if not Submission.objects.filter(id=id).update(
current_testcase=max_position + 1, points=F("points") + sum_points
):
logger.warning("Unknown submission: %s", id)
json_log.error(
self._make_json_log(
packet, action="test-case", info="unknown submission"
)
)
return
bulk_test_case_updates = []
for result in updates:
test_case = SubmissionTestCase(submission_id=id, case=result["position"])
status = result["status"]
if status & 4:
test_case.status = "TLE"
elif status & 8:
test_case.status = "MLE"
elif status & 64:
test_case.status = "OLE"
elif status & 2:
test_case.status = "RTE"
elif status & 16:
test_case.status = "IR"
elif status & 1:
test_case.status = "WA"
elif status & 32:
test_case.status = "SC"
else:
test_case.status = "AC"
test_case.time = result["time"]
test_case.memory = result["memory"]
test_case.points = result["points"]
test_case.total = result["total-points"]
test_case.batch = self.batch_id if self.in_batch else None
test_case.feedback = (result.get("feedback") or "")[:max_feedback]
test_case.extended_feedback = result.get("extended-feedback") or ""
test_case.output = result["output"]
bulk_test_case_updates.append(test_case)
json_log.info(
self._make_json_log(
packet,
action="test-case",
case=test_case.case,
batch=test_case.batch,
time=test_case.time,
memory=test_case.memory,
feedback=test_case.feedback,
extended_feedback=test_case.extended_feedback,
output=test_case.output,
points=test_case.points,
total=test_case.total,
status=test_case.status,
)
)
do_post = True
if id in self.update_counter:
cnt, reset = self.update_counter[id]
cnt += 1
if time.monotonic() - reset > UPDATE_RATE_TIME:
del self.update_counter[id]
else:
self.update_counter[id] = (cnt, reset)
if cnt > UPDATE_RATE_LIMIT:
do_post = False
if id not in self.update_counter:
self.update_counter[id] = (1, time.monotonic())
if do_post:
event.post(
"sub_%s" % Submission.get_id_secret(id),
{
"type": "test-case",
"id": max_position,
},
)
self._post_update_submission(id, state="test-case")
SubmissionTestCase.objects.bulk_create(bulk_test_case_updates)
def on_malformed(self, packet):
logger.error("%s: Malformed packet: %s", self.name, packet)
json_log.exception(
self._make_json_log(sub=self._working, info="malformed json packet")
)
def on_ping_response(self, packet):
end = time.time()
self._ping_average.append(end - packet["when"])
self._time_delta.append((end + packet["when"]) / 2 - packet["time"])
self.latency = sum(self._ping_average) / len(self._ping_average)
self.time_delta = sum(self._time_delta) / len(self._time_delta)
self.load = packet["load"]
self._update_ping()
def _free_self(self, packet):
self.judges.on_judge_free(self, packet["submission-id"])
def _ping_thread(self):
try:
while True:
self.ping()
if self._stop_ping.wait(10):
break
except Exception:
logger.exception("Ping error in %s", self.name)
self.close()
raise
def _make_json_log(self, packet=None, sub=None, **kwargs):
data = {
"judge": self.name,
"address": self.judge_address,
}
if sub is None and packet is not None:
sub = packet.get("submission-id")
if sub is not None:
data["submission"] = sub
data.update(kwargs)
return json.dumps(data)
def _post_update_submission(self, id, state, done=False):
if self._submission_cache_id == id:
data = self._submission_cache
else:
self._submission_cache = data = (
Submission.objects.filter(id=id)
.values(
"problem__is_public",
"contest_object__key",
"user_id",
"problem_id",
"status",
"language__key",
)
.get()
)
self._submission_cache_id = id
if data["problem__is_public"]:
event.post(
"submissions",
{
"type": "done-submission" if done else "update-submission",
"state": state,
"id": id,
"contest": data["contest_object__key"],
"user": data["user_id"],
"problem": data["problem_id"],
"status": data["status"],
"language": data["language__key"],
},
)
def on_cleanup(self):
db.connection.close()