[scheduler] scala native support

getkyo · Nov 26, 2024 · f362727 · f362727
1 parent 70f91b4
commit f362727
Show file tree

Hide file tree

Showing 35 changed files with 51 additions and 19 deletions.
diff --git a/build.sbt b/build.sbt
@@ -141,7 +141,7 @@ lazy val kyoNative = project
     )
 
 lazy val `kyo-scheduler` =
-    crossProject(JSPlatform, JVMPlatform)
+    crossProject(JSPlatform, JVMPlatform, NativePlatform)
         .withoutSuffixFor(JVMPlatform)
         .crossType(CrossType.Full)
         .dependsOn(`kyo-stats-registry`)
@@ -154,6 +154,7 @@ lazy val `kyo-scheduler` =
             libraryDependencies += "ch.qos.logback"  % "logback-classic" % "1.5.12"         % Test
         )
         .jvmSettings(mimaCheck(false))
+        .nativeSettings(`native-settings`)
         .jsSettings(
             `js-settings`,
             libraryDependencies += "org.scala-js" %%% "scala-js-macrotask-executor" % "1.1.1"
@@ -241,7 +242,7 @@ lazy val `kyo-direct` =
         .jsSettings(`js-settings`)
 
 lazy val `kyo-stats-registry` =
-    crossProject(JSPlatform, JVMPlatform)
+    crossProject(JSPlatform, JVMPlatform, NativePlatform)
         .withoutSuffixFor(JVMPlatform)
         .crossType(CrossType.Full)
         .in(file("kyo-stats-registry"))
@@ -253,6 +254,7 @@ lazy val `kyo-stats-registry` =
             crossScalaVersions                       := List(scala3Version, scala212Version, scala213Version)
         )
         .jvmSettings(mimaCheck(false))
+        .nativeSettings(`native-settings`)
         .jsSettings(`js-settings`)
 
 lazy val `kyo-stats-otel` =

diff --git a/...n/scala/kyo/scheduler/InternalClock.scala → ...n/scala/kyo/scheduler/InternalClock.scala b/...n/scala/kyo/scheduler/InternalClock.scala → ...n/scala/kyo/scheduler/InternalClock.scala
diff --git a/...n/scala/kyo/scheduler/InternalTimer.scala → ...n/scala/kyo/scheduler/InternalTimer.scala b/...n/scala/kyo/scheduler/InternalTimer.scala → ...n/scala/kyo/scheduler/InternalTimer.scala
diff --git a/.../src/main/scala/kyo/scheduler/Queue.scala → .../src/main/scala/kyo/scheduler/Queue.scala b/.../src/main/scala/kyo/scheduler/Queue.scala → .../src/main/scala/kyo/scheduler/Queue.scala
diff --git a/.../main/scala/kyo/scheduler/Scheduler.scala → .../main/scala/kyo/scheduler/Scheduler.scala b/.../main/scala/kyo/scheduler/Scheduler.scala → .../main/scala/kyo/scheduler/Scheduler.scala
diff --git a/...src/main/scala/kyo/scheduler/Worker.scala → ...src/main/scala/kyo/scheduler/Worker.scala b/...src/main/scala/kyo/scheduler/Worker.scala → ...src/main/scala/kyo/scheduler/Worker.scala
@@ -4,6 +4,7 @@ import Worker.State
 import java.lang.invoke.MethodHandles
 import java.lang.invoke.VarHandle
 import java.util.concurrent.Executor
+import java.util.concurrent.atomic.AtomicReference
 import java.util.concurrent.atomic.LongAdder
 import kyo.scheduler.top.WorkerStatus
 import scala.annotation.nowarn
