Reimplement scoring and create connection abstraction

Co-authored-by: Felix Auringer <felix.auringer@student.hpi.uni-potsdam.de>

app/controllers/concerns/submission_scoring.rb

@@ -1,57 +1,48 @@
 # frozen_string_literal: true
 
-require 'concurrent/future'
-
 module SubmissionScoring
-  def collect_test_results(submission)
+  def test_result(output, file)
+    submission = self
     # Mnemosyne.trace 'custom.codeocean.collect_test_results', meta: { submission: submission.id } do
-    futures = submission.collect_files.select(&:teacher_defined_assessment?).map do |file|
-      Concurrent::Future.execute do
-        # Mnemosyne.trace 'custom.codeocean.collect_test_results_block', meta: { file: file.id, submission: submission.id } do
-        assessor = Assessor.new(execution_environment: submission.execution_environment)
-        output = execute_test_file(file, submission)
-        assessment = assessor.assess(output)
-        passed = ((assessment[:passed] == assessment[:count]) and (assessment[:score]).positive?)
-        testrun_output = passed ? nil : "status: #{output[:status]}\n stdout: #{output[:stdout]}\n stderr: #{output[:stderr]}"
-        if testrun_output.present?
-          submission.exercise.execution_environment.error_templates.each do |template|
-            pattern = Regexp.new(template.signature).freeze
-            StructuredError.create_from_template(template, testrun_output, submission) if pattern.match(testrun_output)
-          end
-        end
-        testrun = Testrun.create(
-          submission: submission,
-          cause: 'assess', # Required to differ run and assess for RfC show
-          file: file, # Test file that was executed
-          passed: passed,
-          output: testrun_output,
-          container_execution_time: output[:container_execution_time],
-          waiting_for_container_time: output[:waiting_for_container_time]
-        )
-
-        filename = file.name_with_extension
-
-        if file.teacher_defined_linter?
-          LinterCheckRun.create_from(testrun, assessment)
-          switch_locale do
-            assessment = assessor.translate_linter(assessment, I18n.locale)
-
-            # replace file name with hint if linter is not used for grading. Refactor!
-            filename = t('exercises.implement.not_graded') if file.weight.zero?
-          end
-        end
-
-        output.merge!(assessment)
-        output.merge!(filename: filename, message: feedback_message(file, output), weight: file.weight)
-        # end
+    # Mnemosyne.trace 'custom.codeocean.collect_test_results_block', meta: { file: file.id, submission: submission.id } do
+    assessor = Assessor.new(execution_environment: submission.execution_environment)
+    assessment = assessor.assess(output)
+    passed = ((assessment[:passed] == assessment[:count]) and (assessment[:score]).positive?)
+    testrun_output = passed ? nil : "status: #{output[:status]}\n stdout: #{output[:stdout]}\n stderr: #{output[:stderr]}"
+    if testrun_output.present?
+      submission.exercise.execution_environment.error_templates.each do |template|
+        pattern = Regexp.new(template.signature).freeze
+        StructuredError.create_from_template(template, testrun_output, submission) if pattern.match(testrun_output)
       end
     end
-    futures.map(&:value!)
+    testrun = Testrun.create(
+      submission: submission,
+      cause: 'assess', # Required to differ run and assess for RfC show
+      file: file, # Test file that was executed
+      passed: passed,
+      output: testrun_output,
+      container_execution_time: output[:container_execution_time],
+      waiting_for_container_time: output[:waiting_for_container_time]
+    )
+
+    filename = file.name_with_extension
+
+    if file.teacher_defined_linter?
+      LinterCheckRun.create_from(testrun, assessment)
+      assessment = assessor.translate_linter(assessment, I18n.locale)
+
+          # replace file name with hint if linter is not used for grading. Refactor!
+          filename = t('exercises.implement.not_graded') if file.weight.zero?
+        end
+
+    output.merge!(assessment)
+    output.merge!(filename: filename, message: feedback_message(file, output), weight: file.weight)
   end
 
   private :collect_test_results
 
   def execute_test_file(file, submission)
+    # TODO: replace DockerClient here
     DockerClient.new(execution_environment: file.context.execution_environment).execute_test_command(submission,
       file.name_with_extension)
   end
@@ -59,19 +50,19 @@ module SubmissionScoring
   private :execute_test_file
 
   def feedback_message(file, output)
