Reserve runners for the whole duration of their use

This commit changes how runners are reserved and released. Previously, a runner was reserved only for the actual time it was actively used and the aim was to reserve it for a time as short as possible. The previous mechanism worked, but had some drawbacks:

For example, a learner could start a submission scoring, which would perform (1) a file copy, (2) an execution of the first test file, (3) an execution of the second test file, etc. For each of these actions, the runner would be reserved and released. Now, it could happen that another run was triggered between the execution of both test files, causing issues with the remaining scoring.

In the refactored version, a runner is reserved once before initially copying files and only released after the last command finished. This ensures that no one can interrupt an execution started once, hopefully making it more robust (and faster, since less locks are required).

Author: MrSerth

app/models/runner.rb

@@ -4,6 +4,12 @@ class Runner < ApplicationRecord
   include ContributorCreation
   belongs_to :execution_environment
 
+  # Each reservation is extended by the specified duration.
+  # This allows us to extend the reservation in case of a timeout and a subsequent runner swap.
+  # By observation, we know that a runner swap usually takes 500ms on average, sometimes 700ms.
+  # This time includes requesting a new runner and copying the submission files to the new runner.
+  RESERVATION_BUFFER = 1.second
+
   before_validation :request_id
   validates :runner_id, presence: true
 
@@ -43,10 +49,10 @@ def self.for(contributor, execution_environment)
     runner
   end
 
-  def copy_files(files)
-    reserve!
+  def copy_files(files, exclusive: true)
+    reserve! if exclusive
     @strategy.copy_files(files)
-    release!
+    release! if exclusive
   rescue Runner::Error::RunnerInUse => e
     Rails.logger.debug { "Copying files failed because the runner was already in use: #{e.message}" }
     raise e
@@ -55,7 +61,7 @@ def copy_files(files)
     request_new_id
     save
     @strategy.copy_files(files)
-    release!
+    release! if exclusive
   end
 
   def download_file(desired_file, privileged_execution:, exclusive: true, &)
@@ -192,12 +198,17 @@ def reserve!
         @error = Runner::Error::RunnerInUse.new("The desired Runner #{id} is already in use until #{reserved_until.iso8601}.")
         raise @error
       else
-        update!(reserved_until: Time.zone.now + execution_environment.permitted_execution_time.seconds)
-        @error = nil
+        lock_runner_for_permitted_execution_time!
       end
     end
   end
 
+  def extend!
+    with_lock do
+      lock_runner_for_permitted_execution_time!
+    end
+  end
+
   def release!
     return if @error.present?
 
@@ -237,4 +248,13 @@ def request_new_id
       end
     end
   end
+
+  def lock_runner_for_permitted_execution_time!
+    # We reserve the runner for the permitted execution time plus the specified buffer time to ensure that the runner is
+    # not released before the execution is finished. This is especially important for the case that a command execution
+    # timed out, but subsequent commands are scheduled (i.e., when calculating a score and swapping a runner).
+    reservation = execution_environment.permitted_execution_time.seconds + RESERVATION_BUFFER
+    update!(reserved_until: Time.zone.now + reservation)
+    @error = nil
+  end
 end

app/models/submission.rb

@@ -154,7 +154,7 @@ def calculate_score(requesting_user)
       file_scores = assessments.sort_by {|file| file.teacher_defined_linter? ? 0 : 1 }.map.with_index(1) do |file, index|
         output = run_test_file file, runner, waiting_duration
         # If the previous execution failed and there is at least one more test, we request a new runner.
-        runner, waiting_duration = swap_runner(runner) if output[:status] == :timeout && index < assessment_number
+        swap_runner(runner) if output[:status] == :timeout && index < assessment_number
         score_file(output, file, requesting_user)
       end
     end
@@ -169,7 +169,7 @@ def run(file, &block)
     run_command = command_for execution_environment.run_command, file.filepath
     durations = {}
     prepared_runner do |runner, waiting_duration|
-      durations[:execution_duration] = runner.attach_to_execution(run_command, &block)
+      durations[:execution_duration] = runner.attach_to_execution(run_command, exclusive: false, &block)
       durations[:waiting_duration] = waiting_duration
     rescue Runner::Error => e
       e.waiting_duration = waiting_duration
@@ -193,7 +193,7 @@ def test(file, requesting_user)
   def run_test_file(file, runner, waiting_duration)
     test_command = command_for execution_environment.test_command, file.filepath
     result = {file_role: file.role, waiting_for_container_time: waiting_duration}
-    output = runner.execute_command(test_command, raise_exception: false)
+    output = runner.execute_command(test_command, raise_exception: false, exclusive: false)
     result.merge(output)
   end
 
@@ -211,10 +211,11 @@ def build_files_hash(files, attribute)
     files&.map(&attribute.to_proc)&.zip(files).to_h
   end
 
-  def prepared_runner
+  def prepared_runner(existing_runner: nil, exclusive: true)
     request_time = Time.zone.now
     begin
-      runner = Runner.for(contributor, exercise.execution_environment)
+      runner = existing_runner || Runner.for(contributor, exercise.execution_environment)
+      runner.reserve! if exclusive
       files = collect_files
 
       case cause
@@ -233,25 +234,32 @@ def prepared_runner
       end
 
       Rails.logger.debug { "#{Time.zone.now.getutc.inspect}: Copying files to Runner #{runner.id} for #{contributor_type} #{contributor_id} and Submission #{id}." }
-      runner.copy_files(files)
+      # We don't want `copy_files` to be exclusive since we reserve runners for the whole `prepared_runner` block.
+      runner.copy_files(files, exclusive: false)
     rescue Runner::Error => e
       e.waiting_duration = Time.zone.now - request_time
       raise
     end
     waiting_duration = Time.zone.now - request_time
-    yield(runner, waiting_duration)
+    yield(runner, waiting_duration) if block_given?
+  ensure
+    runner&.release! if exclusive
   end
 
-  def swap_runner(old_runner)
-    old_runner.update(runner_id: nil)
-    new_runner = nil
-    new_waiting_duration = nil
-    # We request a new runner that will also include all files of the current submission
-    prepared_runner do |runner, waiting_duration|
-      new_runner = runner
-      new_waiting_duration = waiting_duration
+  def swap_runner(runner, exclusive: true)
+    # We use a transaction to ensure that the runner is swapped atomically.
+    transaction do
+      # Due to the `before_validation` callback in the `Runner` model,
+      # the following line will immediately request a new runner.
+      runner.update(runner_id: nil)
+
+      # With the new runner being ready, we only need to prepare it (by copying the files).
+      # Since no actual execution is performed, we don't need to reserve the runner.
+      prepared_runner(existing_runner: runner, exclusive: false)
+
+      # Now, we update the locks if desired. This is only necessary when a runner is used exclusively.
+      runner.extend! if exclusive
     end
-    [new_runner, new_waiting_duration]
   end
 
   def command_for(template, filepath)