Optimise non-blocking queue polling in QueryRunner and RunOutputGroup

Author: alexjpwalker

state/connection/QueryRunner.kt

@@ -108,7 +108,7 @@ class QueryRunner constructor(
         const val MATCH_GROUP_AGGREGATE_QUERY_NO_RESULT =
             "Match Group Aggregate query did not match any concept groups to aggregate in the database."
 
-        private const val COUNT_DOWN_LATCH_PERIOD_MS: Long = 10
+        private const val COUNT_DOWN_LATCH_PERIOD_MS: Long = 50
         private val RUNNING_INDICATOR_DELAY = Duration.seconds(3)
         private val LOGGER = KotlinLogging.logger {}
     }
@@ -180,7 +180,7 @@ class QueryRunner constructor(
             if (!hasStopSignal.get()) {
                 do {
                     isConsumed = consumerLatch.count == 0L
-                    delay(COUNT_DOWN_LATCH_PERIOD_MS)
+                    if (!isConsumed) delay(COUNT_DOWN_LATCH_PERIOD_MS)
                 } while (!isConsumed && !hasStopSignal.get())
             }
             onComplete()

view/graph/GraphArea.kt

@@ -154,7 +154,7 @@ class GraphArea(transactionState: TransactionState) {
             edgeLabelSizes[edge.label]?.let { viewport.rectIsVisible(edge.geometry.labelRect(it, density)) } ?: false
         }.let {
             // Ensure smooth performance during initial explosion
-            if (it.size < 100 || graph.physics.alpha < 0.25) it.toSet() else emptySet()
+            if (it.size < 200 || graph.physics.alpha < 0.5) it.toSet() else emptySet()
         }
     }
 
@@ -194,7 +194,7 @@ class GraphArea(transactionState: TransactionState) {
         val vertices = graph.vertices.filter { viewport.rectIsVisible(it.geometry.rect) }
         Canvas(Modifier.fillMaxSize()) { vertices.forEach { drawVertexBackground(it) } }
         // Ensure smooth performance when zoomed out, and during initial explosion
-        if (viewport.scale > 0.2 && (vertices.size < 100 || graph.physics.alpha < 0.25)) {
+        if (viewport.scale > 0.2 && (vertices.size < 200 || graph.physics.alpha < 0.5)) {
             vertices.forEach { VertexLabel(it, it.geometry.position) }
         }
     }

view/output/RunOutputGroup.kt

@@ -43,6 +43,7 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.delay
 import mu.KotlinLogging
+import java.util.concurrent.TimeUnit.MILLISECONDS
 
 internal class RunOutputGroup constructor(
     private val runner: QueryRunner,
@@ -66,16 +67,14 @@ internal class RunOutputGroup constructor(
 
     companion object {
         private const val CONSUMER_PERIOD_MS = 33 // 30 FPS
-        private const val COUNT_DOWN_LATCH_PERIOD_MS: Long = 10
+        private const val COUNT_DOWN_LATCH_PERIOD_MS: Long = 50
         private val LOGGER = KotlinLogging.logger {}
 
         private suspend fun <E> LinkedBlockingQueue<E>.takeNonBlocking(periodMS: Long): E {
-            var item: E
-            do {
-                item = this.poll()
-                delay(periodMS)
-            } while (item == null)
-            return item
+            while (true) {
+                val item = this.poll(1, MILLISECONDS)
+                if (item != null) return item else delay(periodMS)
+            }
         }
     }
 
@@ -122,10 +121,9 @@ internal class RunOutputGroup constructor(
     }
 
     private fun concludeRunnerIsConsumed() = coroutineScope.launchAndHandle(GlobalState.notification, LOGGER) {
-        do {
-            val isConsumed = futuresLatch.count == 0L
+        while (futuresLatch.count > 0L) {
             delay(COUNT_DOWN_LATCH_PERIOD_MS)
-        } while (!isConsumed)
+        }
         runner.setConsumed()
         endTime = System.currentTimeMillis()
     }