-    switch_locale do
-      if output[:score] == Assessor::MAXIMUM_SCORE && output[:file_role] == 'teacher_defined_test'
-        I18n.t('exercises.implement.default_test_feedback')
-      elsif output[:score] == Assessor::MAXIMUM_SCORE && output[:file_role] == 'teacher_defined_linter'
-        I18n.t('exercises.implement.default_linter_feedback')
-      else
-        render_markdown(file.feedback_message)
-      end
+    # set_locale
+    if output[:score] == Assessor::MAXIMUM_SCORE && output[:file_role] == 'teacher_defined_test'
+      I18n.t('exercises.implement.default_test_feedback')
+    elsif output[:score] == Assessor::MAXIMUM_SCORE && output[:file_role] == 'teacher_defined_linter'
+      I18n.t('exercises.implement.default_linter_feedback')
+    else
+      # render_markdown(file.feedback_message)
     end
   end
 
-  def score_submission(submission)
-    outputs = collect_test_results(submission)
+  def score_submission(outputs)
+    # outputs = collect_test_results(submission)
+    submission = self
     score = 0.0
     if outputs.present?
       outputs.each do |output|

app/controllers/submissions_controller.rb

@@ -137,18 +137,17 @@ class SubmissionsController < ApplicationController
     end
   end
 
-  def handle_websockets(tubesock, container)
-    socket = container.socket
+  def handle_websockets(tubesock, container, socket)
     tubesock.send_data JSON.dump({'cmd' => 'status', 'status' => :container_running})
     @waiting_for_container_time = Time.zone.now - @container_request_time
     @execution_request_time = Time.zone.now
 
-    socket.on :message do |event|
-      Rails.logger.info("#{Time.zone.now.getutc}: Docker sending: #{event.data}")
-      handle_message(event.data, tubesock)
+    socket.on :message do |data|
+      Rails.logger.info("#{Time.zone.now.getutc}: Docker sending: #{data}")
+      handle_message(data, tubesock)
     end
 
-    socket.on :close do |_event|
+    socket.on :exit do |_exit_code|
       EventMachine.stop_event_loop
       tubesock.send_data JSON.dump({'cmd' => 'timeout'}) if container.status == 'timeouted'
       kill_socket(tubesock)
@@ -184,8 +183,8 @@ class SubmissionsController < ApplicationController
           return
         end
         @container_request_time = Time.zone.now
-        @submission.run(sanitize_filename) do |container|
-          handle_websockets(tubesock, container)
+        @submission.run(sanitize_filename) do |container, socket|
+          handle_websockets(tubesock, container, socket)
         end
       end
     ensure
@@ -330,33 +329,19 @@ class SubmissionsController < ApplicationController
   end
 
   def score
-    hijack do |tubesock|
-      if @embed_options[:disable_score]
-        kill_socket(tubesock)
-        return
-      end
-
-      unless EventMachine.reactor_running? && EventMachine.reactor_thread.alive?
-        Thread.new do
-          EventMachine.run
-        ensure
-          ActiveRecord::Base.connection_pool.release_connection
+    Thread.new do
+      hijack do |tubesock|
+        if @embed_options[:disable_run]
+          return kill_socket(tubesock)
         end
-      end
-      # tubesock is the socket to the client
-
-      # the score_submission call will end up calling docker exec, which is blocking.
-      # to ensure responsiveness, we therefore open a thread here.
-      Thread.new do
-        tubesock.send_data JSON.dump(score_submission(@submission))
-
+        tubesock.send_data(@submission.calculate_score)
         # To enable hints when scoring a submission, uncomment the next line:
         # send_hints(tubesock, StructuredError.where(submission: @submission))
 
         tubesock.send_data JSON.dump({'cmd' => 'exit'})
-      ensure
-        ActiveRecord::Base.connection_pool.release_connection
       end
+    ensure
+      ActiveRecord::Base.connection_pool.release_connection
     end
   end
 

app/models/submission.rb

@@ -4,6 +4,9 @@ class Submission < ApplicationRecord
   include Context
   include Creation
   include ActionCableHelper
+  include SubmissionScoring
+
+  require 'concurrent/future'
 
   CAUSES = %w[assess download file render run save submit test autosave requestComments remoteAssess
               remoteSubmit].freeze
@@ -136,41 +139,61 @@ class Submission < ApplicationRecord
     end
   end
 