@@ -92,7 +93,8 @@ abstract private class Worker(
 
     val a1, a2, a3, a4, a5, a6, a7 = 0L // padding
 
-    @volatile private var state: State  = State.Idle
+    private val state = new AtomicReference[State](State.Idle)
+
     @volatile private var mount: Thread = null
 
     val b1, b2, b3, b4, b5, b6, b7 = 0L // padding
@@ -125,7 +127,7 @@ abstract private class Worker(
       * arrives for an idle worker.
       */
     def wakeup() = {
-        if ((state eq State.Idle) && stateHandle.compareAndSet(this, State.Idle, State.Running))
+        if ((state.get() eq State.Idle) && state.compareAndSet(State.Idle, State.Running))
             exec.execute(this)
     }
 
@@ -163,9 +165,9 @@ abstract private class Worker(
       * If checks fail while Running, transitions to Stalled and drains queue. Used by scheduler to skip workers that can't make progress.
       */
     def checkAvailability(nowMs: Long): Boolean = {
-        val state     = this.state
-        val available = !checkStalling(nowMs) && (state ne State.Stalled) && !isBlocked()
-        if (!available && (state eq State.Running) && stateHandle.compareAndSet(this, State.Running, State.Stalled))
+        val st        = this.state.get()
+        val available = !checkStalling(nowMs) && (st ne State.Stalled) && !isBlocked()
+        if (!available && (st eq State.Running) && state.compareAndSet(State.Running, State.Stalled))
             drain()
         available
     }
@@ -199,7 +201,7 @@ abstract private class Worker(
 
         while (true) {
             // Mark worker as actively running
-            state = State.Running
+            state.set(State.Running)
 
             if (task eq null)
                 // Try to get a task from our own queue first
@@ -227,8 +229,8 @@ abstract private class Worker(
                 }
             } else {
                 // No tasks available - prepare to go idle
-                state = State.Idle
-                if (queue.isEmpty() || !stateHandle.compareAndSet(this, State.Idle, State.Running)) {
+                state.set(State.Idle)
+                if (queue.isEmpty() || !state.compareAndSet(State.Idle, State.Running)) {
                     // Either queue is empty or another thread changed our state
                     // Clean up and exit
                     mount = null
@@ -239,7 +241,7 @@ abstract private class Worker(
 
             // Check if we should stop processing tasks
             if (shouldStop()) {
-                state = State.Idle
+                state.set(State.Idle)
                 // Reschedule current task if we have one
                 if (task ne null) schedule(task)
                 // Drain remaining tasks from queue
@@ -320,10 +322,6 @@ private object Worker {
 
     private[Worker] object internal {
 
-        val stateHandle: VarHandle =
-            MethodHandles.privateLookupIn(classOf[Worker], MethodHandles.lookup())
-                .findVarHandle(classOf[Worker], "state", classOf[State])
-
         val local = new ThreadLocal[Worker]
 
         def setCurrent(worker: Worker): Unit =

diff --git a/...rc/main/scala/kyo/scheduler/package.scala → ...rc/main/scala/kyo/scheduler/package.scala b/...rc/main/scala/kyo/scheduler/package.scala → ...rc/main/scala/kyo/scheduler/package.scala
@@ -1,16 +1,13 @@
 package kyo
 
-import java.util.logging.*
 import kyo.stats.internal.StatsRegistry
 
 package object scheduler {
 
-    private[scheduler] val log = Logger.getLogger("kyo.scheduler")
-
     private[scheduler] def statsScope =
         StatsRegistry.scope("kyo", "scheduler")
 
     private[scheduler] def bug(msg: String, ex: Throwable) =
-        log.log(Level.SEVERE, s"🙈 !!Kyo Scheduler Bug!!", new Exception(msg, ex))
+        new Exception(s"🙈 !!Kyo Scheduler Bug!! " + msg, ex).printStackTrace()
 
 }
diff --git a/...a/kyo/scheduler/regulator/Admission.scala → ...a/kyo/scheduler/regulator/Admission.scala b/...a/kyo/scheduler/regulator/Admission.scala → ...a/kyo/scheduler/regulator/Admission.scala
diff --git a/...kyo/scheduler/regulator/Concurrency.scala → ...kyo/scheduler/regulator/Concurrency.scala b/...kyo/scheduler/regulator/Concurrency.scala → ...kyo/scheduler/regulator/Concurrency.scala
diff --git a/...cala/kyo/scheduler/regulator/Config.scala → ...cala/kyo/scheduler/regulator/Config.scala b/...cala/kyo/scheduler/regulator/Config.scala → ...cala/kyo/scheduler/regulator/Config.scala
diff --git a/...a/kyo/scheduler/regulator/Regulator.scala → ...a/kyo/scheduler/regulator/Regulator.scala b/...a/kyo/scheduler/regulator/Regulator.scala → ...a/kyo/scheduler/regulator/Regulator.scala
diff --git a/...ain/scala/kyo/scheduler/top/Printer.scala → ...ain/scala/kyo/scheduler/top/Printer.scala b/...ain/scala/kyo/scheduler/top/Printer.scala → ...ain/scala/kyo/scheduler/top/Printer.scala
diff --git a/...main/scala/kyo/scheduler/top/Status.scala → ...main/scala/kyo/scheduler/top/Status.scala b/...main/scala/kyo/scheduler/top/Status.scala → ...main/scala/kyo/scheduler/top/Status.scala
diff --git a/.../main/scala/kyo/scheduler/util/Flag.scala → .../main/scala/kyo/scheduler/util/Flag.scala b/.../main/scala/kyo/scheduler/util/Flag.scala → .../main/scala/kyo/scheduler/util/Flag.scala
diff --git a/...ala/kyo/scheduler/util/MovingStdDev.scala → ...ala/kyo/scheduler/util/MovingStdDev.scala b/...ala/kyo/scheduler/util/MovingStdDev.scala → ...ala/kyo/scheduler/util/MovingStdDev.scala
diff --git a/.../scala/kyo/scheduler/util/SelfCheck.scala → .../scala/kyo/scheduler/util/SelfCheck.scala b/.../scala/kyo/scheduler/util/SelfCheck.scala → .../scala/kyo/scheduler/util/SelfCheck.scala
diff --git a/...in/scala/kyo/scheduler/util/Threads.scala → ...in/scala/kyo/scheduler/util/Threads.scala b/...in/scala/kyo/scheduler/util/Threads.scala → ...in/scala/kyo/scheduler/util/Threads.scala
diff --git a/...n/scala/kyo/scheduler/util/XSRandom.scala → ...n/scala/kyo/scheduler/util/XSRandom.scala b/...n/scala/kyo/scheduler/util/XSRandom.scala → ...n/scala/kyo/scheduler/util/XSRandom.scala
diff --git a/...ala/kyo/scheduler/InternalClockTest.scala → ...ala/kyo/scheduler/InternalClockTest.scala b/...ala/kyo/scheduler/InternalClockTest.scala → ...ala/kyo/scheduler/InternalClockTest.scala
diff --git a/...ala/kyo/scheduler/InternalTimerTest.scala → ...ala/kyo/scheduler/InternalTimerTest.scala b/...ala/kyo/scheduler/InternalTimerTest.scala → ...ala/kyo/scheduler/InternalTimerTest.scala
diff --git a/.../test/scala/kyo/scheduler/QueueTest.scala → .../test/scala/kyo/scheduler/QueueTest.scala b/.../test/scala/kyo/scheduler/QueueTest.scala → .../test/scala/kyo/scheduler/QueueTest.scala
diff --git a/...t/scala/kyo/scheduler/SchedulerTest.scala → ...t/scala/kyo/scheduler/SchedulerTest.scala b/...t/scala/kyo/scheduler/SchedulerTest.scala → ...t/scala/kyo/scheduler/SchedulerTest.scala
diff --git a/...c/test/scala/kyo/scheduler/TestTask.scala → ...c/test/scala/kyo/scheduler/TestTask.scala b/...c/test/scala/kyo/scheduler/TestTask.scala → ...c/test/scala/kyo/scheduler/TestTask.scala
diff --git a/.../test/scala/kyo/scheduler/TestTimer.scala → .../test/scala/kyo/scheduler/TestTimer.scala b/.../test/scala/kyo/scheduler/TestTimer.scala → .../test/scala/kyo/scheduler/TestTimer.scala
diff --git a/...test/scala/kyo/scheduler/WorkerTest.scala → ...test/scala/kyo/scheduler/WorkerTest.scala b/...test/scala/kyo/scheduler/WorkerTest.scala → ...test/scala/kyo/scheduler/WorkerTest.scala
diff --git a/...o/scheduler/regulator/AdmissionTest.scala → ...o/scheduler/regulator/AdmissionTest.scala b/...o/scheduler/regulator/AdmissionTest.scala → ...o/scheduler/regulator/AdmissionTest.scala
diff --git a/...scheduler/regulator/ConcurrencyTest.scala → ...scheduler/regulator/ConcurrencyTest.scala b/...scheduler/regulator/ConcurrencyTest.scala → ...scheduler/regulator/ConcurrencyTest.scala
diff --git a/...o/scheduler/regulator/RegulatorTest.scala → ...o/scheduler/regulator/RegulatorTest.scala b/...o/scheduler/regulator/RegulatorTest.scala → ...o/scheduler/regulator/RegulatorTest.scala
diff --git a/...kyo/scheduler/util/MovingStdDevTest.scala → ...kyo/scheduler/util/MovingStdDevTest.scala b/...kyo/scheduler/util/MovingStdDevTest.scala → ...kyo/scheduler/util/MovingStdDevTest.scala
diff --git a/...ala/kyo/scheduler/util/XSRandomTest.scala → ...ala/kyo/scheduler/util/XSRandomTest.scala b/...ala/kyo/scheduler/util/XSRandomTest.scala → ...ala/kyo/scheduler/util/XSRandomTest.scala
diff --git a/kyo-scheduler/native/src/main/scala/kyo/scheduler/top/Reporter.scala b/kyo-scheduler/native/src/main/scala/kyo/scheduler/top/Reporter.scala
@@ -0,0 +1,24 @@
+package kyo.scheduler.top
+
+import java.lang.management.ManagementFactory
+import javax.management.MBeanServer
+import javax.management.ObjectName
+import javax.management.StandardMBean
+import kyo.scheduler.InternalTimer
+import scala.concurrent.duration.*
+
+class Reporter(
+    status: () => Status,
+    enableTopJMX: Boolean,
+    enableTopConsoleMs: Int,
+    timer: InternalTimer
+) extends TopMBean {
+
+    def getStatus() = status()
+
+    def close(): Unit = ()
+}
+
+trait TopMBean {
+    def getStatus(): Status
+}
diff --git a/kyo-scheduler/native/src/main/scala/kyo/scheduler/util/LoomSupport.scala b/kyo-scheduler/native/src/main/scala/kyo/scheduler/util/LoomSupport.scala
@@ -0,0 +1,11 @@
+package kyo.scheduler.util
+
+import java.util.concurrent.Executor
+import java.util.concurrent.Executors
+import java.util.concurrent.ThreadFactory
+import java.util.logging.Logger
+import scala.util.control.NonFatal
+
+object LoomSupport {
+    def tryVirtualize(enabled: Boolean, exec: Executor): Executor = exec
+}
diff --git a/...n/scala/java/lang/ref/WeakReference.scala → ...n/scala/java/lang/ref/WeakReference.scala b/...n/scala/java/lang/ref/WeakReference.scala → ...n/scala/java/lang/ref/WeakReference.scala
diff --git a/...ala/kyo/stats/internal/StatsRefresh.scala → ...ala/kyo/stats/internal/StatsRefresh.scala b/...ala/kyo/stats/internal/StatsRefresh.scala → ...ala/kyo/stats/internal/StatsRefresh.scala
diff --git a/.../kyo/stats/internal/UnsafeHistogram.scala → .../kyo/stats/internal/UnsafeHistogram.scala b/.../kyo/stats/internal/UnsafeHistogram.scala → .../kyo/stats/internal/UnsafeHistogram.scala