[WIP] Split Queue handling and DB writing to address #1216

cms/__init__.py

@@ -39,7 +39,7 @@
     "Address", "ServiceCoord", "ConfigError", "async_config", "config",
     # util
     "mkdir", "rmtree", "utf8_decoder", "get_safe_shard", "get_service_address",
-    "get_service_shards", "contest_id_from_args", "default_argument_parser",
+    "random_service", "get_service_shards", "contest_id_from_args", "default_argument_parser",
     # plugin
     "plugin_list",
 ]
@@ -60,6 +60,7 @@
 # Acts as a "none of the above".
 TOKEN_MODE_MIXED = "mixed"
 
+
 # Feedback level.
 
 # Full information (killing signals, time and memory, status for all
@@ -73,5 +74,6 @@
 from .conf import Address, ServiceCoord, ConfigError, async_config, config
 from .util import mkdir, rmtree, utf8_decoder, get_safe_shard, \
     get_service_address, get_service_shards, contest_id_from_args, \
-    default_argument_parser
+    random_service, default_argument_parser
 from .plugin import plugin_list
+

cms/server/admin/rpc_authorization.py

@@ -27,8 +27,8 @@
 RPCS_ALLOWED_FOR_AUTHENTICATED = [
     ("AdminWebServer", "submissions_status"),
     ("ResourceService", "get_resources"),
-    ("EvaluationService", "workers_status"),
-    ("EvaluationService", "queue_status"),
+    ("QueueService", "workers_status"),
+    ("QueueService", "queue_status"),
     ("LogService", "last_messages"),
 ]
 
@@ -39,9 +39,9 @@
 RPCS_ALLOWED_FOR_ALL = RPCS_ALLOWED_FOR_MESSAGING + [
     ("ResourceService", "kill_service"),
     ("ResourceService", "toggle_autorestart"),
-    ("EvaluationService", "enable_worker"),
-    ("EvaluationService", "disable_worker"),
-    ("EvaluationService", "invalidate_submission"),
+    ("QueueService", "enable_worker"),
+    ("QueueService", "disable_worker"),
+    ("QueueService", "invalidate_submission"),
     ("ScoringService", "invalidate_submission"),
 ]
 

cms/server/admin/server.py

@@ -73,7 +73,10 @@ def __init__(self, shard):
 
         self.admin_web_server = self.connect_to(
             ServiceCoord("AdminWebServer", 0))
