New Scheduler #278

fwbrasil · 2024-04-22T22:58:34Z

Kyo's current scheduler is functional and has been serving the projects need's well but the code is focused on simplicity using global objects and with only indirect coverage via tests that use fibers. This PR introduces a new version of the scheduler with several major improvements:

The scheduler is now in a separate module with no dependencies to other kyo modules. This allows users to adopt the scheduler in isolation, for example with another effect system like ZIO.
The code was refactored to become more testable via function composition and constructor injection. Its test coverage is quite high now.
A new Regulator abstraction was introduced to replace the concurrency management by Coordinator, which has been removed.
In addition to the concurrency regulator, a new Admission regulator probes the scheduler's queuing delay in order to provide a back pressure signal. The signal isn't integrated in other modules yet but it should be used to reject requests in kyo-tapir for example.
The Task interface is now a first-class public API that provides built-in runtime tracking, which simplified IOTask.

I realize this PR is too large to review. I apologize for that but isolating changes would slow things down too much, which doesn't seem necessary in the current phase of the project. Please let me know if this kind of change is getting too intrusive, though. I'm also planning to eventually write a separate readme for the scheduler but I want to have some experience with it first.

fwbrasil · 2024-04-22T22:59:01Z

kyo-core/shared/src/main/scala/kyo/core.scala

@@ -145,12 145,12 @@ object core:
 end Handler

 trait Safepoint[-E]:
- def check(): Boolean
+ def preempt(): Boolean


Minor refactoring for clarity.

fwbrasil · 2024-04-22T23:00:45Z

kyo-core/shared/src/main/scala/kyo/scheduler/IOPromise.scala

@@ -129,7 129,7 @@ private[kyo] class IOPromise[T](state: State[T])
 promise.get() match
 case _: Pending[T] @unchecked =>
 IOs {
- Scheduler.flush()
+ Scheduler.get.flush()


Scheduler isn't an object anymore. The global instance returned by get isn't lazily initialized because it'd require a memory barrier to read the field. It uses a regular val loaded when the companion object class is loaded.

fwbrasil · 2024-04-22T23:01:18Z

kyo-core/shared/src/main/scala/kyo/scheduler/IOTask.scala

- @volatile private var state: Int // Math.abs(state) => runtime; state < 0 => preempting
-) extends IOPromise[T] with Task
+ initialRuntime: Int
+) extends IOPromise[T] with Task(initialRuntime)


Runtime and preemption management are now done by Task.

fwbrasil · 2024-04-22T23:05:50Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/InternalClock.scala

+import java.util.concurrent.Executor
+import java.util.concurrent.locks.LockSupport
+
+final private class InternalClock(executor: Executor):


Task's runtime tracking has to measure execution times and obtaining the time from the system clock is prohibitively expensive. This internal clock provides approximate time measurements by updating a volatile field every ~1ms. Readers only pay the price of the read barrier and eventual cache misses when the field is updated.

fwbrasil · 2024-04-22T23:06:15Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/InternalTimer.scala

+import java.util.concurrent.TimeUnit
+import scala.concurrent.duration.Duration
+
+abstract private class InternalTimer:


Introduced for testability.

fwbrasil · 2024-04-22T23:10:01Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Queue.scala


 private val queue = PriorityQueue[T]()

- @volatile private var items = 0


This is a micro optimization. If items is marked as volatile, methods like add have to use memory barriers to update it within a modify/tryModify block, which is unnecessary since there's already a write barrier at the end. to release the lock.

fwbrasil · 2024-04-22T23:10:28Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Queue.scala


 def size(): Int =
+ VarHandle.acquireFence()
 items


Force a read barrier since the field isn't volatile anymore so all threads see the latest value.

fwbrasil · 2024-04-22T23:16:11Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Queue.scala

