WIP

Author: adamw

build.sbt

@@ -20,6 +20,7 @@ lazy val core: Project = (project in file("core"))
   .settings(
     name := "core",
     libraryDependencies ++= Seq(
+      "org.jctools" % "jctools-core" % "4.0.1",
       scalaTest
     )
   )

core/src/main/scala/ox/channels2/Channel.scala

@@ -1,35 +1,45 @@
 package ox.channels2
 
-import java.util.concurrent.{ArrayBlockingQueue, ConcurrentSkipListSet, CountDownLatch}
+import java.util.concurrent.{ArrayBlockingQueue, ConcurrentSkipListSet, CountDownLatch, Semaphore, SynchronousQueue}
 import java.util.concurrent.atomic.{AtomicLong, AtomicReference, AtomicReferenceArray}
+import java.util.concurrent.locks.LockSupport
 import scala.annotation.tailrec
 
 class Channel[T]:
   private val senders = new AtomicLong(0L)
   private val receivers = new AtomicLong(0L)
   private val buffer: AtomicReferenceArray[State] = AtomicReferenceArray[State](20_000_000) // TODO
-  
+
   //
 
   private object Interrupted
   private class Continuation[E]():
-    private val q = new ArrayBlockingQueue[E | Interrupted.type](1)
+    private val creatingThread = Thread.currentThread()
+    private val data = new AtomicReference[E | Interrupted.type | Null]()
 
-    def tryResume(e: E): Boolean = q.offer(e)
+    def tryResume(e: E): Boolean =
+      val result = data.compareAndSet(null, e)
+      LockSupport.unpark(creatingThread)
+      result
 
     def await(onInterrupt: () => Unit): E =
-      try
-        q.take().asInstanceOf[E] // can't be anything else
-      catch
-        case e: InterruptedException =>
-          if q.offer(Interrupted)
-          then
-            try onInterrupt()
-            catch case ee: Throwable => e.addSuppressed(ee)
-            throw e
-          else
-            Thread.currentThread().interrupt() // propagating the interruption to the next blocking call
-            q.poll().asInstanceOf[E] // another thread has just inserted E
+      var x = 0
+      while data.get() == null do
+        if x <= 0 then
+          Thread.`yield`()
+          x += 1
+        else LockSupport.park()
+
+        if Thread.interrupted() then
+          data.compareAndSet(null, Interrupted) // TODO if
+          val e = new InterruptedException()
+
+          try onInterrupt()
+          catch case ee: Throwable => e.addSuppressed(ee)
+
+          throw e
+
+      data.get().asInstanceOf[E] // another thread has just inserted E
 
   //
 
@@ -134,4 +144,4 @@ class Channel[T]:
       fork(s.add(c.receive()))
     }
   }
-  println(s.size)
+  println(s.size)

core/src/main/scala/ox/channels2/perf/impls.scala

@@ -3,11 +3,8 @@ package ox.channels2.perf
 import ox.*
 import ox.channels2.*
 
+import java.util.concurrent.SynchronousQueue
 import java.util.concurrent.atomic.{AtomicLong, AtomicReferenceArray}
-import java.util.concurrent.{ArrayBlockingQueue, ConcurrentLinkedDeque, Executors}
-import scala.annotation.tailrec
-import scala.concurrent.Await
-import scala.concurrent.duration.*
 
 def usingOx(): Unit =
   val max = 10_000_000
@@ -42,7 +39,7 @@ def passingValues(): Unit =
   }
 
   val max = 10_000_000
-  val st = new AtomicReferenceArray[ArrayBlockingQueue[String]](max)
+  val st = new AtomicReferenceArray[SynchronousQueue[String]](max)
   val s = new AtomicLong(0)
   val r = new AtomicLong(0)
   timed("passingValues") {
@@ -51,23 +48,23 @@ def passingValues(): Unit =
         for (i <- 0 until max) {
           s.incrementAndGet()
           r.get()
-          val q = new ArrayBlockingQueue[String](1)
+          val q = new SynchronousQueue[String]()
           val qq = if st.compareAndSet(i, null, q) then
             q.take()
           else
-            st.get(i).offer("sender")
+            st.get(i).put("sender")
         }
       }
 
       fork {
         for (i <- 0 until max) {
           r.incrementAndGet()
           s.get()
-          val q = new ArrayBlockingQueue[String](1)
+          val q = new SynchronousQueue[String]()
           val qq = if st.compareAndSet(i, null, q) then
             q.take()
           else
-            st.get(i).offer("receiver")
+            st.get(i).put("receiver")
         }
       }.join()
     }

core/src/main/scala/ox/channels2/perf/tests.scala

@@ -3,7 +3,7 @@ package ox.channels2.perf
 @main def run(): Unit =
   for (i <- 1 to 3) {
     println(s"Run $i")
-//    usingOx()
-    passingValues()
+    usingOx()
+//    passingValues()
 //    usingThreads()
   }