Cats-effect: finalizer of `Bracket. guarantee` is not executed upon `fiber.cancel`

[t3hnar]

I found something weird and cannot explain so far.
Looks like only last guarantee is taken into account during cancellation for abstract F[_]

          .guarantee(ref1.set(true))
          .guarantee(ref2.set(true))

However that works fine when called directly against IO
Also looks like @kubukoz can see variant results for both IO and F[_] examples.

Here is reproducible snippet

import cats.effect._
import cats.effect.concurrent.Ref
import cats.effect.implicits._

import scala.concurrent.duration._

object Example extends IOApp {

  def run(args: List[String]): IO[ExitCode] = {
    //    runF[IO]
    runIO
  }


  def runIO: IO[ExitCode] = {
    for {
      ref1 <- Ref.of[IO, Boolean](false)
      ref2 <- Ref.of[IO, Boolean](false)
      fiber <- start[IO](ref1, ref2)
      _ <- fiber.cancel
      _ <- IO.sleep(100.millis)
      r1 <- ref1.get
      r2 <- ref2.get
    } yield {
      println(s"r1: $r1, r2: $r2")
      ExitCode.Success
    }
  }


  def runF[F[_] : Concurrent](implicit timer: Timer[F]) = {
    import cats.implicits._
    for {
      ref1 <- Ref.of[F, Boolean](false)
      ref2 <- Ref.of[F, Boolean](false)
      fiber <- start[F](ref1, ref2)
      _ <- fiber.cancel
      _ <- timer.sleep(100.millis)
      r1 <- ref1.get
      r2 <- ref2.get
    } yield {
      println(s"r1: $r1, r2: $r2")
      ExitCode.Success
    }
  }

  def start[F[_] : Concurrent](ref1: Ref[F, Boolean], ref2: Ref[F, Boolean]) = {
    Concurrent[F].start {
      Async[F].never[Unit]
        .guarantee(ref1.set(true))
        .guarantee(ref2.set(true))
    }
  }
}

[alexandru]

When you’re starting a fiber and then immediately canceling it, you do not have the guarantee that the computation will actually start and thus have no guarantee that a finalizer will be installed.

If you’ll take a look at all of our tests that observe cancellation, we use latches for waiting on the computation to be started, before canceling the fiber.

[t3hnar]

@alexandru the issue I'm trying to show - that there are two finalizers and one is executed but another one - not.
So basically fiber is started.

[alexandru]

In other words your reproducing snippet isn’t correct because its result depends on a race condition, i.e. the fiber has to execute faster than the main thread doing the cancellation. Sometimes this can happen, but most of the times it won’t, due to the fiber execution involving a thread fork, which is expensive.

[t3hnar]

@alexandru I can agree that snippet is not correct from that perspective. Should I fix it ?

Anyhow, was I able to convince you about some strange issue taking place there? :)

[alexandru]

The second finalizer follows the same rules @t3hnar. Due to the auto-cancellation model you can end up with only the first finalizer executing, because the second is never evaluated.

In other words there is nothing on your sample that guarantees _atomic execution_ of the resulting computation.

[t3hnar]

@alexandru does that mean that I should not use fiber.cancel in cases when I need to release expensive resources in finalizers ?

[alexandru]

Note that 2 guarantees one after another is equivalent with:

IO.unit.bracket { _ =>
  IO.unit.bracket { _=>
    IO.never
  } { 
    ref1.set(true)
  }
} {
  ref2.set(true)
}

You should notice that in this case you can end up in a situation in which only ref2 is modified, but not ref1.

[t3hnar]

Exactly. These are results I can see continuously
r1: true, r2: true for runIO
r1: false, r2: true for runF[IO]

Do you happen to know why I'm getting those different for IO and F[_] ?

[alexandru]

On fiber.cancel not sure what the question is, but you should always use it in the context of a bracket release.

Also we are talking about tasks that are never evaluated. So if you have expensive resources, then usage of guarantee is wrong, what you need is a full bracket.

[t3hnar]

what you need is a full bracket.

In my case I want to acquire resource on main thread, to make sure we can acquire it in general.
And then pass it to forked thread, which will release when done.
Hence full bracket does not suit here.

Looking into code I cannot see why guarantee is a bad choice for my case.

[alexandru]

Do you happen to know why I'm getting those different for IO and F[_] ?

I think that’s just non-determinism. Run them enough times and you should see all combinations, including (false, false).

[alexandru]

On guarantee being a bad choice, that’s because you won’t get the guarantee that the finalizer will be installed and executed. As you can see here.

You need to acquire the resource via bracket and then use its release for the finalizer. Note both use and release can fork another thread.

Unfortunately this is the drawback of the auto-cancellation model. Resource acquisition needs bracket.

Monix’s Task can work with non-interruptible flatMaps, so you can have an easier time on the use case you’re thinking of, but we have no type class that exposes this yet, so polymorphic code would be broken.