@@ -61,7 62,7 @@ final private[kyo] class Queue[T](using ord: Ordering[T]) extends AtomicBoolean:
 !isEmpty() && to.isEmpty() && to.tryModify {
 t = queue.dequeue()
 val s = size() - 1
- var i = s - (s / 2)
+ var i = s - Math.ceil(s.toDouble / 2).intValue()


When a worker had 3 tasks, this code was stealing 2 for example. This new logic makes it steal only 1. Leaving a worker with only one task increases the likelihood of it'll soon go to sleep again. If a thief is successful, it'll naturally try stealing again when it's out of tasks.

fwbrasil · 2024-04-22T23:16:50Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Queue.scala

+ items = 0
+ queue.dequeueAll
+ }
+ tasks.foreach(f)


This code was unnecessarily holding the lock when executing the drain function.

fwbrasil · 2024-04-22T23:18:32Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/util/MovingStdDev.scala

@@ -0,0 1,39 @@
+package kyo.scheduler.util
+
+final private[kyo] class MovingStdDev(window: Int):


Since this class isn't used in a very hot path anymore and is called only by regulators that execute periodically, it was updated to favor precision over performance.

fwbrasil · 2024-04-22T23:21:37Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Scheduler.scala

+
+ val a1, a2, a3, a4, a5, a6, a7 = 0L // padding
+
+ @volatile private var cycles = 0L


This field used to be in Coordinator.

fwbrasil · 2024-04-22T23:22:11Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Scheduler.scala

+ ExecutionContext.fromExecutor(asExecutor)
+
+ @tailrec
+ private def schedule(t: Task, submitter: Worker): Unit =


These scheduling methods have not changed.

fwbrasil · 2024-04-22T23:22:30Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Scheduler.scala

+ workers(idx) = new Worker(idx, pool, schedule, steal, () => cycles, clock)
+ allocatedWorkers = 1
+
+ private val cycleTask =


Moved from Coordinator

fwbrasil · 2024-04-22T23:23:21Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Scheduler.scala

+ val w = workers(i)
+ if w != null then
+ if i >= maxConcurrency then
+ w.drain()


Improved logic to continue cycling all workers but also draining workers that have been stopped (are above the concurrency limit)

fwbrasil · 2024-04-22T23:25:44Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Worker.scala

+
+ val a1, a2, a3, a4, a5, a6, a7 = 0L // padding
+
+ @volatile private var running = false


This class hasn't changed much but I've made running a volatile that is updated using VarHandle as a micro optimization to avoid the pointer chasing with AtomicBoolean since the flag used in the hot path of task execution.

fwbrasil · 2024-04-22T23:26:06Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Worker.scala

+ val state = m.getState().ordinal()
+ state == Thread.State.BLOCKED.ordinal() ||
+ state == Thread.State.WAITING.ordinal() ||
+ state == Thread.State.TIMED_WAITING.ordinal()


Previous implementation was missing some of these states.

hearnadam

Looks like a huge improvement! Nice work!

hearnadam · 2024-04-23T03:17:53Z

kyo-scheduler/js/src/main/scala/kyo/scheduler/Scheduler.scala

@@ -3,13 3,19 @@ package kyo.scheduler
 import scala.scalajs.concurrent.JSExecutionContext

 object Scheduler:
+ lazy val get = new Scheduler


I don't think this needs to be lazy

hearnadam · 2024-04-23T03:18:16Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/InternalClock.scala

+import java.util.concurrent.Executor
+import java.util.concurrent.locks.LockSupport
+
+final private class InternalClock(executor: Executor):


hearnadam · 2024-04-23T16:32:07Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/InternalTimer.scala

+
+ def scheduleOnce(delay: Duration)(f: => Unit): TimerTask =
+ val future = executor.schedule((() => f): Runnable, delay.toNanos, TimeUnit.NANOSECONDS)
+ new TimerTask:


Is it possible to avoid this allocation?

You're getting used to Kyo's focus on performance :) In this case, I don't think it's necessary because only two tasks are submitted when the scheduler starts. It'd also be more difficult to mock this class in tests if we used an opaque type to alias ScheduledFuture.

