diff --git a/core/js/src/main/scala/cats/effect/unsafe/BatchingMacrotaskExecutor.scala b/core/js/src/main/scala/cats/effect/unsafe/BatchingMacrotaskExecutor.scala index ea4a4e7e25..8e7bd657cb 100644 --- a/core/js/src/main/scala/cats/effect/unsafe/BatchingMacrotaskExecutor.scala +++ b/core/js/src/main/scala/cats/effect/unsafe/BatchingMacrotaskExecutor.scala @@ -24,7 +24,6 @@ import org.scalajs.macrotaskexecutor.MacrotaskExecutor import scala.collection.mutable import scala.concurrent.ExecutionContextExecutor import scala.scalajs.{js, LinkingInfo} -import scala.util.control.NonFatal /** * An `ExecutionContext` that improves throughput by providing a method to `schedule` fibers to @@ -65,7 +64,7 @@ private[effect] final class BatchingMacrotaskExecutor( val fiber = fibers.take() try fiber.run() catch { - case t if NonFatal(t) => reportFailure(t) + case t if UnsafeNonFatal(t) => reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } i += 1 diff --git a/core/jvm/src/main/scala/cats/effect/IOApp.scala b/core/jvm/src/main/scala/cats/effect/IOApp.scala index 36d84fd4c0..1c7d63fd33 100644 --- a/core/jvm/src/main/scala/cats/effect/IOApp.scala +++ b/core/jvm/src/main/scala/cats/effect/IOApp.scala @@ -19,11 +19,11 @@ package cats.effect import cats.effect.metrics.{CpuStarvationWarningMetrics, JvmCpuStarvationMetrics} import cats.effect.std.Console import cats.effect.tracing.TracingConstants._ +import cats.effect.unsafe.UnsafeNonFatal import cats.syntax.all._ import scala.concurrent.{blocking, CancellationException, ExecutionContext} import scala.concurrent.duration._ -import scala.util.control.NonFatal import java.util.concurrent.{ArrayBlockingQueue, CountDownLatch} import java.util.concurrent.atomic.AtomicInteger @@ -218,7 +218,7 @@ trait IOApp { new ExecutionContext { def reportFailure(t: Throwable): Unit = t match { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => IOApp.this.reportFailure(t).unsafeRunAndForgetWithoutCallback()(runtime) case t => @@ -555,7 +555,7 @@ trait IOApp { throw e case t: Throwable => - if (NonFatal(t)) { + if (UnsafeNonFatal(t)) { if (isForked) { t.printStackTrace() System.exit(1) @@ -571,7 +571,7 @@ trait IOApp { try { r.run() } catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => IOApp.this.reportFailure(t).unsafeRunAndForgetWithoutCallback()(runtime) case t: Throwable => diff --git a/core/jvm/src/main/scala/cats/effect/IOFiberPlatform.scala b/core/jvm/src/main/scala/cats/effect/IOFiberPlatform.scala index fb7b31587d..f4e1e80ef9 100644 --- a/core/jvm/src/main/scala/cats/effect/IOFiberPlatform.scala +++ b/core/jvm/src/main/scala/cats/effect/IOFiberPlatform.scala @@ -16,7 +16,7 @@ package cats.effect -import scala.util.control.NonFatal +import cats.effect.unsafe.UnsafeNonFatal import java.nio.channels.ClosedByInterruptException import java.util.{concurrent => juc} @@ -68,7 +68,7 @@ private[effect] abstract class IOFiberPlatform[A] extends AtomicBoolean(false) { case ex: ClosedByInterruptException => throw ex // this won't suppress InterruptedException: - case t if NonFatal(t) => Left(t) + case t if UnsafeNonFatal(t) => Left(t) } // this is why it has to be a semaphore rather than an atomic boolean diff --git a/core/jvm/src/main/scala/cats/effect/unsafe/SelectorSystem.scala b/core/jvm/src/main/scala/cats/effect/unsafe/SelectorSystem.scala index d921759670..a4a477b771 100644 --- a/core/jvm/src/main/scala/cats/effect/unsafe/SelectorSystem.scala +++ b/core/jvm/src/main/scala/cats/effect/unsafe/SelectorSystem.scala @@ -19,8 +19,6 @@ package unsafe import cats.effect.unsafe.metrics.PollerMetrics -import scala.util.control.NonFatal - import java.nio.channels.{SelectableChannel, SelectionKey} import java.nio.channels.spi.{AbstractSelector, SelectorProvider} import java.util.Iterator @@ -72,7 +70,7 @@ final class SelectorSystem private (provider: SelectorProvider) extends PollingS // reset interest in triggered ops key.interestOps(key.interestOps() & ~readyOps) } catch { - case ex if NonFatal(ex) => + case ex if UnsafeNonFatal(ex) => error = ex readyOps = -1 // interest all waiters } @@ -150,7 +148,7 @@ final class SelectorSystem private (provider: SelectorProvider) extends PollingS cb(Right(Some(cancel))) } catch { - case ex if NonFatal(ex) => + case ex if UnsafeNonFatal(ex) => poller.countErroredOperation(ops) cb(Left(ex)) } diff --git a/core/jvm/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala b/core/jvm/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala index 7585e60113..b5d42d617a 100644 --- a/core/jvm/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala +++ b/core/jvm/src/main/scala/cats/effect/unsafe/WorkStealingThreadPool.scala @@ -36,7 +36,6 @@ import cats.effect.tracing.TracingConstants import scala.collection.mutable import scala.concurrent.ExecutionContextExecutor import scala.concurrent.duration.{Duration, FiniteDuration} -import scala.util.control.NonFatal import java.time.Instant import java.time.temporal.ChronoField @@ -680,7 +679,7 @@ private[effect] final class WorkStealingThreadPool[P <: AnyRef]( try { task.run() } catch { - case ex if NonFatal(ex) => + case ex if UnsafeNonFatal(ex) => reportFailure(ex) } } diff --git a/core/jvm/src/main/scala/cats/effect/unsafe/WorkerThread.scala b/core/jvm/src/main/scala/cats/effect/unsafe/WorkerThread.scala index 7cc0d91a7f..e45c76b589 100644 --- a/core/jvm/src/main/scala/cats/effect/unsafe/WorkerThread.scala +++ b/core/jvm/src/main/scala/cats/effect/unsafe/WorkerThread.scala @@ -24,7 +24,6 @@ import scala.annotation.tailrec import scala.collection.mutable import scala.concurrent.{BlockContext, CanAwait} import scala.concurrent.duration.{Duration, FiniteDuration} -import scala.util.control.NonFatal import java.lang.Long.MIN_VALUE import java.util.concurrent.{ArrayBlockingQueue, ThreadLocalRandom} @@ -376,7 +375,7 @@ private[effect] final class WorkerThread[P <: AnyRef]( pool.notifyParked(rnd) try fiber.run() catch { - case t if NonFatal(t) => pool.reportFailure(t) + case t if UnsafeNonFatal(t) => pool.reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } @@ -393,7 +392,7 @@ private[effect] final class WorkerThread[P <: AnyRef]( // The dequeued element is a single fiber. Execute it immediately. try fiber.run() catch { - case t if NonFatal(t) => pool.reportFailure(t) + case t if UnsafeNonFatal(t) => pool.reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } @@ -443,7 +442,7 @@ private[effect] final class WorkerThread[P <: AnyRef]( // Run the stolen fiber. try fiber.run() catch { - case t if NonFatal(t) => pool.reportFailure(t) + case t if UnsafeNonFatal(t) => pool.reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } } @@ -495,7 +494,7 @@ private[effect] final class WorkerThread[P <: AnyRef]( pool.notifyParked(rnd) try fiber.run() catch { - case t if NonFatal(t) => pool.reportFailure(t) + case t if UnsafeNonFatal(t) => pool.reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } @@ -515,7 +514,7 @@ private[effect] final class WorkerThread[P <: AnyRef]( // The dequeued element is a single fiber. Execute it immediately. try fiber.run() catch { - case t if NonFatal(t) => pool.reportFailure(t) + case t if UnsafeNonFatal(t) => pool.reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } @@ -545,7 +544,7 @@ private[effect] final class WorkerThread[P <: AnyRef]( // Run the stolen fiber. try fiber.run() catch { - case t if NonFatal(t) => pool.reportFailure(t) + case t if UnsafeNonFatal(t) => pool.reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } } @@ -813,7 +812,7 @@ private[effect] final class WorkerThread[P <: AnyRef]( try fiber.run() catch { - case t if NonFatal(t) => pool.reportFailure(t) + case t if UnsafeNonFatal(t) => pool.reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } } else if (element.isInstanceOf[Runnable]) { @@ -827,7 +826,7 @@ private[effect] final class WorkerThread[P <: AnyRef]( // The dequeued element is a single fiber. Execute it immediately. try fiber.run() catch { - case t if NonFatal(t) => pool.reportFailure(t) + case t if UnsafeNonFatal(t) => pool.reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } } @@ -862,7 +861,7 @@ private[effect] final class WorkerThread[P <: AnyRef]( // Run the fiber. try fiber.run() catch { - case t if NonFatal(t) => pool.reportFailure(t) + case t if UnsafeNonFatal(t) => pool.reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } } else { diff --git a/core/native/src/main/scala/cats/effect/unsafe/EventLoopExecutorScheduler.scala b/core/native/src/main/scala/cats/effect/unsafe/EventLoopExecutorScheduler.scala index 0216b3eac1..ac431483fd 100644 --- a/core/native/src/main/scala/cats/effect/unsafe/EventLoopExecutorScheduler.scala +++ b/core/native/src/main/scala/cats/effect/unsafe/EventLoopExecutorScheduler.scala @@ -26,7 +26,6 @@ import scala.scalanative.meta.LinktimeInfo import scala.scalanative.posix.time._ import scala.scalanative.posix.timeOps._ import scala.scalanative.unsafe._ -import scala.util.control.NonFatal import java.util.{ArrayDeque, PriorityQueue} @@ -96,7 +95,7 @@ private[effect] final class EventLoopExecutorScheduler[P]( val task = sleepQueue.poll() try task.runnable.run() catch { - case t if NonFatal(t) => reportFailure(t) + case t if UnsafeNonFatal(t) => reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } } @@ -107,7 +106,7 @@ private[effect] final class EventLoopExecutorScheduler[P]( val runnable = executeQueue.poll() try runnable.run() catch { - case t if NonFatal(t) => reportFailure(t) + case t if UnsafeNonFatal(t) => reportFailure(t) case t: Throwable => IOFiber.onFatalFailure(t) } i += 1 diff --git a/core/shared/src/main/scala/cats/effect/IO.scala b/core/shared/src/main/scala/cats/effect/IO.scala index 7840f2656a..c88f487d82 100644 --- a/core/shared/src/main/scala/cats/effect/IO.scala +++ b/core/shared/src/main/scala/cats/effect/IO.scala @@ -50,7 +50,7 @@ import cats.effect.std.{ UUIDGen } import cats.effect.tracing.{Tracing, TracingEvent} -import cats.effect.unsafe.IORuntime +import cats.effect.unsafe.{IORuntime, UnsafeNonFatal} import cats.syntax._ import cats.syntax.all._ @@ -58,7 +58,6 @@ import scala.annotation.unchecked.uncheckedVariance import scala.concurrent._ import scala.concurrent.duration._ import scala.util.{Failure, Success, Try} -import scala.util.control.NonFatal import java.util.UUID import java.util.concurrent.Executor @@ -1014,7 +1013,7 @@ sealed abstract class IO[+A] private () extends IOPlatform[A] { unsafeRunFiber( cb(Left(new CancellationException("The fiber was canceled"))), t => { - if (!NonFatal(t)) { + if (!UnsafeNonFatal(t)) { t.printStackTrace() } cb(Left(t)) @@ -1027,7 +1026,7 @@ sealed abstract class IO[+A] private () extends IOPlatform[A] { unsafeRunFiber( cb(Outcome.canceled), t => { - if (!NonFatal(t)) { + if (!UnsafeNonFatal(t)) { t.printStackTrace() } cb(Outcome.errored(t)) @@ -1050,7 +1049,7 @@ sealed abstract class IO[+A] private () extends IOPlatform[A] { val _ = unsafeRunFiber( (), t => { - if (NonFatal(t)) { + if (UnsafeNonFatal(t)) { if (runtime.config.reportUnhandledFiberErrors) runtime.compute.reportFailure(t) } else { t.printStackTrace() } diff --git a/core/shared/src/main/scala/cats/effect/IOFiber.scala b/core/shared/src/main/scala/cats/effect/IOFiber.scala index 5ad582aed5..ea9b46b37b 100644 --- a/core/shared/src/main/scala/cats/effect/IOFiber.scala +++ b/core/shared/src/main/scala/cats/effect/IOFiber.scala @@ -23,7 +23,6 @@ import cats.effect.unsafe._ import scala.annotation.{switch, tailrec} import scala.concurrent.ExecutionContext import scala.concurrent.duration._ -import scala.util.control.NonFatal import java.util.concurrent.RejectedExecutionException import java.util.concurrent.atomic.AtomicBoolean @@ -262,7 +261,7 @@ private final class IOFiber[A]( case 1 => val cur = cur0.asInstanceOf[Error] val ex = cur.t - if (!NonFatal(ex)) + if (!UnsafeNonFatal(ex)) onFatalFailure(ex) runLoop(failed(ex, 0), nextCancelation, nextAutoCede) @@ -278,7 +277,7 @@ private final class IOFiber[A]( val r = try cur.thunk() catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => error = t case t: Throwable => onFatalFailure(t) @@ -323,7 +322,7 @@ private final class IOFiber[A]( val result = try f(v) catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => error = t case t: Throwable => onFatalFailure(t) @@ -340,7 +339,7 @@ private final class IOFiber[A]( case 1 => val error = ioe.asInstanceOf[Error] val ex = error.t - if (!NonFatal(ex)) + if (!UnsafeNonFatal(ex)) onFatalFailure(ex) runLoop(failed(ex, 0), nextCancelation - 1, nextAutoCede) @@ -357,7 +356,7 @@ private final class IOFiber[A]( val result = try f(delay.thunk()) catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => error = t case t: Throwable => onFatalFailure(t) @@ -397,7 +396,7 @@ private final class IOFiber[A]( def next(v: Any): IO[Any] = try f(v) catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => failed(t, 0) case t: Throwable => onFatalFailure(t) @@ -411,7 +410,7 @@ private final class IOFiber[A]( case 1 => val error = ioe.asInstanceOf[Error] val ex = error.t - if (!NonFatal(ex)) + if (!UnsafeNonFatal(ex)) onFatalFailure(ex) runLoop(failed(ex, 0), nextCancelation - 1, nextAutoCede) @@ -427,7 +426,7 @@ private final class IOFiber[A]( val result = try f(delay.thunk()) catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => failed(t, 0) case t: Throwable => onFatalFailure(t) @@ -466,7 +465,7 @@ private final class IOFiber[A]( case 1 => val error = ioa.asInstanceOf[Error] val t = error.t - if (!NonFatal(t)) + if (!UnsafeNonFatal(t)) onFatalFailure(t) // We need to augment the exception here because it doesn't get // forwarded to the `failed` path. @@ -485,7 +484,7 @@ private final class IOFiber[A]( val result = try delay.thunk() catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => // We need to augment the exception here because it doesn't // get forwarded to the `failed` path. Tracing.augmentThrowable(runtime.enhancedExceptions, t, tracingEvents) @@ -568,7 +567,7 @@ private final class IOFiber[A]( val next = try cur.body(poll) catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => IO.raiseError(t) case t: Throwable => onFatalFailure(t) @@ -759,7 +758,7 @@ private final class IOFiber[A]( try { body[IO].apply(cb, get, FunctionK.id) } catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => IO.raiseError(t) case t: Throwable => onFatalFailure(t) @@ -1021,7 +1020,7 @@ private final class IOFiber[A]( try { cur.thunk() } catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => error = t case t: Throwable => onFatalFailure(t) @@ -1039,7 +1038,7 @@ private final class IOFiber[A]( try { cur.thunk() } catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => error = t case t: Throwable => onFatalFailure(t) @@ -1216,7 +1215,7 @@ private final class IOFiber[A]( val transformed = try f(result) catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => error = t case t: Throwable => onFatalFailure(t) @@ -1235,7 +1234,7 @@ private final class IOFiber[A]( try f(result) catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => failed(t, depth + 1) case t: Throwable => onFatalFailure(t) @@ -1307,7 +1306,7 @@ private final class IOFiber[A]( try f(error) catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => failed(t, depth + 1) case t: Throwable => onFatalFailure(t) @@ -1427,7 +1426,7 @@ private final class IOFiber[A]( val r = try cur.thunk() catch { - case t if NonFatal(t) => + case t if UnsafeNonFatal(t) => error = t case t: Throwable => onFatalFailure(t) diff --git a/core/shared/src/main/scala/cats/effect/SyncIO.scala b/core/shared/src/main/scala/cats/effect/SyncIO.scala index 2a5f3cd22c..d2c3a3ce33 100644 --- a/core/shared/src/main/scala/cats/effect/SyncIO.scala +++ b/core/shared/src/main/scala/cats/effect/SyncIO.scala @@ -19,6 +19,7 @@ package cats.effect import cats.{Align, Eval, Functor, Now, Show, StackSafeMonad} import cats.data.Ior import cats.effect.syntax.monadCancel._ +import cats.effect.unsafe.UnsafeNonFatal import cats.kernel.{Monoid, Semigroup} import cats.syntax.all._ @@ -26,7 +27,6 @@ import scala.annotation.{switch, tailrec} import scala.annotation.unchecked.uncheckedVariance import scala.concurrent.duration._ import scala.util.Try -import scala.util.control.NonFatal import Platform.static @@ -244,7 +244,7 @@ sealed abstract class SyncIO[+A] private () extends Serializable { val r = try cur.thunk() catch { - case t if NonFatal(t) => error = t + case t if UnsafeNonFatal(t) => error = t } val next = @@ -350,7 +350,7 @@ sealed abstract class SyncIO[+A] private () extends Serializable { val transformed = try f(result) catch { - case t if NonFatal(t) => error = t + case t if UnsafeNonFatal(t) => error = t } if (depth > MaxStackDepth) { @@ -367,7 +367,7 @@ sealed abstract class SyncIO[+A] private () extends Serializable { try f(result) catch { - case t if NonFatal(t) => failed(t, depth + 1) + case t if UnsafeNonFatal(t) => failed(t, depth + 1) } } @@ -376,7 +376,7 @@ sealed abstract class SyncIO[+A] private () extends Serializable { try f(t) catch { - case t if NonFatal(t) => failed(t, depth + 1) + case t if UnsafeNonFatal(t) => failed(t, depth + 1) } } diff --git a/core/shared/src/main/scala/cats/effect/unsafe/UnsafeNonFatal.scala b/core/shared/src/main/scala/cats/effect/unsafe/UnsafeNonFatal.scala new file mode 100644 index 0000000000..be0339e766 --- /dev/null +++ b/core/shared/src/main/scala/cats/effect/unsafe/UnsafeNonFatal.scala @@ -0,0 +1,44 @@ +/* + * Copyright 2020-2025 Typelevel + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package cats.effect +package unsafe + +import scala.util.control.ControlThrowable + +import Platform.static + +/** + * An alternative to [[scala.util.control.NonFatal]] that does not treat + * [[java.lang.InterruptedException]] as fatal. This is intended for the exclusive use of the + * Cats-Effect runtime. It is not recommended to treat interrupts as non-fatal in application + * code, as handling interrupts gracefully is the responsibility of the runtime, so + * [[UnsafeNonFatal]] should only be used in the fiber runtime. + */ +private[effect] object UnsafeNonFatal { + + /** + * Returns true if the provided `Throwable` is to be considered non-fatal, or false if it is + * to be considered fatal + */ + @static def apply(t: Throwable): Boolean = t match { + case _: VirtualMachineError | _: ThreadDeath | _: LinkageError | _: ControlThrowable => + false + case _ => true + } +} + +private[effect] class UnsafeNonFatal