-        self.evaluation_service = self.connect_to(
+        self.queue_service = self.connect_to(
+            ServiceCoord("QueueService", 0))
+        # TODO: does it make sense to use a random one?
+        self.evaluation_services = self.connect_to(
             ServiceCoord("EvaluationService", 0))
         self.scoring_service = self.connect_to(
             ServiceCoord("ScoringService", 0))

cms/server/admin/templates/macro/reevaluation_buttons.html

@@ -36,7 +36,7 @@
   {% set invalidate_arguments = {"contest_id": contest_id} %}
 {% endif %}
 <button onclick="cmsrpc_request(
-                 'EvaluationService', 0,
+                 'QueueService', 0,
                  'invalidate_submission', {
 {% for key, value in invalidate_arguments.items() %}
                  '{{ key }}': {{ value }},
@@ -49,7 +49,7 @@
 {% endif %}
         title="Compilation" >C</button>
 <button onclick="cmsrpc_request(
-                 'EvaluationService', 0,
+                 'QueueService', 0,
                  'invalidate_submission', {
 {% for key, value in invalidate_arguments.items() %}
                  '{{ key }}': {{ value }},

cms/server/admin/templates/overview.html

@@ -63,11 +63,11 @@
 
 function enable_worker(shard) {
     if (confirm("Do you really want to enable worker " + shard + "?")) {
-        cmsrpc_request("EvaluationService", 0,
+        cmsrpc_request("QueueService", 0,
                        "enable_worker",
                        {"shard":shard},
                        function() {
-                          cmsrpc_request("EvaluationService", 0,
+                          cmsrpc_request("QueueService", 0,
                                          "workers_status",
                                          {},
                                          update_workers_status);
@@ -77,11 +77,11 @@
 
 function disable_worker(shard) {
     if (confirm("Do you really want to disable worker " + shard + "?")) {
-        cmsrpc_request("EvaluationService", 0,
+        cmsrpc_request("QueueService", 0,
                        "disable_worker",
                        {"shard":shard},
                        function() {
-                          cmsrpc_request("EvaluationService", 0,
+                          cmsrpc_request("QueueService", 0,
                                          "workers_status",
                                          {},
                                          update_workers_status);
@@ -214,12 +214,12 @@
     if (!update_statuses.queue_request
             || update_statuses.queue_request.state() != "pending") {
         update_statuses.queue_request =
-            cmsrpc_request("EvaluationService", 0,
+            cmsrpc_request("QueueService", 0,
                            "queue_status",
                            {},
                            update_queue_status);
     }
-    cmsrpc_request("EvaluationService", 0,
+    cmsrpc_request("QueueService", 0,
                    "workers_status",
                    {},
                    update_workers_status);

cms/server/contest/handlers/tasksubmission.py

@@ -38,7 +38,8 @@
     import tornado.web as tornado_web
 from sqlalchemy.orm import joinedload
 
-from cms import config, FEEDBACK_LEVEL_FULL
+
+from cms import config, random_service, FEEDBACK_LEVEL_FULL
 from cms.db import Submission, SubmissionResult
 from cms.grading.languagemanager import get_language
 from cms.grading.scoring import task_score
@@ -90,8 +91,8 @@ def post(self, task_name):
             logger.info("Sent error: `%s' - `%s'", e.subject, e.formatted_text)
             self.notify_error(e.subject, e.text, e.text_params)
         else:
-            self.service.evaluation_service.new_submission(
-                submission_id=submission.id)
+            random_service(self.service.evaluation_services)\
+                .new_submission(submission_id=submission.id)
             self.notify_success(N_("Submission received"),
                                 N_("Your submission has been received "
                                    "and is currently being evaluated."))
@@ -104,7 +105,6 @@ def post(self, task_name):
         self.redirect(self.contest_url("tasks", task.name, "submissions",
                                        **query_args))
 
-
 class TaskSubmissionsHandler(ContestHandler):
     """Shows the data of a task in the contest.
 

cms/server/contest/handlers/taskusertest.py

@@ -36,7 +36,8 @@
 except ImportError:
     import tornado.web as tornado_web
 
-from cms import config
+
+from cms import config, random_service
 from cms.db import UserTest, UserTestResult
 from cms.grading.languagemanager import get_language
 from cms.server import multi_contest
@@ -143,8 +144,8 @@ def post(self, task_name):
             logger.info("Sent error: `%s' - `%s'", e.subject, e.formatted_text)
             self.notify_error(e.subject, e.text, e.text_params)
         else:
-            self.service.evaluation_service.new_user_test(
-                user_test_id=user_test.id)
+            random_service(self.service.evaluation_services) \
+                .new_user_test(user_test_id=user_test.id)
             self.notify_success(N_("Test received"),
                                 N_("Your test has been received "
                                    "and is currently being executed."))

cms/server/contest/server.py

@@ -41,7 +41,7 @@
 
 from werkzeug.middleware.shared_data import SharedDataMiddleware
 
-from cms import ConfigError, ServiceCoord, config
+from cms import ConfigError, ServiceCoord, config, get_service_shards
 from cms.io import WebService
 from cms.locale import get_translations
 from cms.server.contest.jinja2_toolbox import CWS_ENVIRONMENT
@@ -116,8 +116,10 @@ def __init__(self, shard, contest_id=None):
         # Retrieve the available translations.
         self.translations = get_translations()
 
-        self.evaluation_service = self.connect_to(
-            ServiceCoord("EvaluationService", 0))
+        self.evaluation_services = [
+            self.connect_to(ServiceCoord("EvaluationService", i))
+            for i in range(get_service_shards("EvaluationService"))]
+
         self.scoring_service = self.connect_to(
             ServiceCoord("ScoringService", 0))