hearnadam · 2024-04-23T16:37:03Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/regulator/Regulator.scala

+ end if
+ catch
+ case ex if NonFatal(ex) =>
+ log.error(s"🙈 !!Kyo Scheduler Bug!! ${getClass.getSimpleName()} regulator's probe collection has failed.", ex)


now that we are using a logger instance for this class, we can remove getClass.getSimpleName()

good catch!

ah actually I'm not using Kyo's logging because the idea is for the scheduler module to be isolated without a dependency to kyo-core. This is the regular slf4j logger.

Right, but you are using SLF4J instance for each class which should include the classname. See line 92:

private[Regulator] val log = LoggerFactory.getLogger(getClass)

This goes throughout

The log in this case would be Regulator, not the subclass. But I've decided to remove the slf4j dependency since logging is used only for bugs. There's the LoomSupport warning but I think it's ok to log that with a regular println. The module now has zero dependencies, which should help avoid issues in case it's used in isolation without other Kyo modules.

Cool emoji!

…rics

…ings

fwbrasil · 2024-04-23T18:11:24Z

Benchmark results with the new scheduler:

The results are mixed and the RandomBench improvement seems just noise. I think the results are reasonable, though. The scheduler now does much more handling blocking, adjusting concurrency, and providing a backpressure signal. Not having a significant regression is already a good win and we can keep optimizing it over time.

@hearnadam would you be able to check out this branch and run this main? It checks that the regulators are working properly, it'd be nice to get data from other machines as well. I'm doing some more testing using containers and CPU quotas to validate the behavior with CPU throttling and I'm planning to merge this PR if everything looks ok.

hearnadam · 2024-04-23T18:20:18Z

@fwbrasil will take a look at running this tonight.

fwbrasil · 2024-04-23T19:02:18Z

Self-check is ok with CPU quotas as well:

fwbrasil@flavios-mbp kyo % docker run --rm --cpus=1 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.02040816326530612, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.8846153846153846, rejectionThreshold -> 0.2)
Success
fwbrasil@flavios-mbp kyo % docker run --rm --cpus=2 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 3, rejectionPercent -> 1.2727272727272727, rejectionThreshold -> 0.2)
Success
fwbrasil@flavios-mbp kyo % docker run --rm --cpus=3 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 3, rejectionPercent -> 0.06521739130434782, rejectionThreshold -> 0.2)
Map(clients -> 4, rejectionPercent -> 0.21951219512195122, rejectionThreshold -> 0.2)
Success
fwbrasil@flavios-mbp kyo % docker run --rm --cpus=4 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 4, rejectionPercent -> 0.06521739130434782, rejectionThreshold -> 0.2)
Map(clients -> 5, rejectionPercent -> 0.21951219512195122, rejectionThreshold -> 0.2)
Success
fwbrasil@flavios-mbp kyo % docker run --rm --cpus=5 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 4, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 5, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 6, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 7, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 8, rejectionPercent -> 0.020833333333333332, rejectionThreshold -> 0.2)
Map(clients -> 9, rejectionPercent -> 1.5, rejectionThreshold -> 0.2)
Success
fwbrasil@flavios-mbp kyo % docker run --rm --cpus=6 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 4, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 5, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 6, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 7, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 8, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 9, rejectionPercent -> 1.380952380952381, rejectionThreshold -> 0.2)
Success
fwbrasil@flavios-mbp kyo % docker run --rm --cpus=7 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 4, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 5, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 6, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 7, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 8, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 9, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 10, rejectionPercent -> 0.1951219512195122, rejectionThreshold -> 0.2)
Map(clients -> 11, rejectionPercent -> 0.7241379310344828, rejectionThreshold -> 0.2)
Success
fwbrasil@flavios-mbp kyo % docker run --rm --cpus=8 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 4, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 5, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 6, rejectionPercent -> 0.041666666666666664, rejectionThreshold -> 0.2)
Map(clients -> 7, rejectionPercent -> 0.53125, rejectionThreshold -> 0.2)
Success
fwbrasil@flavios-mbp kyo % docker run --rm --cpus=9 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 4, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 5, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 6, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 7, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 8, rejectionPercent -> 0.0425531914893617, rejectionThreshold -> 0.2)
Map(clients -> 9, rejectionPercent -> 1.0, rejectionThreshold -> 0.2)
Success
fwbrasil@flavios-mbp kyo % docker run --rm --cpus=10 -v /Users/fwbrasil/workspace/kyo/kyo-scheduler/jvm/target/scala-3.4.1:/app adoptopenjdk/openjdk11 java -cp /app/kyo-scheduler-assembly-0.9.2 71-82298105 20240423-1138-SNAPSHOT.jar kyo.scheduler.util.SelfCheckMain
Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 4, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 5, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 6, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 7, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
Map(clients -> 8, rejectionPercent -> 0.041666666666666664, rejectionThreshold -> 0.2)
Map(clients -> 9, rejectionPercent -> 0.16666666666666666, rejectionThreshold -> 0.2)
Map(clients -> 10, rejectionPercent -> 1.2727272727272727, rejectionThreshold -> 0.2)
Success

