Implement IORuntimeMetrics

[iRevive]

There are a few notable notes regarding this PR:

• The existing implementation of CpuStarvation MBean somewhat breaks the consistency. All previous MBeans were under cats.effect.unsafe.metrics package, while CpuStarvation is available under cats.effect.metrics. I didn't update the MBean name yet.
• Platform-specific metrics can be defined in the IORuntimeMetricsPlatform. For example, JVM IORuntimeMetrics offers compute and cpuStarvation metrics. While JS and Native have only cpuStarvation.

Personally, the whole implementation feels clunky, hopefully we can find the right direction together.

This PR will not pass MiMa checks. Once we agree on a direction, I will address the bin compat issue.

[iRevive]

And a small demo:

Demo

import cats.effect.{IO, IOApp}
import cats.effect.std.Random
import cats.effect.unsafe.IORuntimeMetrics
import cats.syntax.foldable._

import scala.concurrent.duration._

object Test extends IOApp.Simple {

  def run: IO[Unit] = {
    IO.delay(print(runtime.metrics)).flatMap(IO.println).delayBy(500.millis).foreverM.background.surround {
      Random.scalaUtilRandom[IO].flatMap { random =>
        val a = random.betweenInt(10, 3000).flatMap { int =>
          if (int % 2 == 0) IO.blocking(Thread.sleep(int)) else IO.delay(Thread.sleep(int))
        }.start.replicateA(100)

        a.flatMap(_.traverse_(_.join))
      } >> IO.never
    }
  }
  
  def print(metrics: IORuntimeMetrics): String = {
    s"""
      |Compute:
      |workerThreadCount: ${metrics.compute.map(_.workerThreadCount())}
      |activeThreadCount: ${metrics.compute.map(_.activeThreadCount())}
      |searchingThreadCount: ${metrics.compute.map(_.searchingThreadCount())}
      |blockedWorkerThreadCount: ${metrics.compute.map(_.blockedWorkerThreadCount())}
      |localQueueFiberCount: ${metrics.compute.map(_.localQueueFiberCount())}
      |suspendedFiberCount: ${metrics.compute.map(_.suspendedFiberCount())}
      |
      |CPU Starvation:
      |starvationCount: ${metrics.cpuStarvation.starvationCount()}
      |maxClockDriftMs: ${metrics.cpuStarvation.maxClockDriftMs()}
      |currentClockDriftMs: ${metrics.cpuStarvation.currentClockDriftMs()}
      |""".stripMargin
  }
}

Output:

Compute:
workerThreadCount: Some(8)
activeThreadCount: Some(8)
searchingThreadCount: Some(0)
blockedWorkerThreadCount: Some(13)
localQueueFiberCount: Some(0)
suspendedFiberCount: Some(2)

CPU Starvation:
starvationCount: 0
maxClockDriftMs: 0
currentClockDriftMs: 0

....

Compute:
workerThreadCount: Some(8)
activeThreadCount: Some(1)
searchingThreadCount: Some(0)
blockedWorkerThreadCount: Some(13)
localQueueFiberCount: Some(0)
suspendedFiberCount: Some(2)

CPU Starvation:
starvationCount: 0
maxClockDriftMs: 11
currentClockDriftMs: 3

[iRevive]

Mima is sad:

[error] cats-effect: Failed binary compatibility check against org.typelevel:cats-effect_3:3.4.2 (e:info.apiURL=https://typelevel.org/cats-effect/api/3.x/, e:info.versionScheme=early-semver)! Found 7 potential problems (filtered 2)
[error]  * class cats.effect.metrics.CpuStarvation does not have a correspondent in current version
[error]    filter with: ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.CpuStarvation")
[error]  * object cats.effect.metrics.CpuStarvation does not have a correspondent in current version
[error]    filter with: ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.CpuStarvation$")
[error]  * interface cats.effect.metrics.CpuStarvationMBean does not have a correspondent in current version
[error]    filter with: ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.CpuStarvationMBean")
[error]  * interface cats.effect.metrics.CpuStarvationMetrics does not have a correspondent in current version
[error]    filter with: ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.CpuStarvationMetrics")
[error]  * class cats.effect.metrics.JvmCpuStarvationMetrics does not have a correspondent in current version
[error]    filter with: ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.JvmCpuStarvationMetrics")
[error]  * object cats.effect.metrics.JvmCpuStarvationMetrics does not have a correspondent in current version
[error]    filter with: ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.JvmCpuStarvationMetrics$")
[error]  * class cats.effect.metrics.JvmCpuStarvationMetrics#NoOpCpuStarvationMetrics does not have a correspondent in current version
[error]    filter with: ProblemFilters.exclude[MissingClassProblem]("cats.effect.metrics.JvmCpuStarvationMetrics$NoOpCpuStarvationMetrics")

Since these classes were private to metrics package, I assume it's safe to add an exclusion.

[iRevive]

With #3526 the linking issue in ScalaJS is solved.

[djspiewak]

Hey it only took me two years to get to this. New record?

Sorry. :( So the implementation overall looks good. I think this sort of thing is always somewhat clunky so that's fine. We should probably make the mbean paths consistent and just publish the change in the release notes (so people know to update any monitoring keys). The unsafe packages are similarly annoying but I think the optimal solution to this.

So long as all the mbean paths are made consistent, I'm pretty okay merging this. Anyone have any strong objections?

[iRevive]

@djspiewak thanks for the review! Better late than never :)

I changed the namespace of the CPUStarvation MBean to cats.effect.unsafe.metrics.

The names are the following:

• CPU starvation: cats.effect.unsafe.metrics:type=CpuStarvation
• WSTP compute pool: cats.effect.unsafe.metrics:type=ComputePoolSampler-$hash
• WSTP local queue: cats.effect.unsafe.metrics:type=LocalQueueSampler-$hash-$i
• Live fiber dump: cats.effect.unsafe.metrics:type=LiveFiberSnapshotTrigger-$hash

[djspiewak]

Looks good! Once we sort the merge conflicts we can land it.

[iRevive]

Done. Waiting for CI. Nvm, waiting for #4180 first 😅

[armanbilge]

I think we should add metrics for the TimerHeap as well (number of outstanding timers, next one due to fire, maybe last one due to fire?).

I'd also like to add polling system metrics. All polling systems should be able to support # of outstanding I/O operations.

[iRevive]

May we do this in the follow-up PR? The current PR is already big, and I would prefer to operate smaller changesets.

---

Getting stats (at least the number of timers) from the TimerHeap seems relatively simple. However, PollingSystem could require some changes
The naive idea is:

abstract class PollingSystem {
  def stats: PollingSystem.Stats
}
object PollingSystem {
  sealed trait Stats { 
    def outstandingOperations(): Long
  }
}

That way, we can enforce implementations to provide metrics.

I also found this one: #3686.