-  def score(file)
-    score_command = command_for execution_environment.test_command, file
-    container = run_command_with_self score_command
-    container
-    # Todo receive websocket data and pass it to some score function
+  def calculate_score
+    score = nil
+    prepared_container do |container|
+      scores = collect_files.select(&:teacher_defined_assessment?).map do |file|
+        score_command = command_for execution_environment.test_command, file.name_with_extension
+        stdout = ""
+        stderr = ""
+        exit_code = 0
+        container.execute_interactively(score_command) do |container, socket|
+          socket.on :stderr do
+            |data| stderr << data
+          end
+          socket.on :stdout do
+            |data| stdout << data
+          end
+          socket.on :close do |_exit_code|
+            exit_code = _exit_code
+            EventMachine.stop_event_loop
+          end
+        end
+        output = {
+          file_role: file.role,
+          waiting_for_container_time: 1.second, # TODO
+          container_execution_time: 1.second, # TODO
+          status: (exit_code == 0) ? :ok : :failed,
+          stdout: stdout,
+          stderr: stderr,
+        }
+        test_result(output, file)
+      end
+      score = score_submission(scores)
+    end
+    JSON.dump(score)
   end
 
   def run(file, &block)
     run_command = command_for execution_environment.run_command, file
-    execute_interactively(run_command, &block)
-  end
-
-  def run_tests(file, &block)
-    test_command = command_for execution_environment.test_command, file
-    execute_interactively(test_command, &block)
-  end
-
-  def execute_interactively(command)
-    container = nil
-    EventMachine.run do
-      container = run_command_with_self command
-      yield(container) if block_given?
+    prepared_container do |container|
+      container.execute_interactively(run_command, &block)
     end
-    container.destroy
-  end
-
-  def run_command_with_self(command)
-    container = Container.new(execution_environment, execution_environment.permitted_execution_time)
-    container.copy_submission_files self
-    container.execute_interactively(command)
-    container
   end
 
   private
 
+  def prepared_container
+    request_time = Time.now
+    container = Container.new(execution_environment, execution_environment.permitted_execution_time)
+    container.copy_submission_files self
+    container_time = Time.now
+    waiting_for_container_time = Time.now - request_time
+    yield(container) if block_given?
+    execution_time = Time.now - container_time
+    container.destroy
+  end
+
   def command_for(template, file)
     filepath = collect_files.find { |f| f.name_with_extension == file }.filepath
     template % command_substitutions(filepath)

lib/container.rb

@@ -1,10 +1,10 @@
 # frozen_string_literal: true
 
+require 'container_connection'
+
 class Container
   BASE_URL = "http://192.168.178.53:5000"
 
-  attr_accessor :socket
-
   def initialize(execution_environment, time_limit = nil)
     url = "#{BASE_URL}/execution-environments/#{execution_environment.id}/containers/create"
     body = {}
@@ -39,7 +39,11 @@ class Container
 
   def execute_interactively(command)
     websocket_url = execute_command(command)[:websocket_url]
-    @socket = Faye::WebSocket::Client.new(websocket_url,  [], ping: 0.1)
+    EventMachine.run do
+      #socket = Faye::WebSocket::Client.new(websocket_url, [], ping: 0.1)
+      socket = ContainerConnection.new(websocket_url)
+      yield(self, socket) if block_given?
+    end
   end
 
   def destroy

lib/container_connection.rb

@@ -0,0 +1,58 @@
+require 'faye/websocket/client'
+
+class ContainerConnection
+  EVENTS = %i[start message exit stdout stderr].freeze
+
+  def initialize(url)
+    @socket = Faye::WebSocket::Client.new(url, [], ping: 0.1)
+
+    %i[open message error close].each do |event_type|
+      @socket.on event_type, &:"on_#{event_type}"
+    end
+
+    EVENTS.each { |event_type| instance_variable_set(:"@#{event_type}_callback", lambda {}) }
+  end
+
+  def on(event, &block)
+    return unless EVENTS.include? event
+
+    instance_variable_set(:"@#{event}_callback", block)
+  end
+
+  def send(data)
+    @socket.send(data)
+  end
+
+  private
+
+  def parse(event)
+    JSON.parse(event.data).deep_symbolize_keys
+  end
+
+  def on_message(event)
+    event = parse(event)
+    case event[:type]
+    when :exit_code
+      @exit_code = event[:data]
+    when :stderr
+      @stderr_callback.call event[:data]
+      @message_callback.call event[:data]
+    when :stdout
+      @stdout_callback.call event[:data]
+      @message_callback.call event[:data]
+    else
+      :error
+    end
+  end
+
+  def on_open(event)
+    @start_callback.call
+  end
+
+  def on_error(event)
+  end
+
+  def on_close(event)
+    @exit_callback.call @exit_code
+  end
+end