refactored Expirable.caputure to return 3-variant value; added support of cps generators

Author: rssh

build.sbt

@@ -10,7 +10,7 @@ val sharedSettings = Seq(
     scalaVersion := dottyVersion,
     name := "scala-gopher",
     resolvers += "Local Ivy Repository" at "file://"+Path.userHome.absolutePath+"/.ivy2/local",
-    libraryDependencies += "com.github.rssh" %%% "dotty-cps-async" % "0.9.2",
+    libraryDependencies += "com.github.rssh" %%% "dotty-cps-async" % "0.9.3-SNAPSHOT",
     libraryDependencies += "org.scalameta" %%% "munit" % "0.7.27" % Test,
 )
 
@@ -34,6 +34,14 @@ lazy val gopher = crossProject(JSPlatform, JVMPlatform)
     .disablePlugins(SitePreviewPlugin)
     .jvmSettings(
         scalacOptions ++= Seq( "-unchecked", "-Ycheck:macros", "-uniqid", "-Xprint:types" ),
+        fork := true,
+        /*
+        javaOptions ++= Seq(
+         "--add-opens", 
+         "java.base/java.lang=ALL-UNNAMED",
+         s"-javaagent:${System.getProperty("user.home")}/.ivy2/local/com.github.rssh/trackedfuture_3/0.5.0/jars/trackedfuture_3-assembly.jar"
+        )
+        */
     ).jsSettings(
         libraryDependencies += ("org.scala-js" %%% "scalajs-java-logging" % "1.0.0").cross(CrossVersion.for3Use2_13),
         // TODO: switch to ModuleES ?

js/src/main/scala/gopher/JSGopher.scala

@@ -1,9 +1,11 @@
 package gopher
 
-import cps._
+import cps.*
 import java.util.Timer
-import java.util.logging._
-import scala.concurrent.duration._
+import java.util.logging.*
+import scala.concurrent.ExecutionContext
+import scala.concurrent.duration.*
+import scala.scalajs.concurrent.*
 
 class JSGopher[F[_]:CpsSchedulingMonad](cfg: JSGopherConfig) extends Gopher[F]:
 
@@ -28,6 +30,8 @@ class JSGopher[F[_]:CpsSchedulingMonad](cfg: JSGopherConfig) extends Gopher[F]:
    def log(level: Level, message: String, ex: Throwable| Null): Unit =
       currentLogFun.apply(level,message,ex)
 
+   def taskExecutionContext: ExecutionContext = JSExecutionContext.queue 
+
    private var currentLogFun: (Level, String, Throwable|Null )=> Unit = { (level,message,ex) =>
       System.err.println(s"${level}:${message}");
       if !(ex eq null) then
@@ -46,5 +50,6 @@ object JSGopher extends GopherAPI:
    val timer = new Timer("gopher")
 
 
+
 val Gopher = JSGopher   
 

js/src/main/scala/gopher/impl/BaseChannel.scala

@@ -42,12 +42,13 @@ abstract class BaseChannel[F[_],A](override val gopherApi: JSGopher[F]) extends
 
   def addWriter(writer: Writer[A]): Unit =
     if (closed) {
-        writer.capture().foreach{ (a,f) =>
-          writer.markUsed()
-          submitTask( () =>
-            f(Failure(new ChannelClosedException()))
-          )
-        }
+        writer.capture() match
+          case Expirable.Capture.Ready((a,f)) =>
+            writer.markUsed()
+            submitTask( () =>
+              f(Failure(new ChannelClosedException()))
+            )
+          case _ =>  
     } else {
         writers.enqueue(writer)
         process()
@@ -56,13 +57,14 @@ abstract class BaseChannel[F[_],A](override val gopherApi: JSGopher[F]) extends
   def addDoneReader(reader: Reader[Unit]): Unit =
     if (closed && isEmpty) {
         reader.capture() match
-          case Some(f) =>
+          case Expirable.Capture.Ready(f) =>
             reader.markUsed()
             submitTask( () => f(Success(())))
-          case None =>
+          case Expirable.Capture.WaitChangeComplete =>
             // mb is blocked and will be evaluated in 
             doneReaders.enqueue(reader)
             process()
+          case Expirable.Capture.Expired  =>
     } else {
         doneReaders.enqueue(reader)
         process()
@@ -79,10 +81,10 @@ abstract class BaseChannel[F[_],A](override val gopherApi: JSGopher[F]) extends
       val v = queue.dequeue()
       if (!v.isExpired) then
         v.capture() match
-          case Some(a) =>
+          case Expirable.Capture.Ready(a) =>
             v.markUsed()
             action(a)
-          case None =>
+          case _ =>
             // do nothing.
             //   exists case, when this is possible: wheb we close channel from
             // select-group callback, which is evaluated now.

js/src/main/scala/gopher/impl/BufferedChannel.scala

@@ -71,15 +71,15 @@ class BufferedChannel[F[_]:CpsAsyncMonad, A](gopherApi: JSGopher[F], bufSize: In
     if (!writers.isEmpty && !isFull) then
       val writer = writers.dequeue()
       writer.capture() match
-        case Some((a,f)) =>
+        case Expirable.Capture.Ready((a,f)) =>
           internalEnqueue(a)
           writer.markUsed()
           submitTask(  () => f(Success(())) )
           progress = true
-        case None =>
-          if (!writer.isExpired) then
-            // impossible, we have no parallel execution
-            throw DeadlockDetected()
+        case Expirable.Capture.WaitChangeComplete =>
+          // impossible, we have no parallel execution
+          throw DeadlockDetected()
+        case Expirable.Capture.Expired =>
     progress
 
 

js/src/main/scala/gopher/impl/PromiseChannel.scala

@@ -21,31 +21,30 @@ class PromiseChannel[F[_]:CpsAsyncMonad, A](gopherApi: JSGopher[F]) extends Base
     // we have only one writer.
     while (!writers.isEmpty && value.isEmpty) {
       val w = writers.dequeue()
-      if (!w.isExpired) then
-        w.capture() match
-          case Some((a,f)) =>
+      w.capture() match
+          case Expirable.Capture.Ready((a,f)) =>
             w.markUsed()
             submitTask(()=>f(Success(())))
             value = Some(a)
             closed = true
             // we can't havw more than one unexpired 
-          case None =>
-            if (!w.isExpired) then
+          case Expirable.Capture.WaitChangeComplete =>
               // impossible in js, 
+              //  (mb processNextTick()?)
               throw new DeadlockDetected()
+          case Expirable.Capture.Expired =>
     }
     if (!readers.isEmpty && value.isDefined) {
       while(!readers.isEmpty && !readed) {
         val r = readers.dequeue()
-        if (!r.isExpired) then
-          r.capture() match
-            case Some(f) =>
+        r.capture() match
+            case Expirable.Capture.Ready(f) =>
                 r.markUsed()
                 submitTask(()=>f(Success(value.get)))
                 readed = true
-            case None =>
-              if (!r.isExpired) 
+            case Expirable.Capture.WaitChangeComplete =>
                 throw new DeadlockDetected()  
+            case Expirable.Capture.Expired =>
       }
       //if (readed) {
        // processCloseDone()

js/src/main/scala/gopher/impl/UnbufferedChannel.scala

@@ -25,21 +25,25 @@ class UnbufferedChannel[F[_]:CpsAsyncMonad, A](gopherApi: JSGopher[F]) extends B
              findWriter() match 
                 case Some(writer) =>
                   reader.capture() match 
-                    case Some(readFun) =>
+                    case Expirable.Capture.Ready(readFun) =>
                       writer.capture() match 
-                        case Some((a,writeFun)) =>
+                        case Expirable.Capture.Ready((a,writeFun)) =>
                           submitTask( () => readFun(Success(a)))
                           submitTask( () => writeFun(Success(())) )
                           progress = true
                           done = true
                           writer.markUsed()
                           reader.markUsed()
-                        case None =>
+                        case _ =>
                           // impossible, because in js we have-no interleavinf, bug anyway
                           // let's fallback
                           reader.markFree()
                           readers.prepend(reader)
-                    case None =>
+                    case Expirable.Capture.WaitChangeComplete =>
+                      // impossible, but let's fallback
+                      // TODO: prepend reader and skip event
+                      writers.prepend(writer)
+                    case Expirable.Capture.Expired =>
                       // impossible, but let's fallback
                       writers.prepend(writer)
                 case None =>

js/src/test/scala/gopher/util/Debug.scala

@@ -0,0 +1,41 @@
+package gopher.util
+
+import java.util.logging.{Level => LogLevel}
+
+object Debug {
+
+  type InMemoryLog = java.util.concurrent.ConcurrentLinkedQueue[(Long, String, Throwable)]
+
+  def inMemoryLogFun(inMemoryLog: InMemoryLog): (LogLevel, String, Throwable|Null) => Unit =
+    (level,msg, ex) => inMemoryLog.add((Thread.currentThread().getId(), msg,ex)) 
+
+  def showInMemoryLog(inMemoryLog: InMemoryLog): Unit = {
+    while(!inMemoryLog.isEmpty) {
+      val r = inMemoryLog.poll()
+      if (r != null) {
+            println(r)
+      }
+    }
+  }
+  
+
+  def showTraces(maxTracesToShow: Int): Unit = {
+    val traces = Thread.getAllStackTraces();
+    val it = traces.entrySet().iterator()
+    while(it.hasNext()) {
+      val e = it.next();
+      println(e.getKey());
+      val elements = e.getValue()
+      var sti = 0
+      var wasPark = false
+      while(sti < elements.length && sti < maxTracesToShow && !wasPark) {
+          val st = elements(sti)
+          println(" "*10 + st)
+          sti = sti + 1;
+          wasPark = (st.getMethodName == "park")
+      }
+    }
+  } 
+  
+ 
+}

jvm/src/main/scala/gopher/JVMGopher.scala

@@ -9,6 +9,7 @@ import java.util.concurrent.ForkJoinPool
 import java.util.concurrent.atomic.AtomicReference
 import java.util.Timer
 import java.util.logging._
+import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
 
@@ -18,7 +19,7 @@ class JVMGopher[F[_]:CpsSchedulingMonad](cfg: JVMGopherConfig) extends Gopher[F]
 
    def makeChannel[A](bufSize:Int = 0, autoClose: Boolean = false) =
       if autoClose then
-         PromiseChannel[F,A](this, taskExecutor)
+         PromiseChannel[F,A](this, cfg.taskExecutor)
       else
          if (bufSize == 0)
             GuardedSPSCUnbufferedChannel[F,A](this, cfg.controlExecutor,cfg.taskExecutor)
@@ -35,7 +36,7 @@ class JVMGopher[F[_]:CpsSchedulingMonad](cfg: JVMGopherConfig) extends Gopher[F]
    def log(level: Level, message: String, ex: Throwable| Null): Unit =
       currentLogFun.get().apply(level,message,ex)
 
-   def taskExecutor = cfg.taskExecutor
+   lazy val taskExecutionContext = ExecutionContext.fromExecutor(cfg.taskExecutor)
 
    def scheduledExecutor = JVMGopher.scheduledExecutor