ExecutionStore should cycle through queued elements [BA-6487 prerequisite] #5588

cjllanwarne · 2020-07-22T19:30:30Z

Updating keys in the ExecutionStore might leave room for previously "waiting for queue space" jobs to start.

TOL: I'm not 100% sure what this concept of "waiting for queue space" is, it seems to be distinct from the EJEA QueuedInCromwell task status since it's quite possible to have over 100,000 jobs in that status just fine... 🤔

mcovarr · 2020-07-22T20:14:44Z

engine/src/main/scala/cromwell/engine/workflow/lifecycle/execution/stores/ExecutionStore.scala

@@ -154,7 +154,7 @@ final case class SealedExecutionStore private[stores](private val statusStore: M
 sealed abstract class ExecutionStore private[stores](statusStore: Map[JobKey, ExecutionStatus], val needsUpdate: Boolean) {
  // View of the statusStore more suited for lookup based on status
  lazy val store: Map[ExecutionStatus, List[JobKey]] = statusStore.groupBy(_._2).safeMapValues(_.keys.toList)
-  lazy val queuedJobsAboveThreshold = queuedJobs > MaxJobsToStartPerTick
+  lazy val queuedJobsAboveThreshold = queuedJobs >= MaxJobsToStartPerTick


if the variable should truly represent queued jobs above the threshold why would this be >=?

Subsequently replaced with a subtraction

mcovarr · 2020-07-22T20:17:07Z

engine/src/main/scala/cromwell/engine/workflow/lifecycle/execution/stores/ExecutionStore.scala

+    // The store might newly need updating now if keys have changed and either:
+    // - A job has completed -> downstream jobs might now be runnable
+    // - The store had jobs waiting for queue space -> more queue space might have been freed up
+    val needsNewUpdate = values.nonEmpty && (values.values.exists(_.isTerminalOrRetryable) || store.contains(WaitingForQueueSpace))


why does this need the new values.nonEmpty check?

Because my test case ended by updating an empty set of values (ie changing nothing) but nevertheless being considered as newly needing an update

grsterin · 2020-07-22T20:38:57Z

engine/src/main/scala/cromwell/engine/workflow/lifecycle/execution/stores/ExecutionStore.scala

@@ -274,24 +281,24 @@ sealed abstract class ExecutionStore private[stores](statusStore: Map[JobKey, Ex
        // Even if the key is runnable, if it's a call key and there's already too many queued jobs,
        // don't start it and mark it as WaitingForQueueSpace
        // TODO maybe also limit the number of expression keys to run somehow ?
-        case callKey: CallKey if runnable && queuedJobsAboveThreshold =>
+        case callKey: CallKey if runnable && startableJobLimit > 0 =>


I think startableJobLimit > 0 leaves us room for starting more jobs right away? So it should be if runnable && startableJobLimit <= 0 here?

With the test case passing, I think this is the right way round, but maybe the variable is badly named? It's supposed to mean "how many jobs am I going to be allowing myself to start this time". Which you need to be over 0 to start any new jobs

It's supposed to mean "how many jobs am I going to be allowing myself to start this time". Which you need to be over 0 to start any new jobs

Exactly, and this is how I read it. But I think in this case clause we are saying that the job is not runnable this time (by returning false) and postponing the job by transitioning it into the WaitingForQueueSpace state. But since we have startableJobLimit > 0 we should allow to start this job right away, instead of putting it into the waiting state

Oh right, I see what you mean. Yes I think you're right, I'll fix that

grsterin · 2020-07-23T16:27:15Z

.../src/test/scala/cromwell/engine/workflow/lifecycle/execution/stores/ExecutionStoreSpec.scala

+    store.store(WaitingForQueueSpace).size should be(9000)
+    store.store.contains(Running) should be(false)
+
+    while(store.store.getOrElse(Running, List.empty).size < 10000) {


#5590 (comment)

grsterin · 2020-07-23T16:27:53Z

.../src/test/scala/cromwell/engine/workflow/lifecycle/execution/stores/ExecutionStoreSpec.scala

+
+    updateStoreToEnqueueNewlyRunnableJobs()
+
+    while(store.store.getOrElse(Running, List.empty).size < 10000) {


Same concern as above

cjllanwarne · 2020-07-23T18:32:21Z

Closing for now, in hopes that #5590 will work instead

cjllanwarne · 2020-07-24T15:38:25Z

Re-opening because we think we prefer this one to #5590 after all...

grsterin · 2020-07-24T16:50:32Z

There're several comments from me regarding usage of while loops in tests (the same ones as in another PR). Otherwise LGTM.

mcovarr · 2020-07-24T16:50:01Z

engine/src/main/scala/cromwell/engine/workflow/lifecycle/execution/stores/ExecutionStore.scala

@@ -274,24 +281,24 @@ sealed abstract class ExecutionStore private[stores](statusStore: Map[JobKey, Ex
        // Even if the key is runnable, if it's a call key and there's already too many queued jobs,
        // don't start it and mark it as WaitingForQueueSpace
        // TODO maybe also limit the number of expression keys to run somehow ?
-        case callKey: CallKey if runnable && queuedJobsAboveThreshold =>
+        case callKey: CallKey if runnable && startableJobLimit <= 0 =>


the code above keeps startableJobLimit from being less than zero

Suggested change

case callKey: CallKey if runnable && startableJobLimit <= 0 =>

case callKey: CallKey if runnable && startableJobLimit == 0 =>

https://www.phrases.org.uk/meanings/belt-and-braces.html

mcovarr · 2020-07-24T16:51:56Z

engine/src/main/scala/cromwell/engine/workflow/lifecycle/execution/stores/ExecutionStore.scala

    // Start with keys that are waiting for queue space as we know they're runnable already. Then filter the not started ones
    val readyToStart = runnableWaitingForQueueSpace ++ keysWithStatus(NotStarted).toStream.filter(filterFunction)

    // Compute the first ExecutionStore.MaxJobsToStartPerTick + 1 runnable keys
-    val keysToStartPlusOne = readyToStart.take(MaxJobsToStartPerTick + 1).toList
+    val keysToStartPlusOne = readyToStart.take(startableJobLimit + 1).toList


comment no longer matches the code

mcovarr · 2020-07-24T16:52:02Z

engine/src/main/scala/cromwell/engine/workflow/lifecycle/execution/stores/ExecutionStore.scala

@@ -303,6 +310,6 @@ sealed abstract class ExecutionStore private[stores](statusStore: Map[JobKey, Ex
    } else withNeedsUpdateFalse

    // Only take the first ExecutionStore.MaxJobsToStartPerTick from the above list.
-    ExecutionStoreUpdate(keysToStartPlusOne.take(MaxJobsToStartPerTick), updated, internalUpdates)
+    ExecutionStoreUpdate(keysToStartPlusOne.take(startableJobLimit), updated, internalUpdates)


mcovarr · 2020-07-24T16:53:59Z

engine/src/main/scala/cromwell/engine/workflow/lifecycle/execution/stores/ExecutionStore.scala

@@ -154,7 +154,10 @@ final case class SealedExecutionStore private[stores](private val statusStore: M
 sealed abstract class ExecutionStore private[stores](statusStore: Map[JobKey, ExecutionStatus], val needsUpdate: Boolean) {
  // View of the statusStore more suited for lookup based on status
  lazy val store: Map[ExecutionStatus, List[JobKey]] = statusStore.groupBy(_._2).safeMapValues(_.keys.toList)
-  lazy val queuedJobsAboveThreshold = queuedJobs > MaxJobsToStartPerTick
+  lazy val startableJobLimit = {


maybe queueableJobLimit would be a more accurate name?

…ncrease

…prerequisite] (#5588)" This reverts commit 3da6198.

…prerequisite] (#5588)" (#5887) This reverts commit 3da6198.

…prerequisite] (#5588)" This reverts commit 3da6198.

…prerequisite] (#5588)" (#5918)

cjllanwarne added 5 commits July 22, 2020 13:52

Execution Store 10000 elements unit test?

90533ff

small fixup

c7882fc

small fixup

0a8746b

compile fixup

0668ba9

Fix the bug and have a test for it

efade10

cjllanwarne requested review from grsterin and mcovarr July 22, 2020 19:30

Ensure we stay within the queue limit and add a test to that effect

15507d9

mcovarr reviewed Jul 22, 2020

View reviewed changes

grsterin reviewed Jul 22, 2020

View reviewed changes

Got the comparison the wrong way round

3e6c868

cjllanwarne mentioned this pull request Jul 22, 2020

Remove redundant WaitingForQueueSpace execution status [BA-6487 prereq] #5590

Closed

grsterin reviewed Jul 23, 2020

View reviewed changes

cjllanwarne closed this Jul 23, 2020

cjllanwarne reopened this Jul 24, 2020

mcovarr reviewed Jul 24, 2020

View reviewed changes

Fixed a few comments, asserted that Running jobs counts continue to i…

e162428

…ncrease

cjllanwarne requested review from grsterin and mcovarr July 27, 2020 18:35

mcovarr approved these changes Jul 27, 2020

View reviewed changes

grsterin approved these changes Jul 29, 2020

View reviewed changes

cjllanwarne merged commit 3da6198 into develop Jul 30, 2020

cjllanwarne deleted the cjl_execution_store_unit_test branch July 30, 2020 14:34

cjllanwarne added a commit that referenced this pull request Sep 24, 2020

Revert "ExecutionStore should cycle through queued elements [BA-6487 …

9e02e7b

…prerequisite] (#5588)" This reverts commit 3da6198.

cjllanwarne added a commit that referenced this pull request Sep 24, 2020

Revert "ExecutionStore should cycle through queued elements [BA-6487 …

5289f50

…prerequisite] (#5588)" This reverts commit 3da6198.

cjllanwarne added a commit that referenced this pull request Sep 24, 2020

Revert "ExecutionStore should cycle through queued elements [BA-6487 …

9b11416

…prerequisite] (#5588)" (#5887) This reverts commit 3da6198.

mcovarr added a commit that referenced this pull request Oct 5, 2020

Revert "ExecutionStore should cycle through queued elements [BA-6487 …

4295326

…prerequisite] (#5588)" This reverts commit 3da6198.

mcovarr added a commit that referenced this pull request Oct 7, 2020

Revert "ExecutionStore should cycle through queued elements [BA-6487 …

171f12c

…prerequisite] (#5588)" (#5918)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ExecutionStore should cycle through queued elements [BA-6487 prerequisite] #5588

ExecutionStore should cycle through queued elements [BA-6487 prerequisite] #5588

cjllanwarne commented Jul 22, 2020

mcovarr Jul 22, 2020

cjllanwarne Jul 22, 2020

mcovarr Jul 22, 2020

cjllanwarne Jul 22, 2020

grsterin Jul 22, 2020

cjllanwarne Jul 22, 2020

grsterin Jul 22, 2020 •

edited

cjllanwarne Jul 22, 2020

grsterin Jul 23, 2020

grsterin Jul 23, 2020

cjllanwarne commented Jul 23, 2020

cjllanwarne commented Jul 24, 2020

grsterin commented Jul 24, 2020 •

edited

mcovarr Jul 24, 2020

cjllanwarne Jul 27, 2020

mcovarr Jul 24, 2020

mcovarr Jul 24, 2020

mcovarr Jul 24, 2020


		updateStoreToEnqueueNewlyRunnableJobs()

		while(store.store.getOrElse(Running, List.empty).size < 10000) {

	case callKey: CallKey if runnable && startableJobLimit <= 0 =>
	case callKey: CallKey if runnable && startableJobLimit == 0 =>

ExecutionStore should cycle through queued elements [BA-6487 prerequisite] #5588

ExecutionStore should cycle through queued elements [BA-6487 prerequisite] #5588

Conversation

cjllanwarne commented Jul 22, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

grsterin Jul 22, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cjllanwarne commented Jul 23, 2020

cjllanwarne commented Jul 24, 2020

grsterin commented Jul 24, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

grsterin Jul 22, 2020 •

edited

grsterin commented Jul 24, 2020 •

edited