[core] introduce reactive Signal (#1082)

This new effect is in honor of the 600th person to start the repository.
Congrats and thank you @diegobernardes!! 🎉

### Problem

Kyo doesn't offer a solution to handle reactive signals, which is a
common need in UI solutions for example.

### Solution

Introduce `Signal` to `kyo-core` with mutable and constant versions and
methods to stream the state.

### Notes

- This is inspired by fs2's `Signal`/`SignallingRef` but the
implementation is more optimized with two separate `AtomicRef`s to
manage the state and listeners, which reduces allocations.
- Please see the scaladocs for an explanation of the functionality and
implementation.
- I had to add a method to allow masking unsafe promises.
- I've also changed the atomic classes to provide `updateAndGet` +
`getAndUpdate` for consistency.

---------

Signed-off-by: Akhilender Bongirwar <[email protected]>
Co-authored-by: Adam Hearn <[email protected]>
Co-authored-by: Matteo Bilardi <[email protected]>
Co-authored-by: Akhilender Bongirwar <[email protected]>

Author: 4 people

kyo-core/js-native/src/main/scala/kyo/internal/OSSignalPlatformSpecific.scala

@@ -0,0 +1,4 @@
+package kyo.internal
+
+private[internal] class OsSignalPlatformSpecific:
+    val handle: OsSignal.Handler = OsSignal.Handler.Noop

kyo-core/js-native/src/main/scala/kyo/internal/SignalPlatformSpecific.scala

@@ -1,6 +1,6 @@
 package kyo.internal
 
-import Signal.Handler
+import OsSignal.Handler
 import java.lang.invoke.MethodHandle
 import java.lang.invoke.MethodHandles
 import java.lang.invoke.MethodType
@@ -13,7 +13,7 @@ import kyo.Result
 /** This class provides a jvm-specific implementation of signal handling. It uses reflection to install signal handlers as they may not be
   * available on all implementations of the JVM.
   */
-private[internal] class SignalPlatformSpecific:
+private[internal] class OsSignalPlatformSpecific:
     private val logger = Logger.getLogger("kyo.internal.Signal")
     val handle: Handler = {
         for
@@ -47,7 +47,7 @@ private[internal] class SignalPlatformSpecific:
                         this
 
             val proxy = Proxy.newProxyInstance(
-                Signal.getClass.getClassLoader,
+                OsSignal.getClass.getClassLoader,
                 Array(signalHandlerClass),
                 invocationHandler
             )
@@ -89,4 +89,4 @@ private[internal] class SignalPlatformSpecific:
             )
         }
     end initHandleStaticMethodHandle
-end SignalPlatformSpecific
+end OsSignalPlatformSpecific

kyo-core/jvm/src/main/scala/kyo/internal/SignalPlatformSpecific.scala renamed to kyo-core/jvm/src/main/scala/kyo/internal/OSSignalPlatformSpecific.scala

@@ -4,12 +4,12 @@ import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 import kyo.Test
 
