ConcurrentHashMap with functional programming. Is suspending unsafeRun safe?

Question

Question: Is it safe to suspend unsafeRunSync with IO? E.g.

val io: IO[Unit] = //...
val io2: IO[Unit] = IO(io.unsafeRunSync)

The reason I would do so is that I have some class parameterized with F[_]: Effect which is like a cache:

import cats.effect.Effect

final class MyChache[F[_]](implicit F: Effect[F]) {
  private val cache = new ConcurrentHashMap[Int, String]

  def getOrCreate(key: Int): F[String] = F delay {
    cache.computeIfAbsent(
      key, 
      k => longRunningEffecfulComputation(k).toIO.unsafeRunSync() // <-- Here
    )
  }
}


object MyCache {
  def longRunningEffecfulComputation[F[_] : Effect](key: Int): F[String] = {
    //...
  }
}

The point is I want to run this long running effectfull computation only once for each key (it's pretty infrequent). Yet I would like to stay non-blocking when retrieving existing key.

ConcurrentHashMap seems to be a perfect choice, but it requires this ugly trick with running and suspending the effect. Is there a better way to go?

Answer 1

This is at least potentially unsafe. Suppose that your long running computation used a fixed-size thread pool:

import java.util.concurrent.Executors
import scala.concurrent.ExecutionContext
import cats.effect.Async

object MyCache {
  val smallThreadPool = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(1))

  def longRunningEffectfulComputation[F[_] : Effect](key: Int): F[String] = {
    Effect[F].flatMap(Async.shift[F](smallThreadPool))(_ => Effect[F].delay("test"))
  }
}

And your cache was used on the same thread pool:

val io = for {
  _ <- IO.shift(MyCache.smallThreadPool)
  x <- new MyCache[IO].getOrCreate(1)
} yield x

When you call io.unsafeRunSync(), you will see that it does not terminate.

Instead, you can use a cache api that supports cats-effect, like ScalaCache.