Skip to content

Commit

Permalink
Changed locking mechanism in InteractiveSynchronousRunManager
Browse files Browse the repository at this point in the history
  • Loading branch information
@lucaro
lucaro committed Jan 4, 2024
1 parent 0d362b1 commit 6456a53
Show file tree
Hide file tree
Showing 2 changed files with 33 additions and 36 deletions.
2 changes: 1 addition & 1 deletion backend/src/main/kotlin/dev/dres/DRES.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ import kotlin.system.exitProcess
*/
object DRES {
/** Version of DRES. */
const val VERSION = "2.0.0-RC1"
const val VERSION = "2.0.0-RC2"

/** Application root; should be relative to JAR file or classes path. */
val APPLICATION_ROOT: Path =
Expand Down
67 changes: 32 additions & 35 deletions backend/src/main/kotlin/dev/dres/run/InteractiveSynchronousRunManager.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ class InteractiveSynchronousRunManager(
private val updatables = mutableListOf<Updatable>()

/** A lock for state changes to this [InteractiveSynchronousRunManager]. */
private val stateLock = ReentrantReadWriteLock()
//private val stateLock = ReentrantReadWriteLock()

private fun checkContext(context: RunActionContext) {
if (!context.isAdmin)
Expand Down Expand Up @@ -130,7 +130,7 @@ class InteractiveSynchronousRunManager(
}
}

override fun start(context: RunActionContext) = this.stateLock.write {
override fun start(context: RunActionContext) = synchronized(this.status) {
checkStatus(RunManagerStatus.CREATED)
checkContext(context)

Expand All @@ -145,7 +145,7 @@ class InteractiveSynchronousRunManager(
this.status = RunManagerStatus.ACTIVE
}

override fun end(context: RunActionContext) = this.stateLock.write {
override fun end(context: RunActionContext) = synchronized(this.status) {
checkStatus(RunManagerStatus.CREATED, RunManagerStatus.ACTIVE /*RunManagerStatus.TASK_ENDED*/)
checkContext(context)

Expand Down Expand Up @@ -181,12 +181,12 @@ class InteractiveSynchronousRunManager(
*
* @return [DbTaskTemplate]
*/
override fun currentTaskTemplate(context: RunActionContext): ApiTaskTemplate = this.stateLock.write {
override fun currentTaskTemplate(context: RunActionContext): ApiTaskTemplate {
checkStatus(RunManagerStatus.CREATED, RunManagerStatus.ACTIVE)
this.evaluation.getCurrentTaskTemplate()
return this.evaluation.getCurrentTaskTemplate()
}

override fun previous(context: RunActionContext): Boolean = this.stateLock.write {
override fun previous(context: RunActionContext): Boolean {
checkContext(context)
val newIndex = this.evaluation.templateIndex - 1
return try {
Expand All @@ -197,7 +197,7 @@ class InteractiveSynchronousRunManager(
}
}

override fun next(context: RunActionContext): Boolean = this.stateLock.write {
override fun next(context: RunActionContext): Boolean {
checkContext(context)
val newIndex = this.evaluation.templateIndex + 1
return try {
Expand All @@ -208,7 +208,7 @@ class InteractiveSynchronousRunManager(
}
}

override fun goTo(context: RunActionContext, index: Int) {
override fun goTo(context: RunActionContext, index: Int) = synchronized(this) {
checkStatus(RunManagerStatus.ACTIVE)
assureNoRunningTask()
this.evaluation.taskRuns.any { it.status == ApiTaskStatus.RUNNING }
Expand All @@ -225,7 +225,7 @@ class InteractiveSynchronousRunManager(
}
}

override fun startTask(context: RunActionContext) = this.stateLock.write {
override fun startTask(context: RunActionContext) = synchronized(this) {
checkStatus(RunManagerStatus.ACTIVE)
assureNoRunningTask()
checkContext(context)
Expand Down Expand Up @@ -267,7 +267,7 @@ class InteractiveSynchronousRunManager(
this.evaluation.currentTaskRun!!.taskId
}

override fun abortTask(context: RunActionContext) = this.stateLock.write {
override fun abortTask(context: RunActionContext) = synchronized(this) {
checkStatus(RunManagerStatus.ACTIVE)
assertTaskPreparingOrRunning()
checkContext(context)
Expand All @@ -291,15 +291,15 @@ class InteractiveSynchronousRunManager(
* @param context The [RunActionContext] used for the invocation.
* @return [DbTask] or null
*/
override fun currentTask(context: RunActionContext) = this.stateLock.read {
override fun currentTask(context: RunActionContext) =
when (this.evaluation.currentTaskRun?.status) {
ApiTaskStatus.PREPARING,
ApiTaskStatus.RUNNING,
ApiTaskStatus.ENDED -> this.evaluation.currentTaskRun

else -> null
}
}


/**
* Returns [DbTask]s for a specific task [EvaluationId]. May be empty.
Expand All @@ -315,19 +315,16 @@ class InteractiveSynchronousRunManager(
* @param context The [RunActionContext] used for the invocation.
* @return List of [DbSubmission]s.
*/
override fun allSubmissions(context: RunActionContext): List<DbSubmission> = this.stateLock.read {
this.evaluation.taskRuns.flatMap { it.getDbSubmissions() }
}
override fun allSubmissions(context: RunActionContext): List<DbSubmission> = this.evaluation.taskRuns.flatMap { it.getDbSubmissions() }

/**
* Returns the [DbSubmission]s for all currently active [DbTask]s or an empty [List], if no such task is active.
*
* @param context The [RunActionContext] used for the invocation.
* @return List of [DbSubmission]s for the currently active [DbTask]
*/
override fun currentSubmissions(context: RunActionContext): List<DbSubmission> = this.stateLock.read {
this.currentTask(context)?.getDbSubmissions()?.toList() ?: emptyList()
}
override fun currentSubmissions(context: RunActionContext): List<DbSubmission> = this.currentTask(context)?.getDbSubmissions()?.toList() ?: emptyList()


/**
* Returns the number of [DbTask]s held by this [RunManager].
Expand All @@ -345,7 +342,7 @@ class InteractiveSynchronousRunManager(
* @throws IllegalArgumentException If the specified correction cannot be applied.
* @throws IllegalStateException If [RunManager] was not in wrong [RunManagerStatus].
*/
override fun adjustDuration(context: RunActionContext, s: Int): Long = this.stateLock.read {
override fun adjustDuration(context: RunActionContext, s: Int): Long = synchronized(this) {
checkContext(context)

/* Obtain task and perform sanity check. */
Expand All @@ -369,7 +366,7 @@ class InteractiveSynchronousRunManager(
*
* @return Time remaining until the task will end or -1, if no task is running.
*/
override fun timeLeft(context: RunActionContext): Long = this.stateLock.read {
override fun timeLeft(context: RunActionContext): Long {
return if (this.evaluation.currentTaskRun?.status == ApiTaskStatus.RUNNING) {
val currentTaskRun = this.currentTask(context)
?: throw IllegalStateException("SynchronizedRunManager is in status ${this.status} but has no active TaskRun. This is a serious error!")
Expand All @@ -388,7 +385,7 @@ class InteractiveSynchronousRunManager(
*
* @return Time remaining until the task will end or -1, if no task is running.
*/
override fun timeElapsed(context: RunActionContext): Long = this.stateLock.read {
override fun timeElapsed(context: RunActionContext): Long {
return if (this.evaluation.currentTaskRun?.status == ApiTaskStatus.RUNNING) {
val currentTaskRun = this.currentTask(context)
?: throw IllegalStateException("SynchronizedRunManager is in status ${this.status} but has no active TaskRun. This is a serious error!")
Expand All @@ -410,7 +407,7 @@ class InteractiveSynchronousRunManager(
*
* @param viewerId The ID of the viewer's WebSocket session.
*/
override fun overrideReadyState(context: RunActionContext, viewerId: String): Boolean = this.stateLock.read {
override fun overrideReadyState(context: RunActionContext, viewerId: String): Boolean {
checkStatus(RunManagerStatus.ACTIVE)
assertTaskPreparingOrRunning()
checkContext(context)
Expand Down Expand Up @@ -489,7 +486,7 @@ class InteractiveSynchronousRunManager(
* @param context The [RunActionContext] used for the invocation.
* @param submission [ApiSubmission] that should be registered.
*/
override fun postSubmission(context: RunActionContext, submission: ApiClientSubmission) = this.stateLock.read {
override fun postSubmission(context: RunActionContext, submission: ApiClientSubmission) : ApiSubmission {

/* Phase 1: Basic lookups required for validation (read-only). */
val (task, transformedSubmission) = this.store.transactional(true) {
Expand All @@ -514,7 +511,7 @@ class InteractiveSynchronousRunManager(
}

/* Phase 2: Create DbSubmission, apply transformers and validate it. */
return@read this.store.transactional {
return this.store.transactional {
/* Convert submission to database representation. */
val db = transformedSubmission.toNewDb()

Expand Down Expand Up @@ -544,7 +541,7 @@ class InteractiveSynchronousRunManager(
context: RunActionContext,
submissionId: SubmissionId,
submissionStatus: ApiVerdictStatus
): Boolean = this.stateLock.read {
): Boolean {
val (taskId, status) = this.store.transactional {
val answerSet = DbAnswerSet.filter { it.submission.submissionId eq submissionId }.singleOrNull()
?: throw IllegalArgumentException("Could not find submission with ID ${submissionId}.")
Expand Down Expand Up @@ -576,12 +573,12 @@ class InteractiveSynchronousRunManager(
*/
override fun run() {
/* Preparation . */
this.stateLock.read {
//this.stateLock.read {
this.store.transactional {
this.updatables.sortBy { it.phase } /* Sort list of by [Phase] in ascending order. */
AccessManager.registerRunManager(this) /* Register the run manager with the access manager. */
}
}
//}

/** Add this InteractiveSynchronousRunManager as a listener for changes to the data store. */
this.store.addListener(this)
Expand All @@ -593,15 +590,15 @@ class InteractiveSynchronousRunManager(
while (this.status != RunManagerStatus.TERMINATED) {
try {
/* Obtain lock on current state. */
this.stateLock.read {
//this.stateLock.read {
this.store.transactional {
/* 1) Invoke all relevant [Updatable]s. */
this.invokeUpdatables()

/* 2) Process internal state updates (if necessary). */
this.internalStateUpdate()
}
}
// }

/* 3) Yield to other threads. */
Thread.sleep(500)
Expand Down Expand Up @@ -630,12 +627,12 @@ class InteractiveSynchronousRunManager(
this.store.removeListener(this)

/* Finalization. */
this.stateLock.read {
//this.stateLock.read {
this.store.transactional {
this.invokeUpdatables() /* Invoke [Updatable]s one last time. */
AccessManager.deregisterRunManager(this) /* De-register this run manager with the access manager. */
}
}
//}

LOGGER.info("SynchronousRunManager ${this.id} reached end of run logic.")
}
Expand All @@ -660,7 +657,7 @@ class InteractiveSynchronousRunManager(
private fun internalStateUpdate() {
/** Case 1: Facilitates internal transition from RunManagerStatus.PREPARING_TASK to RunManagerStatus.RUNNING_TASK. */
if (this.evaluation.currentTaskRun?.status == ApiTaskStatus.PREPARING && this.readyLatch.allReadyOrTimedOut()) {
this.stateLock.write {
//this.stateLock.write {
this.evaluation.currentTaskRun!!.start()
AuditLogger.taskStart(
this.id,
Expand All @@ -669,15 +666,15 @@ class InteractiveSynchronousRunManager(
AuditLogSource.INTERNAL,
null
)
}
//}

// /* Enqueue WS message for sending */
// RunExecutor.broadcastWsMessage(ServerMessage(this.id, ServerMessageType.TASK_START, this.evaluation.currentTask?.taskId))
}

/** Case 2: Facilitates internal transition from RunManagerStatus.RUNNING_TASK to RunManagerStatus.TASK_ENDED due to timeout. */
if (this.evaluation.currentTaskRun?.status == ApiTaskStatus.RUNNING) {
this.stateLock.write {
//this.stateLock.write {
val task = this.evaluation.currentTaskRun!!
val timeLeft = max(
0L,
Expand All @@ -691,7 +688,7 @@ class InteractiveSynchronousRunManager(
// /* Enqueue WS message for sending */
// RunExecutor.broadcastWsMessage(ServerMessage(this.id, ServerMessageType.TASK_END, this.evaluation.currentTask?.taskId))
}
}
//}
}
}

Expand Down

0 comments on commit 6456a53

Please sign in to comment.