…when Loom is disabled

fwbrasil · 2024-04-23T22:15:02Z

I've made a few new optimizations and the benchmark results are reporting more consistent good results:

The main cause for the regression in ForkSpawnBench is the path for the fiber to check for the preemption signal involving more code now that Task manages the preemption instead of IOTask. Since Task is a trait, Scala encodes methods in the companion object and accessing the state field has to go through a getter.

hearnadam · 2024-04-24T00:46:31Z

@fwbrasil

sbt:kyo-scheduler> runMain kyo.scheduler.util.SelfCheckMain
[info] running (fork) kyo.scheduler.util.SelfCheckMain 
[info] Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 4, rejectionPercent -> 0.02127659574468085, rejectionThreshold -> 0.2)
[info] Map(clients -> 5, rejectionPercent -> 0.75, rejectionThreshold -> 0.2)
[info] Failure: Expected between 6.4 and 14.0 clients for 8 cores but found 5.
[success] Total time: 25 s, completed Apr 23, 2024, 5:45:49 PM
sbt:kyo-scheduler> runMain kyo.scheduler.util.SelfCheckMain
[info] running (fork) kyo.scheduler.util.SelfCheckMain 
[info] Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 4, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 5, rejectionPercent -> 0.14285714285714285, rejectionThreshold -> 0.2)
[info] Map(clients -> 6, rejectionPercent -> 2.0625, rejectionThreshold -> 0.2)
[info] Failure: Expected between 6.4 and 14.0 clients for 8 cores but found 6.
[success] Total time: 30 s, completed Apr 23, 2024, 5:47:04 PM
sbt:kyo-scheduler> runMain kyo.scheduler.util.SelfCheckMain
[info] running (fork) kyo.scheduler.util.SelfCheckMain 
[info] Map(clients -> 0, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 1, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 2, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 3, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 4, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 5, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 6, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 7, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 8, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 9, rejectionPercent -> 0.0, rejectionThreshold -> 0.2)
[info] Map(clients -> 10, rejectionPercent -> 0.1951219512195122, rejectionThreshold -> 0.2)
[info] Map(clients -> 11, rejectionPercent -> 4.333333333333333, rejectionThreshold -> 0.2)
[info] Success

On: 6e838b14047e9eb0f57caab112c34ba49196ef2b

fwbrasil · 2024-04-24T06:15:59Z

@hearnadam can you check if you have other processes that could be using some of the CPU?

sideeffffect · 2024-04-24T11:15:10Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/package.scala

