First discovered this when debugging Native multithreading. It's also affecting CE 3.6.0-RC1 on the JVM, so it must be a WSTP / polling system integration bug.
//> using dep org.typelevel::cats-effect::3.6.0-RC1
import cats.effect.*
import java.nio.ByteBuffer
import java.nio.channels.Pipe
import java.nio.channels.SelectionKey._
import scala.concurrent.duration._
object Bug extends IOApp.Simple {
override def computeWorkerThreadCount = 1 // changing to 2 fixes
def run = mkPipe.use { pipe =>
for {
selector <- getSelector
buf <- IO(ByteBuffer.allocate(4))
_ <- IO(pipe.sink.write(ByteBuffer.wrap(Array(1, 2, 3)))).background
.surround {
selector.select(pipe.source, OP_READ) *> IO(pipe.source.read(buf))
}
_ <- IO(pipe.sink.write(ByteBuffer.wrap(Array(42)))).background.surround {
selector.select(pipe.source, OP_READ) *> IO(pipe.source.read(buf))
}
} yield assert(buf.array().toList == List[Byte](1, 2, 3, 42))
}.timeout(1.second)
def getSelector: IO[Selector] =
IO.pollers
.map(_.collectFirst { case selector: Selector => selector })
.map(_.get)
def mkPipe: Resource[IO, Pipe] =
Resource
.eval(getSelector)
.flatMap { selector =>
Resource.make(IO(selector.provider.openPipe())) { pipe =>
IO(pipe.sink().close()).guarantee(IO(pipe.source().close()))
}
}
.evalTap { pipe =>
IO {
pipe.sink().configureBlocking(false)
pipe.source().configureBlocking(false)
}
}
}
First discovered this when debugging Native multithreading. It's also affecting CE 3.6.0-RC1 on the JVM, so it must be a WSTP / polling system integration bug.