Refactor WorkerThread runloop; avoid pathological starvation of pollers

[armanbilge]

Fixes #4228.

At a high level, a WorkerThread is always in one of the following three states:

1. Working (primarily on the local queue).
2. Looking for work (external or stolen).
3. Parked.

Previously, (3) was a separate parkLoop(), but (1) and (2) were entangled in the top-level runloop. Now, (2) is refactored into a separate lookForWork() loop. Thus, the top-level runloop is only responsible for tracking "ticks", such that polling runs every 64 ticks.

[armanbilge]

Actually, we could lift case 0 out of the loop. It's always guaranteed to run exactly once, and never again. The loop just transitions between cases 1 and 2.

[armanbilge]

I think the failures are legit 😕 something must be broken.

[armanbilge]

Aha, all the CI failures are on JDK 21, and this is the test it's hanging on.

cats-effect/tests/jvm/src/test/scala/cats/effect/IOPlatformSpecification.scala

Lines 615 to 630 in fc5818d

	if (javaMajorVersion >= 21)
	"block in-place on virtual threads" in real {
	val loomExec = classOf[Executors]
	.getDeclaredMethod("newVirtualThreadPerTaskExecutor")
	.invoke(null)
	.asInstanceOf[ExecutorService]
	val loomEc = ExecutionContext.fromExecutor(loomExec)
	IO.blocking {
	classOf[Thread]
	.getDeclaredMethod("isVirtual")
	.invoke(Thread.currentThread())
	.asInstanceOf[Boolean]
	}.evalOn(loomEc)
	}

Edit: but ... we are getting this weirdness, on the new test I added in this PR.

java.lang.InterruptedException
  | => tat java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1100)
        at java.base/java.util.concurrent.CountDownLatch.await(CountDownLatch.java:230)
        at cats.effect.IOPlatformSpecification.$anonfun$platformSpecs$184(IOPlatformSpecification.scala:601)
        at cats.effect.unsafe.WorkerThread.lookForWork$1(WorkerThread.scala:391)
        at cats.effect.unsafe.WorkerThread.run(WorkerThread.scala:819)

[armanbilge]

The stray InterruptedException is apparently fatal and was nuking all the runtimes, causing remaining tests to hang. We only noticed this on JDK 21 CI runners because the last test in the suite happens to be JDK 21+ only.

[djspiewak]

Haven't reviewed the code yet, but I ran some benchmarks to sanity check. Looks like a very slight regression (note the parTraverse results in particular), but considering it fixes a number of fairness issues, I don't consider that to be an impediment.

Before

[info] Benchmark                                             (cpuTokens)   (size)   Mode  Cnt     Score   Error    Units
[info] ParallelBenchmark.parTraverse                               10000     1000  thrpt   10   755.287 ± 1.087    ops/s
[info] ParallelBenchmark.traverse                                  10000     1000  thrpt   10    68.832 ± 0.112    ops/s
[info] WorkStealingBenchmark.alloc                                   N/A  1000000  thrpt   10    11.866 ± 0.089  ops/min
[info] WorkStealingBenchmark.manyThreadsSchedulingBenchmark          N/A  1000000  thrpt   10    24.042 ± 2.950  ops/min
[info] WorkStealingBenchmark.runnableScheduling                      N/A  1000000  thrpt   10  2541.428 ± 5.020  ops/min
[info] WorkStealingBenchmark.runnableSchedulingScalaGlobal           N/A  1000000  thrpt   10  1874.609 ± 6.710  ops/min
[info] WorkStealingBenchmark.scheduling                              N/A  1000000  thrpt   10    25.641 ± 3.496  ops/min

After

[info] Benchmark                                             (cpuTokens)   (size)   Mode  Cnt     Score   Error    Units
[info] ParallelBenchmark.parTraverse                               10000     1000  thrpt   10   732.683 ± 1.429    ops/s
[info] ParallelBenchmark.traverse                                  10000     1000  thrpt   10    68.975 ± 0.100    ops/s
[info] WorkStealingBenchmark.alloc                                   N/A  1000000  thrpt   10    11.892 ± 0.135  ops/min
[info] WorkStealingBenchmark.manyThreadsSchedulingBenchmark          N/A  1000000  thrpt   10    30.569 ± 3.926  ops/min
[info] WorkStealingBenchmark.runnableScheduling                      N/A  1000000  thrpt   10  2576.645 ± 5.654  ops/min
[info] WorkStealingBenchmark.runnableSchedulingScalaGlobal           N/A  1000000  thrpt   10  1905.347 ± 2.478  ops/min
[info] WorkStealingBenchmark.scheduling                              N/A  1000000  thrpt   10    25.254 ± 4.653  ops/min