2.1.0-RC2: Something is hanging in the fiber world

[dispalt]

So Ill summarize the code since it's not public, but luckily this is reliably happening.

If I change my dependency to 2.1.0-RC2 from 2.0.22 there appears to be a hang stemming from some change between the two. For context I am building a scala interface on top of a java project. So there is code that treat CompleteableFutures with care, particularly because it's tied to the transaction (database system).

So I have a test where I insert a bunch of elements, and to insert those elements I do it with foreachParDiscard and use the parallelism modifier. However, I noticed in 2.1.0-RC2 if I set parallelism to 32 things hang forever, if I change it back to 16 everything works fine.

I'd be happy to show snippets but can't show the full project yet.

[ghostdogpr]

Does it make a difference if you apply Runtime.enableAutoBlockingExecutor? If your threads are blocking, that behavior could make sense.

[kyri-petrou]

@dispalt just one more thing; what JDK are you testing with? if it's 21, could you also try using the Runtime.enableLoomBasedExecutor?

[dispalt]

I am running with 21, the other thing is this code is calling stuff from java and I guess it would be useful to show you what that looks like. I will also try with 17.

def runAsync[R, A](fn: => ZIO[FdbRecordContext with R, Throwable, A])(implicit trace: Trace): ZIO[R, Throwable, A] =
  ZIO.scoped[R] {
    for {
      runtime <- ZIO.runtime[R]
      id      <- idRef.updateAndGet(_ + 1)
      _       <- recordOpenTransaction(id) *> ZIO.addFinalizer(recordClosedTransaction(id))
      clock   <- ZIO.clock
      ts      <- clock.nanoTime

      functionToRun = { (txn: FDBRecordContext) =>
        val cf = new CompletableFuture[A]()
        runtime.unsafe.fork {
          FdbRecordDatabase.logTxnId(id) {
            ZIO.logDebug(s"Acquiring new transaction context") &>
              fn.provideSomeEnvironment[R](
                _.add[FdbRecordContext](new FdbRecordContext(txn, this))
              ).foldZIO(
                { exception =>
                  cf.completeExceptionally(exception)

                  clock.nanoTime.flatMap { endTs =>
                    ZIO.logWarningCause(
                      s"Transaction failed, took ${Duration(endTs - ts, TimeUnit.NANOSECONDS).toMillis}ms.",
                      Cause.fail(exception)
                    )
                  }
                },
                { value =>
                  cf.complete(value)

                  clock.nanoTime.flatMap { endTs =>
                    ZIO.logDebug(
                      s"Transaction complete, took ${Duration(endTs - ts, TimeUnit.NANOSECONDS).toMillis}ms."
                    )
                  }
                }
              )
          }
        }

        cf.asInstanceOf[CompletableFuture[_ <: A]]

      }.asJava
      result       <- ZIO.fromCompletableFuture[A](db.runAsync[A](functionToRun))
    } yield result
  }

I will try with autoBlocker, I think I tried with loom too but will try again!

[kyri-petrou]

@dispalt I think there are a couple of issues with this implementation but I think the main culprit here is that you need to wrap the fn effect in ZIO.blocking. If you were using JDK21 before, Loom was very likely taking care of this for you in RC1, but unless this is an internal project, you can't / shouldn't expect that the users will be using Loom or have auto-blocking enabled.

A few more things:

• I'm not entirely sure why you need the CompletableFuture in this case if you're going to run it with ZIO.fromCompletableFuture, but in case I'm missing something, you should try and avoid manually constructing it and just use the .toCompletableFuture method on the ZIO effect
• Avoid running simple effects such as ZIO.logDebug in parallel, as that causes additional fibers to be created
• If you MUST use this pattern, then please consider fulfilling the CompletableFuture using something like this:

  val fiber = Runtime.default.unsafe.fork(???)
  val cf    = new CompletableFuture[A]
  fiber.unsafe.addObserver {
    case Exit.Success(value)                    => cf.complete(value)
    case Exit.Failure(c) if c.isInterruptedOnly => cf.cancel(true) // or false
    case Exit.Failure(c)                        => cf.completeExceptionally(c.squash)
  }

[dispalt]

So if I change the code to look closer to yours and wrap the fn with ZIO.blocking it still doesn't work, however if I enableAutoBlocking it does work...

The reason for the wonkiness with CF is that the database transaction completes (or rolls back) is based on the lifetime of the inner completeable db.runAsync call. It's signature is CF<t> runAsync(cf: CF<t>) I thought it would be worse to call
runtime.unsafe.run(fn.toCompleteableFuture) but maybe it's not? I mean it should only be the materialization of the future that blocks, not the future itself, right? which I assume wouldn't be too terrible?

enabling loom did not seem to make a difference.

[dispalt]

Aha, I had one interface with java where I was calling run (before this during the opening of the database) without the Blocking combinator, I added ZIO.blocking and it started working, sorry for the false alarm!