[t3hnar]

I changed example and can confirm that this is non-determinism for both F[_] and IO

import cats.effect._
import cats.effect.concurrent.{Deferred, Ref}
import cats.effect.implicits._
import cats.implicits._

import scala.concurrent.duration._

object Example extends IOApp {

  def run(args: List[String]): IO[ExitCode] = {
    val fas = List.fill(100)(runF[IO]) ++ List.fill(100)(runIO)
    fas.foldLeft(IO.unit)(_ *> _).as(ExitCode.Success)
  }


  def runIO: IO[Unit] = {
    for {
      deferred <- Deferred[IO, Unit]
      ref1 <- Ref.of[IO, Boolean](false)
      ref2 <- Ref.of[IO, Boolean](false)
      fiber <- start[IO](deferred, ref1, ref2)
      _ <- deferred.get
      _ <- fiber.cancel
      _ <- IO.sleep(100.millis)
      r1 <- ref1.get
      r2 <- ref2.get
    } yield {
      println(s"IO r1: $r1, r2: $r2")
    }
  }


  def runF[F[_] : Concurrent](implicit timer: Timer[F]): F[Unit] = {
    for {
      deferred <- Deferred[F, Unit]
      ref1 <- Ref.of[F, Boolean](false)
      ref2 <- Ref.of[F, Boolean](false)
      fiber <- start[F](deferred, ref1, ref2)
      _ <- deferred.get
      _ <- fiber.cancel
      _ <- timer.sleep(100.millis)
      r1 <- ref1.get
      r2 <- ref2.get
    } yield {
      println(s"F[_] r1: $r1, r2: $r2")
    }
  }

  def start[F[_] : Concurrent](
    deferred: Deferred[F, Unit],
    ref1: Ref[F, Boolean],
    ref2: Ref[F, Boolean]) = {

    Concurrent[F].start {
      for {
        _ <- deferred.complete(())
        _ <- Async[F].never[Unit]
          .guarantee(ref1.set(true))
          .guarantee(ref2.set(true))
      } yield {}
    }
  }
}

[t3hnar]

Here is working example of feature I was looking for
Basically implementation of forking usage of Resource and still making sure that acquire did work before we move on with the main thread.

import cats.effect._
import cats.effect.concurrent.{Deferred, Ref}
import cats.implicits._
import cats.effect.implicits._

object Example extends IOApp {

  def run(args: List[String]): IO[ExitCode] = {
    val fas = List.fill(10000)(runF[IO])
    fas.foldLeft(IO.unit)(_ *> _).as(ExitCode.Success)
  }


  def runF[F[_] : Concurrent]: F[Unit] = {
    for {
      ref <- Ref.of[F, Boolean](false)
      res = Resource.make(().pure[F])(_ => ref.set(true))
      fiber <- startRes(res)(_.pure[F])
      _ <- fiber.cancel
      result <- ref.get
    } yield {
      if (!result) println(s"wrong")
    }
  }

  def startRes[F[_] : Concurrent, A, B](res: Resource[F, A])(use: A => F[B]): F[Fiber[F, B]] = {
    for {
      acquired <- Deferred[F, Option[Throwable]]
      released <- Deferred[F, Unit]
      fiber <- Concurrent[F].start {
        Bracket[F, Throwable].bracket {
          {
            for {
              a <- res.allocated.attempt
              _ <- acquired.complete(a.fold(_.some, _ => none))
            } yield a
          }.rethrow
        } { case (a, _) => use(a)
        } { case (_, release) => release.guarantee(released.complete(())) }
      }
      _ <- acquired.get
    } yield {
      new Fiber[F, B] {

        def cancel = {
          for {
            _ <- fiber.cancel
            _ <- released.get
          } yield {}
        }

        def join = fiber.join
      }
    }
  }
}

@alexandru do you have any concerns about this?

[t3hnar]

or even simpler case, but with using allocate and release on different threads.

import cats.effect.concurrent.Deferred
import cats.effect.implicits._
import cats.effect.{Concurrent, Fiber, Resource}
import cats.implicits._

object StartRes {

  def apply[F[_] : Concurrent, A, B](res: Resource[F, A])(use: A => F[B]): F[Fiber[F, B]] = {
    for {
      allocated    <- res.allocated
      (a, release)  = allocated
      released     <- Deferred[F, Unit]
      fiber        <- Concurrent[F].start {
        use(a).guarantee(release.guarantee(released.complete(())))
      }
    } yield {
      new Fiber[F, B] {

        def cancel = {
          for {
            _ <- fiber.cancel
            _ <- released.get
          } yield {}
        }

        def join = fiber.join
      }
    }
  }
}

[alexandru]

Usage of allocated outside of a bracket or uncancelable context is a resource leak waiting to happen.

[t3hnar]

Thx for help. Closing the issue.