-class SignalTest extends Test:
+class OsSignalTest extends Test:
 
     "handles on signal" in {
         val wasHandled = new CountDownLatch(1)
 
-        Signal.handle("USR2", wasHandled.countDown())
+        OsSignal.handle("USR2", wasHandled.countDown())
 
         val signal = new sun.misc.Signal("USR2")
         sun.misc.Signal.raise(signal)
@@ -19,7 +19,7 @@ class SignalTest extends Test:
 
     "lazy" in {
         var wasHandled = false
-        Signal.handle("USR2", { wasHandled = true })
+        OsSignal.handle("USR2", { wasHandled = true })
         assert(!wasHandled)
     }
-end SignalTest
+end OsSignalTest

kyo-core/jvm/src/test/scala/kyo/internal/SignalTest.scala renamed to kyo-core/jvm/src/test/scala/kyo/internal/OSSignalTest.scala

@@ -91,6 +91,22 @@ final case class AtomicInt private (unsafe: AtomicInt.Unsafe):
       */
     inline def addAndGet(v: Int)(using inline frame: Frame): Int < IO = IO.Unsafe(unsafe.addAndGet(v))
 
+    /** Atomically updates the current value using the given function and returns the old value.
+      * @param f
+      *   The function to apply to the current value
+      * @return
+      *   The previous value
+      */
+    inline def getAndUpdate(inline f: Int => Int)(using inline frame: Frame): Int < IO = IO.Unsafe(unsafe.getAndUpdate(f))
+
+    /** Atomically updates the current value using the given function and returns the updated value.
+      * @param f
+      *   The function to apply to the current value
+      * @return
+      *   The updated value
+      */
+    inline def updateAndGet(inline f: Int => Int)(using inline frame: Frame): Int < IO = IO.Unsafe(unsafe.updateAndGet(f))
+
     /** Returns a string representation of the current value.
       * @return
       *   A string representation of the atomic integer
@@ -158,6 +174,8 @@ object AtomicInt:
             inline def getAndDecrement()(using inline allow: AllowUnsafe): Int                       = self.getAndDecrement()
             inline def getAndAdd(v: Int)(using inline allow: AllowUnsafe): Int                       = self.getAndAdd(v)
             inline def addAndGet(v: Int)(using inline allow: AllowUnsafe): Int                       = self.addAndGet(v)
+            inline def getAndUpdate(inline f: Int => Int)(using inline allow: AllowUnsafe): Int      = self.getAndUpdate(f(_))
+            inline def updateAndGet(inline f: Int => Int)(using inline allow: AllowUnsafe): Int      = self.updateAndGet(f(_))
             inline def safe: AtomicInt                                                               = AtomicInt(self)
         end extension
     end Unsafe
@@ -252,6 +270,22 @@ final case class AtomicLong private (unsafe: AtomicLong.Unsafe):
       */
     inline def addAndGet(v: Long)(using inline frame: Frame): Long < IO = IO.Unsafe(unsafe.addAndGet(v))
 
+    /** Atomically updates the current value using the given function and returns the old value.
+      * @param f
+      *   The function to apply to the current value
+      * @return
+      *   The previous value
+      */
+    inline def getAndUpdate(inline f: Long => Long)(using inline frame: Frame): Long < IO = IO.Unsafe(unsafe.getAndUpdate(f(_)))
+
+    /** Atomically updates the current value using the given function and returns the updated value.
+      * @param f
+      *   The function to apply to the current value
+      * @return
+      *   The updated value
+      */
+    inline def updateAndGet(inline f: Long => Long)(using inline frame: Frame): Long < IO = IO.Unsafe(unsafe.updateAndGet(f(_)))
+
     /** Returns a string representation of the current value.
       * @return
       *   A string representation of the atomic long
@@ -320,6 +354,8 @@ object AtomicLong:
             inline def getAndDecrement()(using inline allow: AllowUnsafe): Long                        = self.getAndDecrement()
             inline def getAndAdd(v: Long)(using inline allow: AllowUnsafe): Long                       = self.getAndAdd(v)
             inline def addAndGet(v: Long)(using inline allow: AllowUnsafe): Long                       = self.addAndGet(v)
+            inline def getAndUpdate(inline f: Long => Long)(using inline allow: AllowUnsafe): Long     = self.getAndUpdate(f(_))
+            inline def updateAndGet(inline f: Long => Long)(using inline allow: AllowUnsafe): Long     = self.updateAndGet(f(_))
             inline def safe: AtomicLong                                                                = AtomicLong(self)
         end extension
     end Unsafe
@@ -494,19 +530,21 @@ final case class AtomicRef[A] private (unsafe: AtomicRef.Unsafe[A]):
       */
     inline def compareAndSet(curr: A, next: A)(using inline frame: Frame): Boolean < IO = IO.Unsafe(unsafe.compareAndSet(curr, next))
 
-    /** Atomically updates the current value using the given function.
+    /** Atomically updates the current value using the given function and returns the old value.
       * @param f
       *   The function to apply to the current value
+      * @return
+      *   The previous value
       */
-    inline def update[S](f: A => A)(using inline frame: Frame): Unit < IO = updateAndGet(f).unit
+    inline def getAndUpdate(inline f: A => A)(using inline frame: Frame): A < IO = IO.Unsafe(unsafe.getAndUpdate(f(_)))
 
     /** Atomically updates the current value using the given function and returns the updated value.
       * @param f
       *   The function to apply to the current value
       * @return
       *   The updated value
       */
-    inline def updateAndGet[S](f: A => A)(using inline frame: Frame): A < IO = IO.Unsafe(unsafe.updateAndGet(f(_)))
+    inline def updateAndGet(inline f: A => A)(using inline frame: Frame): A < IO = IO.Unsafe(unsafe.updateAndGet(f(_)))
 
     /** Returns a string representation of the current value.
       * @return
@@ -556,8 +594,8 @@ object AtomicRef:
             inline def lazySet(v: A)(using inline allow: AllowUnsafe): Unit                      = self.lazySet(v)
             inline def getAndSet(v: A)(using inline allow: AllowUnsafe): A                       = self.getAndSet(v)
             inline def compareAndSet(curr: A, next: A)(using inline allow: AllowUnsafe): Boolean = self.compareAndSet(curr, next)
-            def update[S](f: A => A)(using AllowUnsafe): Unit                                    = discard(self.updateAndGet(f(_)))
-            def updateAndGet[S](f: A => A)(using AllowUnsafe): A                                 = self.updateAndGet(f(_))
+            inline def getAndUpdate(inline f: A => A)(using inline allow: AllowUnsafe): A        = self.getAndUpdate(f(_))
+            inline def updateAndGet(inline f: A => A)(using inline allow: AllowUnsafe): A        = self.updateAndGet(f(_))
             inline def safe: AtomicRef[A]                                                        = AtomicRef(self)
         end extension
     end Unsafe

kyo-core/shared/src/main/scala/kyo/Atomic.scala

@@ -679,6 +679,10 @@ object Fiber extends FiberPlatformSpecific:
 
             def init[E, A]()(using AllowUnsafe): Unsafe[E, A] = IOPromise()
 
+            def initMasked[E, A]()(using AllowUnsafe): Unsafe[E, A] =
+                new IOPromise[E, A]:
+                    override def interrupt[E2 >: E](error: Result.Error[E2]): Boolean = false
+
             private[kyo] def fromIOPromise[E, A](p: IOPromise[E, A]): Unsafe[E, A] = p
 
             extension [E, A](self: Unsafe[E, A])