Fiexed order of read-done and read in DuplicatedChannel, to make done always be called first.

Author: rssh

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

@@ -72,4 +72,6 @@ object JVMGopher extends GopherAPI:
       }
 
 
+   final val MAX_SPINS = 400
+
 val Gopher = JVMGopher

jvm/src/main/scala/gopher/impl/GuardedSPSCBaseChannel.scala

@@ -12,6 +12,8 @@ import scala.util.Try
 import scala.util.Success
 import scala.util.Failure
 
+import java.util.logging.{Level => LogLevel}
+
 
 /**
  * Guarded channel work in the next way:
@@ -33,7 +35,6 @@ abstract class GuardedSPSCBaseChannel[F[_]:CpsAsyncMonad,A](override val gopherA
   protected val stepGuard = new AtomicInteger(STEP_FREE)
 
   protected val stepRunnable: Runnable = (()=>entryStep())
-
 
   def addReader(reader: Reader[A]): Unit =
     if (reader.canExpire) then
@@ -66,6 +67,7 @@ abstract class GuardedSPSCBaseChannel[F[_]:CpsAsyncMonad,A](override val gopherA
 
   protected def entryStep(): Unit =
     var done = false
+    var nSpins = 0
     while(!done) {
        if (stepGuard.compareAndSet(STEP_FREE,STEP_BUSY)) {
           done = true
@@ -77,6 +79,7 @@ abstract class GuardedSPSCBaseChannel[F[_]:CpsAsyncMonad,A](override val gopherA
           done = true
        } else {
          // other set updates, we should spinLock
+         nSpins = nSpins + 1
          Thread.onSpinWait()
        }
     }
@@ -99,7 +102,7 @@ abstract class GuardedSPSCBaseChannel[F[_]:CpsAsyncMonad,A](override val gopherA
     while(!readers.isEmpty) {
       val r = readers.poll()
       if (!(r eq null) && !r.isExpired) then
-         r.capture() match
+        r.capture() match
             case Some(f) =>
               progress = true
               taskExecutor.execute(() => f(Failure(new ChannelClosedException())) )
@@ -173,9 +176,12 @@ abstract class GuardedSPSCBaseChannel[F[_]:CpsAsyncMonad,A](override val gopherA
     if (!v.isExpired) 
       if (queue.isEmpty)
         Thread.onSpinWait()
+        // if (nSpins > JVMGopher.MAX_SPINS)
+        //   Thread.`yield`()
       queue.addLast(v)
 
 
+
 object GuardedSPSCBaseChannel:
 
   final val STEP_FREE = 0

jvm/src/main/scala/gopher/impl/GuardedSPSCBufferedChannel.scala

@@ -9,6 +9,7 @@ import scala.util.Try
 import scala.util.Success
 import scala.util.Failure
 
+import java.util.logging.{Level => LogLevel}
 
 class GuardedSPSCBufferedChannel[F[_]:CpsAsyncMonad,A](gopherApi: JVMGopher[F], bufSize: Int,
 controlExecutor: ExecutorService, 

jvm/src/test/scala/gopher/channels/DuppedChannelsMultipleSuite.scala

@@ -0,0 +1,87 @@
+package gopher.channels
+
+import cps._
+import cps.monads.FutureAsyncMonad
+import gopher._
+import munit._
+
+import scala.concurrent._
+import scala.concurrent.duration._
+import scala.language.postfixOps
+import scala.util._
+
+class DuppedChannelsMultipleSuite extends FunSuite  {
+
+  import scala.concurrent.ExecutionContext.Implicits.global
+  given gopherApi: Gopher[Future] = SharedGopherAPI.apply[Future]()
+
+  val inMemoryLog = new java.util.concurrent.ConcurrentLinkedQueue[(Int, Long, String, Throwable)]()
+    
+
+  test("on closing of main stream dupped outputs also closed N times.") {
+    val N = 1000
+    var logIndex = 0
+    gopherApi.setLogFun( (level,msg, ex) => inMemoryLog.add((logIndex, Thread.currentThread().getId(), msg,ex)) )
+    for(i <- 1 to N) {
+      logIndex = i
+      val ch = makeChannel[Int](1)
+      val (in1, in2) = ch.dup()
+      val f1 = async{
+        ch.write(1)
+        ch.close()
+      }
+      val f = for{ fx <- f1
+         x <- in1.aread()
+         r <- in1.aread().transformWith {
+           case Success(u) => 
+               Future failed new IllegalStateException("Mist be closed")
+           case Failure(u) => 
+               Future successful (assert(x == 1))
+         }
+      } yield {
+        r
+      }
+      try {
+        val r = Await.result(f, 30 seconds);
+      }catch{
+        case ex: TimeoutException =>
+          showTraces(20)
+          println("---")
+          showInMemoryLog()
+          throw ex
+      }
+    }
+
+  }
+
+  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")
+      }
+    }
+  } 
+
+  def showInMemoryLog(): Unit = {
+    while(!inMemoryLog.isEmpty) {
+      val r = inMemoryLog.poll()
+      if (r != null) {
+        println(r)
+      }
+    }
+  }
+
+
+}
+
+

shared/src/main/scala/gopher/ReadChannel.scala

@@ -7,6 +7,9 @@ import scala.util.Success
 import scala.util.Failure
 import scala.concurrent.duration.Duration
 
+import java.util.logging.{Level => LogLevel}
+
+
 trait ReadChannel[F[_], A]:
 
    thisReadChannel =>

shared/src/main/scala/gopher/SelectGroup.scala

@@ -10,6 +10,8 @@ import scala.util._
 import scala.concurrent.duration._
 import scala.language.postfixOps
 
+import java.util.logging.{Level => LogLevel}
+
 
 /**
  * Select group is a virtual 'lock' object, where only

shared/src/main/scala/gopher/SelectLoop.scala

@@ -5,6 +5,9 @@ import scala.quoted._
 import scala.compiletime._
 import scala.concurrent.duration._
 
+import java.util.logging.{Level => LogLevel}
+
+
 class SelectLoop[F[_]](api: Gopher[F]) extends SelectGroupBuilder[F,Boolean, Unit](api):
 
 
@@ -25,7 +28,3 @@ class SelectLoop[F[_]](api: Gopher[F]) extends SelectGroupBuilder[F,Boolean, Uni
     }
 
 
-   
-  
-
-

shared/src/main/scala/gopher/impl/DuppedInput.scala

@@ -8,6 +8,7 @@ import scala.util._
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicInteger
 
+import java.util.logging.{Level => LogLevel}
 
 
 
@@ -21,15 +22,16 @@ class DuppedInput[F[_],A](origin:ReadChannel[F,A], bufSize: Int=1)(using api:Gop
 
   given CpsSchedulingMonad[F] = api.asyncMonad
 
-  val runner = SelectLoop[F](api).onReadAsync(origin){a => async{
-    val f1 = sink1.write(a)
-    val f2 = sink2.write(a)
-    true
-  }}.onRead(origin.done){ _ => 
-    sink1.close()
-    sink2.close()
-    false
-  }.runAsync()
-  api.asyncMonad.spawn(runner)
+  val runner = SelectLoop[F](api).
+    onRead(origin.done){ _ => 
+      sink1.close()
+      sink2.close()
+      false
+    }.onReadAsync(origin){a => async{
+      val f1 = sink1.write(a)
+      val f2 = sink2.write(a)
+      true
+    }}.runAsync()
+    api.asyncMonad.spawn(runner)
 
 }