+ private def bug(msg: String, ex: Throwable) =
+ (new Exception("🙈 !!Kyo Scheduler Bug!! " msg, ex)).printStackTrace(System.err)


Suggested change

private def bug(msg: String, ex: Throwable) =

(new Exception("🙈 !!Kyo Scheduler Bug!! " msg, ex)).printStackTrace(System.err)

import java.util.logging.*

private lazy val logger = Logger.getLogger(getClass.getName)

private def bug(msg: String, ex: Throwable) =

logger.log(Level.SEVERE, "🙈 !!Kyo Scheduler Bug!!", new Exception(msg, ex))

Since you want 0 dependencies, we can still use java.util.logging, since that's part of the standard library. This will give people better interoperability with other logging frameworks or other tools hooked on logging, like Sentry.

good idea! 🙏

You can define your own logging and then there will be a adapter.

Netty 5 is using slf4j again

This reverts commit e762687.

fwbrasil · 2024-04-24T17:37:47Z

I've changed the code use java logging as @sideeffffect suggested. I've also tried another optimization moving the preemption handling from Task to Worker but the results weren't good because obtaining the current worker is more expensive.

I'm merging this PR since it keeps getting larger but please feel free to provide feedback on the change!

He-Pin · 2024-04-24T17:35:09Z

kyo-scheduler/js/src/main/scala/kyo/scheduler/InternalClock.scala

+@nowarn
+final class InternalClock(executor: Executor = null):
+
+ def currentMillis(): Long = System.currentTimeMillis()


Should it be nano time?

System.nanoTime isn't monotonic across threads. It also helps to hide the imprecision of the clock.

He-Pin · 2024-04-24T17:36:16Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/InternalClock.scala

+
+ @volatile private var millis = System.currentTimeMillis()
+
+ val b1, b2, b3, b4, b5, b6, b7 = 0L // padding


Cool padding

He-Pin · 2024-04-24T17:42:00Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/Scheduler.scala

+ i = 1
+ end while
+ if worker != null then
+ worker.steal(thief)


Worker stealing thief not thief stealing worker?

That's thief taking tasks from worker but yeah, the API isn't very clear.

Maybe stealingBy or divisionBy

He-Pin · 2024-04-24T17:45:32Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/util/LoomSupport.scala

+
+ classOf[Executors]
+ .getMethod("newThreadPerTaskExecutor", classOf[ThreadFactory])
+ .invoke(null, factory).asInstanceOf[Executor]


Cool, I just did the same too
toohttps://github.com/apache/rocketmq/pull/8063

So you don't plan to pooling the virtual thread right?

Cool! I tried polling but putting virtual threads to sleep is too expensive and it's cheaper to allocate new ones. Virtual threads are allocated only when workers start, though. A worker mounted on a virtual thread can execute several tasks until there's no more pending work, which avoids the cost of the virtual thread creation on each task execution like a plain newThreadPerTaskExecutor does.

Yeah, If totally move to none pooling will need to change a lot of the current work base at work, where thread pool is using like some kind of concurrency limiter too

sideeffffect · 2024-04-25T08:38:47Z

kyo-scheduler/jvm/src/main/scala/kyo/scheduler/package.scala

+ "kyo" :: "scheduler" :: path.toList
+
+ private def bug(msg: String, ex: Throwable) =
+ log.severe(s"🙈 !!Kyo Scheduler Bug!! $msg \n Caused by: ${stackTrace(ex)}")


Just curious, why not use the overload of this method where you can pass the Throwable as an argument?

I was hoping that the logging backend will do the printing of the Throwable's stack trace for us, instead of us having to do it manually with out own ad hoc method (private def stackTrace(ex: Throwable)).

