We had issues with a service (which happens to use http4s Ember) getting killed by k8s because it became unresponsive. The unresponsiveness includes the liveness check, hence the termination by k8s.
A workaround which seems to have completely eliminate the problem was to
private implicit val network: Network[IO] = Network.forAsync[IO]
when creating the Ember server.
If I understand that correctly, that works around the CE normal way of accepting connections and uses a dedicated thread (pool?) instead.
(edit by djspiewak: the following reproducer was discussed later in the thread and found to not be valid; leaving up for historical context but if you're looking into this bug, this doesn't actually represent the problem)
Here is a reproducer (AI generated 😇 ), which could hopefully lead to a proper fix in the core of the CE runtime.
//> using scala 2.13.14
//> using dep org.typelevel::cats-effect:3.7.0
//> using javaOpt -XX:ActiveProcessorCount=4
/*
* cats-effect 3.7.0 — a ready I/O event registered on the integrated `Selector` is NOT delivered
* for as long as its owning worker is not the one that parks-in-poll, EVEN WHILE other worker
* threads sit idle. Run: `scala-cli run CeLostWakeup.scala` (pins the compute pool to 4 workers).
*
* WHY (all in core/jvm-native/src/main/scala/cats/effect/unsafe, v3.7.0):
*
* 1. The poller is PER-WORKER: `WorkStealingThreadPool.pollers` holds one `java.nio.Selector`
* per worker thread.
*
* 2. `SelectorSystem.select(ch, ops)` registers interest via `ctx.accessPoller { poller => ... }`,
* and `WorkStealingThreadPool.accessPoller` always runs that callback on the CURRENT worker,
* passing THAT worker's own selector (`cb(worker.poller())`). So an interest is always armed on
* whichever worker happens to run the registering fiber, and `select` issues NO
* `selector.wakeup()`.
*
* 3. When a worker runs out of work it decides whether to poll with
* `needsPoll = !ownSelector.keys().isEmpty()` (`WorkerThread.scala`), then parks. In
* `parkLoop` / `parkLoopUntilNextSleeper` it blocks in `system.poll(selector, ...)` ONLY when
* `needsPoll` is true (its OWN selector has registered keys); otherwise it parks via plain
* `LockSupport.park()` / `parkNanos(...)` and polls NO selector at all.
*
* => A worker only ever polls its own selector, and any given FD is registered on exactly one
* worker's selector. If that worker is not the one that next parks-in-poll (e.g. it is busy, or
* an I/O loop's re-arm fiber migrated the interest onto it), the ready FD is serviced by NOBODY:
* idle peer workers park "simple" and never poll a selector they don't own. The event waits
* until the owning worker itself next polls, or until some worker's `select(nanosUntilNext-
* Sleeper)` times out at the next scheduled timer — seconds later, with CPU idle.
*
* This program arms MANY OP_READ interests on separate pipes, each on its OWN fiber, so the
* interests get spread across EVERY worker's selector. It then occupies all-but-one worker with
* non-yielding work and makes all the FDs ready at once from a non-cats-effect thread. Every
* readiness callback should fire in microseconds; instead only the interests that happened to be
* armed on the single still-idle worker's selector are delivered promptly — the rest, armed on the
* busy workers' selectors, are stranded ~the full hold while at least one core sits idle. Because
* the interests are spread over all workers, many are stranded on EVERY run (no reliance on a
* single lucky/unlucky placement). A thread dump taken while the events are pending shows the busy
* workers RUNNABLE — none of them polling the selectors on which the stranded FDs are registered —
* while the lone idle worker polls ONLY its own selector (and so delivers just its share). A worker
* with no keys of its own would instead park "simple" in `LockSupport.park`, polling nothing at
* all. Either way the stranded FDs sit on the busy workers' selectors, serviced by nobody.
* Switching fs2-io/NIO from the integrated `Selector` to a NIO2 `AsynchronousChannelGroup` (whose
* dedicated I/O thread delivers completions) avoids this entirely.
*/
import cats.effect._
import cats.syntax.all._
import java.nio.ByteBuffer
import java.nio.channels.Pipe
import java.nio.channels.SelectionKey.OP_READ
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
object CeLostWakeup extends IOApp.Simple {
private val Workers = Runtime.getRuntime.availableProcessors // = 4 (pinned via javaOpt above)
private val HoldMillis = 2000L // how long all-but-one worker is occupied
private val Interests = Workers * 8 // arm many, so every worker's selector gets some
private val Trials = 6
// A non-cats-effect thread that runs an action on request (so "becomes ready" is genuinely external).
private final class External {
private val q = new java.util.concurrent.SynchronousQueue[Runnable]
private val t = new Thread(() => while (true) q.take().run(), "external-io"); t.setDaemon(true); t.start()
def run(action: => Unit): Unit = q.put(() => action)
}
// One-shot dump: print the compute workers' states while the events are still pending.
private def dumpWorkers(): Unit = {
val sb = new StringBuilder(" [thread dump while the ready events are pending]\n")
Thread.getAllStackTraces.asScala.toSeq.filter(_._1.getName.contains("io-compute")).sortBy(_._1.getName).foreach {
case (th, st) =>
val top = st.take(6).map(_.toString)
val inEpoll = top.exists(s => s.contains("EPoll") || s.contains("SelectorImpl"))
val parkSimp = !inEpoll && top.exists(_.contains("LockSupport.park"))
val tag = if (inEpoll) " <EPOLL: polling its own selector>" else if (parkSimp) " <PARK-SIMPLE: polling NOTHING>" else ""
sb.append(s" ${th.getName} [${th.getState}]$tag\n")
}
System.out.println(sb.toString)
}
def run: IO[Unit] = Selector.get.flatMap { selector =>
val external = new External
val openPipe = IO {
val p = selector.provider.openPipe()
p.source.configureBlocking(false); p.sink.configureBlocking(false); p
}
def oneTrial(i: Int): IO[Unit] =
Resource.make(openPipe)(p => IO { p.sink.close(); p.source.close() }).replicateA(Interests).use { pipes =>
val spin = IO { val end = System.nanoTime() + HoldMillis * 1000000L; while (System.nanoTime() < end) {} }
for {
// (a) Arm OP_READ on each pipe, EACH ON ITS OWN FIBER, so the interests spread across every
// worker's selector. Each fiber records the moment its own event is delivered.
armed <- pipes.traverse(p => (selector.select(p.source, OP_READ) *> IO(System.nanoTime())).start)
_ <- IO.sleep(100.millis) // let (a) finish registering and spreading across the selectors
// (b) Occupy all-but-one worker with non-yielding work, so most workers can neither park-
// in-poll nor steal — but at least one worker stays idle and parks "simple".
spins <- List.fill(Workers - 1)(spin).traverse(_.start)
// (c) Make EVERY FD ready from the external thread, and snapshot a dump ~600ms later while
// the events are (provably) still pending and a core is idle.
firedAt <- IO {
external.run {
pipes.foreach(_.sink.write(ByteBuffer.wrap(Array[Byte](1))))
if (i == 0) { Thread.sleep(600); dumpWorkers() }
}
System.nanoTime()
}
_ <- spins.traverse_(_.joinWithNever)
deliveredAt <- armed.traverse(_.joinWithNever)
_ <- pipes.traverse_(p => IO(p.source.read(ByteBuffer.allocate(8))))
latencies = deliveredAt.map(d => (d - firedAt) / 1000000.0)
stranded = latencies.count(_ > 100)
fast = latencies.size - stranded
// NB: `fast` are the interests that happened to be armed on the one worker left free to
// poll; `stranded` landed on a held worker and waited ~the whole hold, with a core idle.
_ <- IO.println(
f"trial $i: $Interests interests -> fast=$fast stranded=$stranded " +
f"(min=${latencies.min}%.1f ms, max=${latencies.max}%.1f ms)" +
(if (stranded > 0) " <-- STRANDED (no CPU needed to deliver them; a core was idle)" else ""))
} yield ()
}
IO.println(
s"workers=$Workers, arming $Interests interests, holding ${Workers - 1} of them busy for ${HoldMillis}ms per trial\n") *>
(0 until Trials).toList.traverse_(oneTrial) *>
IO.println(
"\nExpected on a healthy runtime: every interest ~<1 ms. Observed: the majority ~the hold time, with a core idle.")
}
}
We had issues with a service (which happens to use http4s Ember) getting killed by k8s because it became unresponsive. The unresponsiveness includes the liveness check, hence the termination by k8s.
A workaround which seems to have completely eliminate the problem was to
when creating the Ember server.
If I understand that correctly, that works around the CE normal way of accepting connections and uses a dedicated thread (pool?) instead.
(edit by djspiewak: the following reproducer was discussed later in the thread and found to not be valid; leaving up for historical context but if you're looking into this bug, this doesn't actually represent the problem)
Here is a reproducer (AI generated 😇 ), which could hopefully lead to a proper fix in the core of the CE runtime.