Supplying an intact Throwable instance to the logger will also have better interop with 3rd parties. (You can always turn a Throwable to String, but you can't go the other way around.)

fwbrasil commented Apr 22, 2024

View reviewed changes

hearnadam reviewed Apr 23, 2024

View reviewed changes

fwbrasil added 13 commits April 23, 2024 10:03

more scheduler to new module with no dependencies

db9eab2

avoid depending on Coordinator outside of the scheduler

8a8d581

add logs back

15df4c2

move metrics receiver to scheduler module reintroduce scheduler met…

df71a53

…rics

refactor scheduler classes for testing

355296c

refactor scheduler classes for testing

ca102d8

admission scheduler.asExecutor/asExecutionContext config refactor…

7c564a0

…ings

introducing scheduler regulators

52abfaf

refactorings

c8eb122

queue optimizations and tests

b5a1a8e

worker thread state fix tests

a18349e

move queue to scheduler package

3d35f47

scheduler tests

8b5a78c

fwbrasil added 5 commits April 23, 2024 11:23

remove unused dependency

37e01b7

remove admission probe cache and jctools dependency

04b5ab6

zero-dependency scheduler module (remove slf4j dep)

8229810

more convenient self-check main

45762b1

relax self-check thresholds

3c00a87

fwbrasil added 5 commits April 23, 2024 12:51

minor refactorings

7ce6395

optimization: avoid thread local lookup to obtain the current worker …

da7c47f

…when Loom is disabled

avoid initialRuntime field in Task/IOTask

bc762a9

avoid instanceof Object in core.transform

a55cdcb

fix tests

6e838b1

sideeffffect reviewed Apr 24, 2024

View reviewed changes

fwbrasil added 5 commits April 24, 2024 09:11

move preemption handling from Task to Worker

e762687

Revert "move preemption handling from Task to Worker"

5331241

This reverts commit e762687.

task ordering fix test

3218aeb

test task ordering with preemption

2592980

use java logging fix flaky test

5952bc5

fwbrasil merged commit fd554d5 into main Apr 24, 2024
3 checks passed

fwbrasil deleted the scheduler-review branch April 24, 2024 17:38

He-Pin reviewed Apr 24, 2024

View reviewed changes

fwbrasil mentioned this pull request Apr 24, 2024

improve scheduler readability #283

Merged

sideeffffect reviewed Apr 25, 2024

View reviewed changes

sideeffffect mentioned this pull request Apr 25, 2024

Log the exception in logger #286

Merged

fwbrasil mentioned this pull request May 3, 2024

fix io task runtime tracking #314

Merged


		private val queue = PriorityQueue[T]()

		@volatile private var items = 0

		@@ -0,0 1,39 @@
		package kyo.scheduler.util

		final private[kyo] class MovingStdDev(window: Int):


		val a1, a2, a3, a4, a5, a6, a7 = 0L // padding

		@volatile private var cycles = 0L


		val a1, a2, a3, a4, a5, a6, a7 = 0L // padding

		@volatile private var running = false

		private def bug(msg: String, ex: Throwable) =
		(new Exception("🙈 !!Kyo Scheduler Bug!! " msg, ex)).printStackTrace(System.err)


		@volatile private var millis = System.currentTimeMillis()

		val b1, b2, b3, b4, b5, b6, b7 = 0L // padding

New Scheduler #278

New Scheduler #278

Conversation

fwbrasil commented Apr 22, 2024

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

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

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

hearnadam left a 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

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

Choose a reason for hiding this comment

fwbrasil commented Apr 23, 2024 • edited Loading

hearnadam commented Apr 23, 2024

fwbrasil commented Apr 23, 2024

fwbrasil commented Apr 23, 2024

hearnadam commented Apr 24, 2024 • edited Loading

fwbrasil commented Apr 24, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fwbrasil commented Apr 24, 2024

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

He-Pin Apr 24, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fwbrasil Apr 24, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fwbrasil commented Apr 23, 2024 •

edited

Loading

hearnadam commented Apr 24, 2024 •

edited

Loading

He-Pin Apr 24, 2024 •

edited

Loading

fwbrasil Apr 24, 2024 •

edited

Loading