From 321ba1e3871404e570267d51c42e62f71e7b0d7c Mon Sep 17 00:00:00 2001 From: Pierre Ricadat Date: Sat, 24 Oct 2026 17:21:00 +0900 Subject: [PATCH 1/2] Add a direct backend for plain blocking code Add `sage-client-direct`, a client for plain blocking code on virtual threads with no effect library dependency. Each command blocks the calling thread and returns its result, failures are thrown, and interrupting the thread cancels the wait. - `sage-compat-direct` binds `kyo.compat` with plain JDK code. Long flatMap chains continue on a fresh virtual thread, and lock bodies always run on the caller's thread so they keep its thread-local context. - Streaming helpers return `Iterator`, subscriptions return a closeable blocking iterator, and `onMessage`, `onPatternMessage`, and `onShardMessage` deliver messages to a callback on a background virtual thread. - Add the direct cells to CI, the examples, the benchmarks, and the docs, plus two tests for interrupted closes in the shared runtime. --- .github/workflows/ci.yml | 2 +- README.md | 2 +- benchmarks/README.md | 1 + .../main/scala/sage/benchmarks/Clients.scala | 33 ++ .../sage/benchmarks/DirectBenchmarks.scala | 13 + benchmarks/merge-results.sh | 4 +- benchmarks/run.sh | 2 +- build.sbt | 61 ++- docs/client-side-caching.md | 2 +- docs/commands.md | 2 +- docs/distributed-locks.md | 2 +- docs/error-handling.md | 3 +- docs/faq.md | 1 + docs/getting-started.md | 41 +- docs/index.md | 4 +- docs/json.md | 4 +- docs/pipelines-transactions.md | 6 +- docs/pubsub.md | 24 +- docs/rate-limiting.md | 2 +- docs/streams.md | 13 +- examples/README.md | 6 +- .../examples/direct/CachedReadsExample.scala | 20 + .../examples/direct/CommandsExample.scala | 36 ++ .../sage/examples/direct/LockExample.scala | 21 + .../examples/direct/PipelinesExample.scala | 21 + .../sage/examples/direct/PubSubExample.scala | 30 ++ .../examples/direct/RateLimiterExample.scala | 20 + .../sage/examples/direct/StreamsExample.scala | 28 ++ .../scala/sage/examples/direct/Tour.scala | 22 + .../examples/direct/TransactionsExample.scala | 21 + .../sage/integration/DirectSmokeSuite.scala | 261 ++++++++++++ .../main/scala/sage/backend/SageClient.scala | 268 ++++++++++++ .../client/internal/DirectBindingSpec.scala | 96 +++++ .../internal/DirectLockCancellationSpec.scala | 36 ++ .../internal/MultiplexedConnectionSpec.scala | 15 + .../client/internal/SocketTransportSpec.scala | 35 ++ sage-compat-direct/README.md | 24 ++ .../scala/kyo/compat/CAtomicBoolean.scala | 53 +++ .../main/scala/kyo/compat/CAtomicInt.scala | 83 ++++ .../main/scala/kyo/compat/CAtomicLong.scala | 83 ++++ .../main/scala/kyo/compat/CAtomicRef.scala | 64 +++ .../src/main/scala/kyo/compat/CChannel.scala | 48 +++ .../src/main/scala/kyo/compat/CChunk.scala | 55 +++ .../src/main/scala/kyo/compat/CFiber.scala | 35 ++ .../src/main/scala/kyo/compat/CIO.scala | 382 ++++++++++++++++++ .../src/main/scala/kyo/compat/CLatch.scala | 43 ++ .../src/main/scala/kyo/compat/CLocal.scala | 57 +++ .../src/main/scala/kyo/compat/CMeter.scala | 60 +++ .../src/main/scala/kyo/compat/CPromise.scala | 62 +++ .../src/main/scala/kyo/compat/CStream.scala | 131 ++++++ .../main/scala/kyo/compat/DirectRuntime.scala | 190 +++++++++ 51 files changed, 2482 insertions(+), 46 deletions(-) create mode 100644 benchmarks/direct/src/main/scala/sage/benchmarks/Clients.scala create mode 100644 benchmarks/direct/src/main/scala/sage/benchmarks/DirectBenchmarks.scala create mode 100644 examples/direct/src/main/scala/sage/examples/direct/CachedReadsExample.scala create mode 100644 examples/direct/src/main/scala/sage/examples/direct/CommandsExample.scala create mode 100644 examples/direct/src/main/scala/sage/examples/direct/LockExample.scala create mode 100644 examples/direct/src/main/scala/sage/examples/direct/PipelinesExample.scala create mode 100644 examples/direct/src/main/scala/sage/examples/direct/PubSubExample.scala create mode 100644 examples/direct/src/main/scala/sage/examples/direct/RateLimiterExample.scala create mode 100644 examples/direct/src/main/scala/sage/examples/direct/StreamsExample.scala create mode 100644 examples/direct/src/main/scala/sage/examples/direct/Tour.scala create mode 100644 examples/direct/src/main/scala/sage/examples/direct/TransactionsExample.scala create mode 100644 integration-tests/direct/src/test/scala/sage/integration/DirectSmokeSuite.scala create mode 100644 sage-client/direct/src/main/scala/sage/backend/SageClient.scala create mode 100644 sage-client/direct/src/test/scala/sage/client/internal/DirectBindingSpec.scala create mode 100644 sage-client/direct/src/test/scala/sage/client/internal/DirectLockCancellationSpec.scala create mode 100644 sage-compat-direct/README.md create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CAtomicBoolean.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CAtomicInt.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CAtomicLong.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CAtomicRef.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CChannel.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CChunk.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CFiber.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CIO.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CLatch.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CLocal.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CMeter.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CPromise.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/CStream.scala create mode 100644 sage-compat-direct/src/main/scala/kyo/compat/DirectRuntime.scala diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 02be0049..a863fb3d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -89,7 +89,7 @@ jobs: strategy: fail-fast: false matrix: - backend: [Zio, Ce, Ox, Pekko, Kyo] + backend: [Zio, Ce, Ox, Pekko, Kyo, Direct] steps: - name: Checkout current branch uses: actions/checkout@v6 diff --git a/README.md b/README.md index 3d78fb37..6908ce75 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ Sage is a Redis and Valkey client for Scala 3. It implements the Redis protocol directly in Scala instead of wrapping a Java client. -- Choose the artifact for [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), [Ox](https://ox.softwaremill.com), or [Pekko](https://pekko.apache.org). Each artifact uses its ecosystem's native types. +- Choose the artifact for [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), [Ox](https://ox.softwaremill.com), or [Pekko](https://pekko.apache.org). Each artifact uses its ecosystem's native types. The direct artifact has no effect library dependency and serves plain blocking code. - The core implements RESP3, commands, and codecs in Scala 3. - Redis 8+ and Valkey 8+ support includes auto-pipelining, transactions, cluster routing, sharded pub/sub, client-side caching, rate limiting, distributed locks, and TLS. diff --git a/benchmarks/README.md b/benchmarks/README.md index e2f31571..de6c704b 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -25,6 +25,7 @@ Each backend runs in a separate JVM and `projectMatrix` cell. The Sage backends | `sage-ox` / `lettuce` / `rediscala` / `jedis` | `benchmarksOx` | `lettuce` is the asynchronous, auto-pipelined Java baseline. `jedis` is the synchronous RESP3 Java client and uses one pooled connection per concurrency lane. | | `sage-pekko` | `benchmarksPekko` | no native Pekko (Future) competitor exists. | | `sage-kyo` | `benchmarksKyo` | No native Kyo competitor exists. | +| `sage-direct` | `benchmarksDirect` | Runs each concurrency lane on a virtual thread. Compare it with `jedis` and `lettuce` from the Ox cell. | ## Running diff --git a/benchmarks/direct/src/main/scala/sage/benchmarks/Clients.scala b/benchmarks/direct/src/main/scala/sage/benchmarks/Clients.scala new file mode 100644 index 00000000..9ba25899 --- /dev/null +++ b/benchmarks/direct/src/main/scala/sage/benchmarks/Clients.scala @@ -0,0 +1,33 @@ +package sage.benchmarks + +import java.util.concurrent.Executors + +import scala.util.Using + +import sage.backend.* +import sage.client.{Endpoint, SageConfig, Topology} + +/** + * The direct benchmark cell contains only Sage. The Ox cell already measures Jedis and Lettuce from blocking code on virtual threads. + */ +object Clients { + def build(host: String, port: Int, name: String): BenchClient = name match { + case "sage-direct" => new SageDirectBench(host, port) + case other => throw new IllegalArgumentException(s"unknown client: $other") + } +} + +/** + * Runs each lane on its own virtual thread, as a framework that serves each request on a virtual thread would. + */ +final class SageDirectBench(host: String, port: Int) extends SageBench[[A] =>> A] { + + protected val client: SageClient = SageClient.connect(SageConfig(topology = Topology.Standalone(Endpoint(host, port)))) + + protected def run[A](result: A): Unit = () + + protected def inLanes[A](work: Payloads.Workload)(perKey: String => A): Any = + Using.resource(Executors.newVirtualThreadPerTaskExecutor()) { executor => + work.lanes.map(lane => executor.submit[Unit](() => lane.foreach(perKey))).foreach(_.get()) + } +} diff --git a/benchmarks/direct/src/main/scala/sage/benchmarks/DirectBenchmarks.scala b/benchmarks/direct/src/main/scala/sage/benchmarks/DirectBenchmarks.scala new file mode 100644 index 00000000..946bfdbb --- /dev/null +++ b/benchmarks/direct/src/main/scala/sage/benchmarks/DirectBenchmarks.scala @@ -0,0 +1,13 @@ +package sage.benchmarks + +import org.openjdk.jmh.annotations.Param + +class ThroughputBench extends ThroughputBenchBase { + + @Param(Array("sage-direct")) var client: String = "sage-direct" +} + +class CollectionBench extends CollectionBenchBase { + + @Param(Array("sage-direct")) var client: String = "sage-direct" +} diff --git a/benchmarks/merge-results.sh b/benchmarks/merge-results.sh index ce77cb9f..954520ab 100755 --- a/benchmarks/merge-results.sh +++ b/benchmarks/merge-results.sh @@ -1,11 +1,11 @@ #!/usr/bin/env bash -# Merge the per-cell JMH JSONs written by `sbt benchAll` into one all.json covering every client (5 sage backends + zio-redis + redis4cats + +# Merge the per-cell JMH JSONs written by `sbt benchAll` into one all.json covering every client (6 sage backends + zio-redis + redis4cats + # lettuce + rediscala + jedis). Cells use the same benchmark class and method names and a unique `client` parameter. The resulting all.json # can be compared as one run or uploaded to https://jmh.morethan.io. set -euo pipefail cd "$(dirname "$0")/results" -cells=(zio.json ce.json ox.json pekko.json kyo.json) +cells=(zio.json ce.json ox.json pekko.json kyo.json direct.json) present=() for f in "${cells[@]}"; do [ -s "$f" ] && present+=("$f"); done if [ ${#present[@]} -eq 0 ]; then diff --git a/benchmarks/run.sh b/benchmarks/run.sh index 68372604..599171e6 100755 --- a/benchmarks/run.sh +++ b/benchmarks/run.sh @@ -14,7 +14,7 @@ root="$(cd "$(dirname "$0")/.." && pwd)" cd "$root" args="$*" -cells=(benchmarksZio:zio benchmarksCe:ce benchmarksOx:ox benchmarksPekko:pekko benchmarksKyo:kyo) +cells=(benchmarksZio:zio benchmarksCe:ce benchmarksOx:ox benchmarksPekko:pekko benchmarksKyo:kyo benchmarksDirect:direct) failed=0 rm -f benchmarks/results/*.json diff --git a/build.sbt b/build.sbt index ebbb36eb..b6319b03 100644 --- a/build.sbt +++ b/build.sbt @@ -32,6 +32,10 @@ val PekkoLib = CompatBackendAxis.external("pekko", "Pekko", "-pekko", Set("jvm") // suffixes match the removed built-in, which keeps cell ids and artifact names unchanged. val CeLib = CompatBackendAxis.local("ce", "Ce", "-ce", Set("jvm")) +// The direct backend binds `kyo.compat` to plain blocking code on JDK virtual threads, implemented locally in `sage-compat-direct`. Its +// artifacts depend only on Sage modules and the Scala library. +val DirectLib = CompatBackendAxis.local("direct", "Direct", "-direct", Set("jvm")) + inThisBuild( List( scalaVersion := scala3Version, @@ -50,30 +54,34 @@ name := "sage" addCommandAlias( "fmt", "all scalafmtSbt scalafmt test:scalafmt " + - "benchmarksZio/scalafmt benchmarksCe/scalafmt benchmarksOx/scalafmt benchmarksPekko/scalafmt benchmarksKyo/scalafmt" + "benchmarksZio/scalafmt benchmarksCe/scalafmt benchmarksOx/scalafmt benchmarksPekko/scalafmt benchmarksKyo/scalafmt " + + "benchmarksDirect/scalafmt" ) addCommandAlias( "check", "all scalafmtSbtCheck scalafmtCheck test:scalafmtCheck " + - "benchmarksZio/scalafmtCheck benchmarksCe/scalafmtCheck benchmarksOx/scalafmtCheck benchmarksPekko/scalafmtCheck benchmarksKyo/scalafmtCheck" + "benchmarksZio/scalafmtCheck benchmarksCe/scalafmtCheck benchmarksOx/scalafmtCheck benchmarksPekko/scalafmtCheck benchmarksKyo/scalafmtCheck " + + "benchmarksDirect/scalafmtCheck" ) addCommandAlias( "testUnit", - "all core/test opentelemetry/test clientZio/test clientCe/test clientOx/test clientPekko/test clientKyo/test " + + "all core/test opentelemetry/test clientZio/test clientCe/test clientOx/test clientPekko/test clientKyo/test clientDirect/test " + "clientFuture/Test/compile integrationTestsFuture/Test/compile integrationTestsPekko/Test/compile " + - "benchmarksZio/compile benchmarksCe/compile benchmarksOx/compile benchmarksPekko/compile benchmarksKyo/compile " + + "benchmarksZio/compile benchmarksCe/compile benchmarksOx/compile benchmarksPekko/compile benchmarksKyo/compile benchmarksDirect/compile " + "examplesZio/Compile/compile examplesCe/Compile/compile examplesOx/Compile/compile examplesPekko/Compile/compile " + - "examplesKyo/Compile/compile examplesFuture/Compile/compile; " + - // The conformance suite has wall-clock parallelism checks. Run it after the parallel block so five forked test cells do not starve it. - "ceConformanceCe/test" + "examplesKyo/Compile/compile examplesDirect/Compile/compile examplesFuture/Compile/compile; " + + // The conformance suite has wall-clock parallelism checks. Run it after the parallel block so six forked test cells do not starve it. + "ceConformanceCe/test; directConformanceDirect/test" ) addCommandAlias("conformanceCe", "ceConformanceCe/test") +addCommandAlias("conformanceDirect", "directConformanceDirect/test") addCommandAlias("itZio", "integrationTestsZio/test") addCommandAlias("itCe", "integrationTestsCe/test") addCommandAlias("itOx", "integrationTestsOx/test") addCommandAlias("itPekko", "integrationTestsPekko/test") addCommandAlias("itKyo", "integrationTestsKyo/test") +addCommandAlias("itDirect", "integrationTestsDirect/test") addCommandAlias("exampleKyo", "examplesKyo/run") @@ -86,7 +94,7 @@ addCommandAlias( addCommandAlias( "docAll", - "all core/doc opentelemetry/doc compatCe/doc clientZio/doc clientCe/doc clientOx/doc clientPekko/doc clientKyo/doc" + "all core/doc opentelemetry/doc compatCe/doc compatDirect/doc clientZio/doc clientCe/doc clientOx/doc clientPekko/doc clientKyo/doc clientDirect/doc" ) lazy val root = project @@ -96,7 +104,7 @@ lazy val root = project // on-demand benchmark runs. Use benchAll or benchmarks/Jmh/run to run them directly. .aggregate( core.projectRefs ++ client.projectRefs ++ opentelemetry.projectRefs ++ integrationTests.projectRefs ++ examples.projectRefs ++ - Seq[ProjectReference](compatCe): _* + Seq[ProjectReference](compatCe, compatDirect): _* ) // Pure sans-IO core: RESP3 protocol, command model, codecs. Zero external dependencies. @@ -156,6 +164,20 @@ lazy val ceConformance = (projectMatrix in file("sage-compat-ce/.conformance")) .bindLocally(CeLib, compatCe) .compatConformance() +// A dependency-free binding of the `kyo.compat` package for blocking code on virtual threads. The `direct` cells of the client matrices bind +// to this project through `bindLocally`. +lazy val compatDirect = project + .in(file("sage-compat-direct")) + .settings(name := "sage-compat-direct") + .settings(commonSettings) + +// Run the bundled `kyo.compat` conformance suite against the direct binding, as `ceConformance` does for the Cats Effect binding. +lazy val directConformance = (projectMatrix in file("sage-compat-direct/.conformance")) + .settings(publish / skip := true) + .compatLibrary(DirectLib)(VirtualAxis.jvm)(Seq(scala3Version)) + .bindLocally(DirectLib, compatDirect) + .compatConformance() + // Runtime written once against kyo-compat, cross-published per backend. JDK 21+. lazy val client = (projectMatrix in file("sage-client")) .dependsOn(core) @@ -184,8 +206,9 @@ lazy val client = (projectMatrix in file("sage-client")) else Seq.empty } ) - .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib)(VirtualAxis.jvm)(Seq(scala3Version)) + .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib, DirectLib)(VirtualAxis.jvm)(Seq(scala3Version)) .bindLocally(CeLib, compatCe) + .bindLocally(DirectLib, compatDirect) // Run each backend's smoke suite against a real server. Run the shared command, security, cluster, master-replica, rate-limit, and lock suites // on one backend because those features use the same implementation in every backend. @@ -218,10 +241,11 @@ lazy val integrationTests = (projectMatrix in file("integration-tests")) Tests.Filter(name => !isAnchor && (isDesignated || !onceOnly.exists(name.startsWith))) } ) - .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib)(VirtualAxis.jvm)(Seq(scala3Version)) + .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib, DirectLib)(VirtualAxis.jvm)(Seq(scala3Version)) .bindLocally(CeLib, compatCe) + .bindLocally(DirectLib, compatDirect) -// Build runnable, unpublished examples for ZIO, Cats Effect, Ox, Kyo, and Pekko in separate cells so each uses its native client artifact. The +// Build runnable, unpublished examples for ZIO, Cats Effect, Ox, Kyo, Pekko, and the direct client in separate cells so each uses its native client artifact. The // root project compiles them in CI through testUnit. The Future cell compiles examples/shared. Run an example against a local server with a // command such as `sbt examplesZio/run`. See examples/README.md. lazy val examples = (projectMatrix in file("examples")) @@ -232,12 +256,13 @@ lazy val examples = (projectMatrix in file("examples")) // These runnable samples are not published. Skip API documentation because third-party parent types such as Cats Effect's IOApp contain // links that cannot be resolved here. .settings(Compile / doc / sources := Seq.empty) - .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib)(VirtualAxis.jvm)(Seq(scala3Version)) + .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib, DirectLib)(VirtualAxis.jvm)(Seq(scala3Version)) .bindLocally(CeLib, compatCe) + .bindLocally(DirectLib, compatDirect) // Runtime end-to-end benchmark harness (JMH), used only during development. Each backend has a separate cell. The ZIO cell includes -// zio-redis, the Cats Effect cell includes redis4cats, and the Ox cell includes Lettuce, Rediscala, and Jedis. The Kyo and Pekko cells include -// only Sage. See benchmarks/README.md. +// zio-redis, the Cats Effect cell includes redis4cats, and the Ox cell includes Lettuce, Rediscala, and Jedis. The Kyo, Pekko, and direct +// cells include only Sage. See benchmarks/README.md. lazy val benchmarks = (projectMatrix in file("benchmarks")) .dependsOn(client) .enablePlugins(JmhPlugin) @@ -262,8 +287,9 @@ lazy val benchmarks = (projectMatrix in file("benchmarks")) else Seq.empty } ) - .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib)(VirtualAxis.jvm)(Seq(scala3Version)) + .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib, DirectLib)(VirtualAxis.jvm)(Seq(scala3Version)) .bindLocally(CeLib, compatCe) + .bindLocally(DirectLib, compatDirect) // The runtime benchmark harness: runs every backend cell's JMH suite against its own self-provisioned Redis (the future anchor cell is skipped), // each writing JMH JSON to benchmarks/results/.json. benchmarks/merge-results.sh merges them into one all.json covering every client @@ -274,7 +300,8 @@ addCommandAlias( ";benchmarksCe/Jmh/run -rf json -rff benchmarks/results/ce.json " + ";benchmarksOx/Jmh/run -rf json -rff benchmarks/results/ox.json " + ";benchmarksPekko/Jmh/run -rf json -rff benchmarks/results/pekko.json " + - ";benchmarksKyo/Jmh/run -rf json -rff benchmarks/results/kyo.json" + ";benchmarksKyo/Jmh/run -rf json -rff benchmarks/results/kyo.json " + + ";benchmarksDirect/Jmh/run -rf json -rff benchmarks/results/direct.json" ) lazy val commonSettings = Def.settings( diff --git a/docs/client-side-caching.md b/docs/client-side-caching.md index ac43e329..50a30d4a 100644 --- a/docs/client-side-caching.md +++ b/docs/client-side-caching.md @@ -6,7 +6,7 @@ Caching is opt-in per call. An ordinary `get` always goes to the server; only `c ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.set("cached:key", "v1") // first fetches and caches; second is a local hit val v1 = client.cached(Commands.get[String, String]("cached:key"), 1.minute) diff --git a/docs/commands.md b/docs/commands.md index 70b8a1cc..9df72191 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -8,7 +8,7 @@ A `Command[Out]` describes a server command without running it. Client methods c ::: code-group -```scala [Ox] +```scala [Ox · Direct] // per-command sugar val greeting = client.get[String]("greeting") diff --git a/docs/distributed-locks.md b/docs/distributed-locks.md index c8cedac0..dd422623 100644 --- a/docs/distributed-locks.md +++ b/docs/distributed-locks.md @@ -55,7 +55,7 @@ Lock checks share `dedicatedPool` with blocking commands and transactions. A ful If replicas do not acknowledge acquisition in time, Sage raises `SageException.TimedOut` without starting the body. Acquisition can also raise `SageException.LockLost` if the granting master changes role or the lease expires before the body starts. Sage retries temporary renewal failures while the usable lease remains. If it cannot recover, or loses ownership, the operation fails with `SageException.LockLost` and attempts to cancel the body. Server and connection errors propagate through your backend's normal failure channel. Cleanup preserves an existing body failure or cancellation. -Cancellation is cooperative and cannot undo completed work. Uninterruptible work can delay completion. Pekko uses `Future`, so its body may continue after lock loss even though Sage stops renewing the lock. A timed-out or cancelled acquisition can leave the lock held until its lease expires. +Cancellation is cooperative and cannot undo completed work. Uninterruptible work can delay completion. Pekko uses `Future`, so its body may continue after lock loss even though Sage stops renewing the lock. With the direct client, the body runs on the calling thread, so it sees that thread's thread-local values, such as a framework's transaction context. Sage interrupts that thread to stop the body. A timed-out or cancelled acquisition can leave the lock held until its lease expires. ## Limits diff --git a/docs/error-handling.md b/docs/error-handling.md index 690b388b..a3cdbf18 100644 --- a/docs/error-handling.md +++ b/docs/error-handling.md @@ -1,6 +1,6 @@ # Error handling -Every Sage failure is a `SageException` in a sealed hierarchy that you can match exhaustively. ZIO and Kyo include `SageException` in the error type as `IO[SageException, *]` and `Abort[SageException]`. Cats Effect raises the exception in `IO`, Ox throws it, and Pekko returns a failed `scala.concurrent.Future`. +Every Sage failure is a `SageException` in a sealed hierarchy that you can match exhaustively. ZIO and Kyo include `SageException` in the error type as `IO[SageException, *]` and `Abort[SageException]`. Cats Effect raises the exception in `IO`, Ox and the direct client throw it, and Pekko returns a failed `scala.concurrent.Future`. ## The hierarchy @@ -77,4 +77,5 @@ Each ecosystem reports the same `SageException` through its normal failure chann - Cats Effect raises the exception in `IO`. Recover with `handleErrorWith` or `recoverWith` and match on `SageException`. - Kyo returns an `Abort[SageException]`. Handle it with the `Abort` combinators. - Ox throws the exception in direct style. Handle it with `try` and `catch`. +- The direct client throws the exception from the blocking call. Handle it with `try` and `catch`. - Pekko returns a failed `scala.concurrent.Future`. Recover with `recover` or `recoverWith`. diff --git a/docs/faq.md b/docs/faq.md index d2bca12c..fec8bc3c 100644 --- a/docs/faq.md +++ b/docs/faq.md @@ -15,6 +15,7 @@ One per Scala stack, all sharing the same runtime: - Kyo: `sage-client-kyo` - Ox: `sage-client-ox` - Pekko: `sage-client-pekko` +- Plain blocking code, with no effect library: `sage-client-direct` `sage-core` comes in transitively, so you depend on the backend artifact only. See [Getting started](/getting-started). diff --git a/docs/getting-started.md b/docs/getting-started.md index 791eb9fb..ba19a0c7 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -2,7 +2,7 @@ Sage is a [Redis](https://redis.io) and [Valkey](https://valkey.io) client for [Scala 3](https://www.scala-lang.org/). It implements RESP3, commands, and codecs directly in Scala. The core has no dependencies on an effect system. -Sage provides integrations for [Ox](https://ox.softwaremill.com), [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), and [Apache Pekko](https://pekko.apache.org). Each integration uses its ecosystem's native types. Sage targets Redis 8+ and Valkey 8+. It supports Scala 3.9.x LTS and later and requires JDK 21 or later. +Sage provides integrations for [Ox](https://ox.softwaremill.com), [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), and [Apache Pekko](https://pekko.apache.org). Each integration uses its ecosystem's native types. The direct client has no dependency on an effect library: each command blocks the calling thread and returns its result, which suits frameworks that run each request on a virtual thread. Sage targets Redis 8+ and Valkey 8+. It supports Scala 3.9.x LTS and later and requires JDK 21 or later. ## Installation @@ -30,13 +30,17 @@ Add the artifact for your Scala stack. The core is pulled in transitively, so yo "com.github.ghostdogpr" %% "sage-client-pekko" % "@VERSION@" ``` +```scala [Direct] +"com.github.ghostdogpr" %% "sage-client-direct" % "@VERSION@" +``` + ::: Import `sage.*` for commands and connection configuration. Import `sage.backend.*` for the client. Every backend uses these imports, but each one has a different dependency. ## Your first connection -A `SageClient` owns all connections to one server or cluster. Build it from a `SageConfig` with the resource type for your backend. Ox and Kyo use a scope, ZIO uses a `ZLayer`, Cats Effect uses a `Resource`, and Pekko uses a `use` block that closes the client when the program finishes. All five integrations provide the same commands. +A `SageClient` owns all connections to one server or cluster. Build it from a `SageConfig` with the resource type for your backend. Ox and Kyo use a scope, ZIO uses a `ZLayer`, Cats Effect uses a `Resource`, and Pekko and the direct client use a `use` block that closes the client when the program finishes. All six integrations provide the same commands. ::: code-group @@ -58,6 +62,22 @@ import sage.backend.* } ``` +```scala [Direct] +import sage.* +import sage.backend.* + +@main def main(): Unit = { + val config = SageConfig( + topology = Topology.Standalone(Endpoint("localhost", 6379)) + ) + SageClient.use(config) { client => + client.set("greeting", "hello") + val greeting = client.get[String]("greeting") + println(s"greeting=$greeting") // Some("hello") + } +} +``` + ```scala [ZIO] import zio.* @@ -170,7 +190,7 @@ Method names match Redis commands and are grouped by family (strings, hashes, li ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.set("greeting", "hello") val greeting = client.get[String]("greeting") // Some("hello") client.incrBy("counter", 10) @@ -205,7 +225,7 @@ Put commands in a pipeline to send them in one round trip and receive a typed tu ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.set("pipe:a", "x") client.set("pipe:n", 10) val tuple = client.pipeline( @@ -235,7 +255,7 @@ A transaction runs a pipeline atomically with `MULTI` and `EXEC`. You can use `W ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.set("tx:n", 1) val result = client.transaction { tx => tx.watch("tx:n") @@ -270,7 +290,7 @@ The distinction is covered in [Pipelines & transactions](/pipelines-transactions ### Pub/Sub -Subscribing returns the backend's native stream type. Ox returns `Flow`, ZIO returns `ZStream`, Cats Effect returns an fs2 `Stream`, Kyo returns `Stream`, and Pekko returns `Source`. Ending the stream or closing its scope unsubscribes. +Subscribing returns the backend's native stream type. Ox returns `Flow`, ZIO returns `ZStream`, Cats Effect returns an fs2 `Stream`, Kyo returns `Stream`, Pekko returns `Source`, and the direct client returns a blocking `Iterator` that you close to unsubscribe. Ending the stream or closing its scope unsubscribes. These examples publish immediately after subscribing, so they use the variant that waits for the server to confirm the subscription first. [Pub/Sub](/pubsub) explains when you need it. @@ -282,6 +302,13 @@ val news = client.subscribeScoped[String]("news") val messages = news.take(3).runToList() ``` +```scala [Direct] +val messages = Using.resource(client.subscribe[String]("news")) { news => + (1 to 3).foreach(i => client.publish("news", s"item-$i")) + news.take(3).map(_.payload).toList +} +``` + ```scala [ZIO] ZIO.scoped { for { @@ -337,7 +364,7 @@ Opt a read into client-side caching per call. The first read fetches and caches; ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.set("cached:key", "v1") // first fetches and caches; second is a local hit val v1 = client.cached(Commands.get[String, String]("cached:key"), 1.minute) diff --git a/docs/index.md b/docs/index.md index 9ce094c8..c5cc99e8 100644 --- a/docs/index.md +++ b/docs/index.md @@ -4,7 +4,7 @@ layout: home hero: name: "Sage" text: "A Redis & Valkey client for Scala 3" - tagline: Native Scala clients for ZIO, Cats Effect, Kyo, Ox, and Pekko. + tagline: Native Scala clients for ZIO, Cats Effect, Kyo, Ox, Pekko, and plain blocking code. image: light: /sage.svg dark: /sage-dark.svg @@ -19,7 +19,7 @@ hero: features: - title: Native backend types - details: Choose a ZIO, Cats Effect, Kyo, Ox, or Pekko artifact. Each one uses its ecosystem's effect and stream types. + details: Choose a ZIO, Cats Effect, Kyo, Ox, or Pekko artifact, each with its ecosystem's effect and stream types, or the direct artifact for plain blocking code. - title: Redis protocol in Scala details: Sage implements RESP3, commands, and codecs directly in Scala 3 instead of wrapping a Java client. - title: Redis and Valkey support diff --git a/docs/json.md b/docs/json.md index d55494da..c41bb40b 100644 --- a/docs/json.md +++ b/docs/json.md @@ -25,7 +25,7 @@ A value you write is raw JSON text. The built-in `String` codec passes it throug ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.jsonSet("user:1", JsonPath.root, """{"name":"Ada","age":36,"tags":["a","b"]}""") val name = client.jsonGet[String]("user:1", JsonPath("$.name")) // Some("[\"Ada\"]") val age = client.jsonGet[String]("user:1", JsonPath("$.age")) // Some("[36]") @@ -71,7 +71,7 @@ Numbers, strings, booleans, arrays, and objects each have their commands. All of ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.jsonSet("doc", JsonPath.root, """{"n":1,"s":"ab","flag":false,"xs":[1,2,3]}""") client.jsonNumIncrBy("doc", JsonPath("$.n"), 5) // Vector(Some(6.0)) client.jsonStrAppend("doc", JsonPath("$.s"), "\"c\"") // Vector(Some(3)) diff --git a/docs/pipelines-transactions.md b/docs/pipelines-transactions.md index 11ee8d8f..023939bb 100644 --- a/docs/pipelines-transactions.md +++ b/docs/pipelines-transactions.md @@ -8,7 +8,7 @@ A pipeline sends several `Command` values in one round trip and returns their re ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.set("pipe:a", "x") client.set("pipe:n", 10) val tuple = client.pipeline( @@ -46,7 +46,7 @@ By default, a pipeline fails as a whole if any command fails. Use `pipelineAttem ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.set("pipe:str", "hello") // INCR on a non-numeric string fails only at its own position; // the GET still succeeds @@ -84,7 +84,7 @@ A transaction runs a pipeline atomically with `MULTI`/`EXEC` on a temporary dedi ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.set("tx:n", 1) val result = client.transaction { tx => tx.watch("tx:n") diff --git a/docs/pubsub.md b/docs/pubsub.md index 6aeb3adf..53e773f3 100644 --- a/docs/pubsub.md +++ b/docs/pubsub.md @@ -1,6 +1,6 @@ # Pub/Sub -Subscribing returns the backend's native stream type. Ox returns `Flow`, ZIO returns `ZStream`, Cats Effect returns an fs2 `Stream`, Kyo returns `Stream`, and Pekko returns `Source`. Each message contains its channel and a payload decoded by a `ValueCodec`. Ending the stream or closing its scope unsubscribes. If the server refuses a subscribed name, for example with `NOPERM` from an ACL, the stream fails with that `ServerError`, including when a reconnect subscribes the name again. Temporary errors such as `BUSY` or `LOADING` do not end the stream; Sage subscribes again after a backoff. +Subscribing returns the backend's native stream type. Ox returns `Flow`, ZIO returns `ZStream`, Cats Effect returns an fs2 `Stream`, Kyo returns `Stream`, Pekko returns `Source`, and the direct client returns a blocking `Iterator` that you close to unsubscribe. Each message contains its channel and a payload decoded by a `ValueCodec`. Ending the stream or closing its scope unsubscribes. If the server refuses a subscribed name, for example with `NOPERM` from an ACL, the stream fails with that `ServerError`, including when a reconnect subscribes the name again. Temporary errors such as `BUSY` or `LOADING` do not end the stream; Sage subscribes again after a backoff. ## Classic channels @@ -14,6 +14,13 @@ val news = client.subscribeScoped[String]("news") val messages = news.take(3).runToList() ``` +```scala [Direct] +val messages = Using.resource(client.subscribe[String]("news")) { news => + (1 to 3).foreach(i => client.publish("news", s"item-$i")) + news.take(3).map(_.payload).toList +} +``` + ```scala [ZIO] ZIO.scoped { for { @@ -69,6 +76,7 @@ The plain `subscribe` returns the stream immediately and registers the subscript - **ZIO / Kyo**: `subscribeScoped`, a scoped effect. - **Cats Effect**: `subscribeResource`, a `Resource`. - **Ox**: `subscribeScoped`, bound to the enclosing Ox scope. +- **Direct**: `subscribe` waits for confirmation before it returns. Close the iterator to unsubscribe, for example with `scala.util.Using`. - **Pekko**: plain `subscribe` returns a `Source` whose materialized `Future[Done]` completes once registered; await it before publishing. On a standalone or master-replica server, confirmation guarantees that the subscription is ready. In a cluster, this guarantee is best-effort. With the scoped and resource variants, the subscription remains active until the scope closes. On Pekko, ending the `Source` unsubscribes. @@ -78,6 +86,20 @@ On a standalone or master-replica server, confirmation guarantees that the subsc All classic subscriptions share one **subscription connection**, created the first time you subscribe and closed when the last subscription ends. Ordinary commands use a different connection. A slow consumer may delay its own subscriptions, but it does not delay command replies. The subscription connection re-issues every active subscription automatically on reconnect. ::: +## Callbacks on the direct client + +The direct client can also call a function for each message instead of returning an iterator. `onMessage`, `onPatternMessage`, and `onShardMessage` subscribe, wait for confirmation, and deliver each message in order on a virtual thread that Sage starts. They return an `AutoCloseable`. Closing it unsubscribes and waits until the function is no longer running. + +```scala +val subscription = client.onMessage[String]("orders") { message => + println(s"${message.channel}: ${message.payload}") +} +// later, for example when the application shuts down +subscription.close() +``` + +If the function throws, the failure goes to the delivery thread's uncaught exception handler and delivery continues with the next message. The delivery thread does not inherit the subscriber's inheritable thread-local values, so request or transaction context does not leak into it. + ## Sharded channels (cluster) In a cluster, a **shard channel** keeps its traffic within the shard that owns the channel's slot. `sSubscribe` and `sPublish` target that node instead of broadcasting across the whole cluster. Shard channels do not support pattern subscriptions, and they deliver ordinary messages. diff --git a/docs/rate-limiting.md b/docs/rate-limiting.md index 1b9921c2..4fb8d11e 100644 --- a/docs/rate-limiting.md +++ b/docs/rate-limiting.md @@ -8,7 +8,7 @@ The limiter stores its state on the server rather than in process memory. The li ::: code-group -```scala [Ox] +```scala [Ox · Direct] val limiter = client.rateLimiter[String](RateLimit.perSecond(100)) val decision = limiter.tryAcquire("user:42") val allowed = decision.isAllowed diff --git a/docs/streams.md b/docs/streams.md index 7e35db1e..f517a6d1 100644 --- a/docs/streams.md +++ b/docs/streams.md @@ -8,7 +8,7 @@ A stream is an append-only log. Each entry has an ID made from a millisecond tim ::: code-group -```scala [Ox] +```scala [Ox · Direct] client.del("stream:orders") client.xAdd("stream:orders")(("item", "book"), ("qty", "2")) client.xAdd("stream:orders")(("item", "pen"), ("qty", "5")) @@ -37,7 +37,7 @@ A consumer group divides a stream's entries among several consumers without over ::: code-group -```scala [Ox] +```scala [Ox · Direct] // create the group reading from the start of the stream client.xGroupCreate( "stream:orders", @@ -74,6 +74,8 @@ Each command uses a separate type for the ID values it accepts. `XADD` takes an For a long-running worker, `xConsume` tails a group and runs your handler on each entry. It first replays entries that this consumer left pending during a previous run. It then blocks while waiting for new entries. Sage acknowledges an entry after the handler succeeds. If the handler fails, the entry remains in the PEL for another attempt. +With the direct client, `xConsume` blocks the calling thread until the thread is interrupted or the handler throws. Run it on its own virtual thread and interrupt that thread to stop it. + On Pekko, a `Future` cannot be canceled. The loop runs in the background, and `xConsume` returns a `RunningConsumer`. Call `stop()` to halt it between entries, then await `completion`. ::: tip At-least-once delivery @@ -89,6 +91,13 @@ client.xConsume[String, String]("workers", "w1", "stream:orders") { } ``` +```scala [Direct] +// blocks the calling thread until it is interrupted +client.xConsume[String, String]("workers", "w1", "stream:orders") { + entry => println(s"got ${entry.id}: ${entry.fields}") +} +``` + ```scala [ZIO] client.xConsume[String, String]("workers", "w1", "stream:orders") { entry => Console.printLine(s"got ${entry.id}: ${entry.fields}") diff --git a/examples/README.md b/examples/README.md index 2dd66638..10e9f4db 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,6 +1,6 @@ # Examples -These runnable examples cover every Sage backend. They use ZIO `Task`, Cats Effect `IO`, Ox direct style, Kyo computations, or Pekko `Future` and Pekko Streams. The build does not publish this module, but CI compiles it. +These runnable examples cover every Sage backend. They use ZIO `Task`, Cats Effect `IO`, Ox direct style, Kyo computations, Pekko `Future` and Pekko Streams, or plain blocking calls with the direct client. The build does not publish this module, but CI compiles it. Import `sage.*` for commands and connection configuration. Import `sage..*` for the client. @@ -13,9 +13,10 @@ ce/ …Example.scala + Tour the Cats Effect tour, plus the TLS/ACL spotl ox/ …Example.scala + Tour the Ox tour, plus the master-replica + ReadFrom spotlight kyo/ …Example.scala + Tour the Kyo tour pekko/ …Example.scala + Tour the Pekko tour (scala.concurrent.Future + Pekko Streams) +direct/ …Example.scala + Tour the direct tour (plain blocking calls, iterators, and callbacks) ``` -Each backend's `Tour` is a runnable entry point. It constructs the client with a ZIO `layer`, a Cats Effect `resource`, an Ox or Kyo `scoped` value, or Pekko `connect` with an explicit `close` and a typed `ActorSystem`. Each tour runs commands from several families, a pipeline, a `WATCH` transaction, classic pub/sub, and a cached read. The `...Example` objects contain the individual snippets used by the tour. +Each backend's `Tour` is a runnable entry point. It constructs the client with a ZIO `layer`, a Cats Effect `resource`, an Ox or Kyo `scoped` value, Pekko `connect` with an explicit `close` and a typed `ActorSystem`, or the direct client's `use`, which closes the client when its block returns. Each tour runs commands from several families, a pipeline, a `WATCH` transaction, classic pub/sub, and a cached read. The `...Example` objects contain the individual snippets used by the tour. ## Running the tours @@ -33,6 +34,7 @@ sbt examplesCe/run sbt examplesOx/run sbt exampleKyo sbt examplesPekko/run +sbt examplesDirect/run ``` ## Spotlights diff --git a/examples/direct/src/main/scala/sage/examples/direct/CachedReadsExample.scala b/examples/direct/src/main/scala/sage/examples/direct/CachedReadsExample.scala new file mode 100644 index 00000000..9af5047b --- /dev/null +++ b/examples/direct/src/main/scala/sage/examples/direct/CachedReadsExample.scala @@ -0,0 +1,20 @@ +package sage.examples.direct + +import scala.concurrent.duration.* + +import sage.* +import sage.backend.* + +/** + * A Cached Read opts in to client-side caching per call: the first read fetches and caches, the second is served from the local cache until + * a server invalidation push or the TTL evicts it. + */ +object CachedReadsExample { + + def run(client: SageClient): Unit = { + client.set("cached:key", "v1") + val first = client.cached(Commands.get[String, String]("cached:key"), 1.minute) // fetch + cache + val second = client.cached(Commands.get[String, String]("cached:key"), 1.minute) // local hit + println(s"first=$first second=$second") + } +} diff --git a/examples/direct/src/main/scala/sage/examples/direct/CommandsExample.scala b/examples/direct/src/main/scala/sage/examples/direct/CommandsExample.scala new file mode 100644 index 00000000..fb4b057b --- /dev/null +++ b/examples/direct/src/main/scala/sage/examples/direct/CommandsExample.scala @@ -0,0 +1,36 @@ +package sage.examples.direct + +import scala.concurrent.duration.* + +import sage.* +import sage.backend.* +import sage.examples.User + +/** + * Commands across several families. Each method blocks the calling thread until the reply arrives and returns the value directly. + */ +object CommandsExample { + + def run(client: SageClient): Unit = { + // strings, numbers, and a conditional write with a TTL + client.set("greeting", "hello") + val greeting = client.get[String]("greeting") + client.incrBy("counter", 10) + client.set("session", "token", expiry = SetExpiry.In(1.minute), condition = SetCondition.IfNotExists) + // hashes + client.hSet("user:1", ("name", "Ada"), ("age", "36")) + val profile = client.hGetAll[String, String]("user:1") + // lists: push data before the blocking pop so this example completes immediately + client.rPush("queue", "a", "b", "c") + val head = client.blPop[String]("queue")(BlockTimeout.Forever) + // sets and sorted sets + client.sAdd("tags", "scala", "redis") + client.zAdd("scores")(("ada", 36.0)) + // store and retrieve a custom type using its ValueCodec + client.set("user:ada", User("Ada", 36)) + val ada = client.get[User]("user:ada") + // a raw Lua eval yields a Frame; decode it with the strict helpers + val sum = client.eval("return 1 + 1") + println(s"greeting=$greeting profile=$profile head=$head ada=$ada sum=${sum.asLong}") + } +} diff --git a/examples/direct/src/main/scala/sage/examples/direct/LockExample.scala b/examples/direct/src/main/scala/sage/examples/direct/LockExample.scala new file mode 100644 index 00000000..d3b59e56 --- /dev/null +++ b/examples/direct/src/main/scala/sage/examples/direct/LockExample.scala @@ -0,0 +1,21 @@ +package sage.examples.direct + +import scala.concurrent.duration.* + +import sage.backend.* + +/** + * Serializes a counter update across processes that use the same lock namespace and key. + */ +object LockExample { + def run(client: SageClient): Unit = { + val locks = client.lock[String](namespace = "example-locks") + val next = locks.withLock("counter", waitTimeout = 2.seconds) { + val current = client.get[Int]("example:locked-counter") + val next = current.getOrElse(0) + 1 + client.set("example:locked-counter", next) + next + } + println(s"Updated counter to $next") + } +} diff --git a/examples/direct/src/main/scala/sage/examples/direct/PipelinesExample.scala b/examples/direct/src/main/scala/sage/examples/direct/PipelinesExample.scala new file mode 100644 index 00000000..23aa3864 --- /dev/null +++ b/examples/direct/src/main/scala/sage/examples/direct/PipelinesExample.scala @@ -0,0 +1,21 @@ +package sage.examples.direct + +import sage.* +import sage.backend.* + +/** + * A Pipeline is an applicative composition of Commands sent in one round-trip, yielding a typed tuple. `pipelineAttempt` keeps each + * position's failure separate instead of failing the whole call. + */ +object PipelinesExample { + + def run(client: SageClient): Unit = { + client.set("pipe:a", "x") + client.set("pipe:n", 10) + val tuple = client.pipeline((Commands.get[String, String]("pipe:a"), Commands.incrBy("pipe:n", 5))) + client.set("pipe:str", "hello") + // INCR on a non-numeric string fails only at its own position; the GET still succeeds + val attempt = client.pipelineAttempt((Commands.get[String, String]("pipe:str"), Commands.incr("pipe:str"))) + println(s"tuple=$tuple attempt=$attempt") + } +} diff --git a/examples/direct/src/main/scala/sage/examples/direct/PubSubExample.scala b/examples/direct/src/main/scala/sage/examples/direct/PubSubExample.scala new file mode 100644 index 00000000..03af8643 --- /dev/null +++ b/examples/direct/src/main/scala/sage/examples/direct/PubSubExample.scala @@ -0,0 +1,30 @@ +package sage.examples.direct + +import java.util.concurrent.{LinkedBlockingQueue, TimeUnit} + +import scala.util.Using + +import sage.* +import sage.backend.* + +/** + * Classic channel pub/sub. `subscribe` returns a blocking iterator, and closing it unsubscribes. `onMessage` calls a function for each message + * on a background virtual thread until the returned handle is closed. The cluster example covers sharded pub/sub. + */ +object PubSubExample { + + def run(client: SageClient): Unit = { + // the subscription is confirmed when subscribe returns, so the messages published next are delivered + val messages = Using.resource(client.subscribe[String]("news")) { news => + (1 to 3).foreach(i => client.publish("news", s"item-$i")) + news.take(3).map(_.payload).toList + } + println(s"received=$messages") + + val alerts = new LinkedBlockingQueue[String]() + Using.resource(client.onMessage[String]("alerts")(message => alerts.put(message.payload))) { _ => + client.publish("alerts", "disk almost full") + println(s"alert=${alerts.poll(5, TimeUnit.SECONDS)}") + } + } +} diff --git a/examples/direct/src/main/scala/sage/examples/direct/RateLimiterExample.scala b/examples/direct/src/main/scala/sage/examples/direct/RateLimiterExample.scala new file mode 100644 index 00000000..1283b96f --- /dev/null +++ b/examples/direct/src/main/scala/sage/examples/direct/RateLimiterExample.scala @@ -0,0 +1,20 @@ +package sage.examples.direct + +import sage.* +import sage.backend.* + +/** + * A distributed token-bucket rate limiter: `rateLimiter` binds a policy, each `tryAcquire` is one atomic allow/deny check on server-side state. + */ +object RateLimiterExample { + + def run(client: SageClient): Unit = { + val limiter = client.rateLimiter[String](RateLimit.perMinute(2)) + val first = limiter.tryAcquire("user:42") + val second = limiter.tryAcquire("user:42") + val third = limiter.tryAcquire("user:42") + println(s"1st allowed=${first.isAllowed} remaining=${first.remainingTokens}") + println(s"2nd allowed=${second.isAllowed} remaining=${second.remainingTokens}") + println(s"3rd allowed=${third.isAllowed}") + } +} diff --git a/examples/direct/src/main/scala/sage/examples/direct/StreamsExample.scala b/examples/direct/src/main/scala/sage/examples/direct/StreamsExample.scala new file mode 100644 index 00000000..92de384f --- /dev/null +++ b/examples/direct/src/main/scala/sage/examples/direct/StreamsExample.scala @@ -0,0 +1,28 @@ +package sage.examples.direct + +import sage.* +import sage.backend.* + +/** + * Redis Streams: append entries, read them back by range, then consume them through a Consumer Group with explicit acknowledgement. The + * stream is reset first so the example is deterministic on re-run. + */ +object StreamsExample { + + def run(client: SageClient): Unit = { + client.del("stream:orders") + client.xAdd("stream:orders")(("item", "book"), ("qty", "2")) + client.xAdd("stream:orders")(("item", "pen"), ("qty", "5")) + val len = client.xLen("stream:orders") + + val entries = client.xRange[String, String]("stream:orders") + + // a Consumer Group reading from the start of the stream + client.xGroupCreate("stream:orders", "workers", id = GroupStartId.At(StreamId.Zero)) + val batches = client.xReadGroup[String, String]("workers", "w1")(("stream:orders", GroupReadId.New))() + val ids = batches.flatMap(_._2).map(_.id) + client.xAck("stream:orders", "workers")(ids.head, ids.tail*) + + println(s"len=$len read=${entries.size} acked=${ids.size}") + } +} diff --git a/examples/direct/src/main/scala/sage/examples/direct/Tour.scala b/examples/direct/src/main/scala/sage/examples/direct/Tour.scala new file mode 100644 index 00000000..4fd6e93b --- /dev/null +++ b/examples/direct/src/main/scala/sage/examples/direct/Tour.scala @@ -0,0 +1,22 @@ +package sage.examples.direct + +import sage.* +import sage.backend.* + +/** + * Runnable tour of the direct client. It opens the client with `use`, which closes it when the block returns, and shares that client across + * the examples. Start a server on localhost:6379 as described in examples/README.md, then run `sbt examplesDirect/run`. + */ +@main def tour(): Unit = { + val config = SageConfig(topology = Topology.Standalone(Endpoint("localhost", 6379))) + SageClient.use(config) { client => + CommandsExample.run(client) + PipelinesExample.run(client) + TransactionsExample.run(client) + PubSubExample.run(client) + CachedReadsExample.run(client) + RateLimiterExample.run(client) + LockExample.run(client) + StreamsExample.run(client) + } +} diff --git a/examples/direct/src/main/scala/sage/examples/direct/TransactionsExample.scala b/examples/direct/src/main/scala/sage/examples/direct/TransactionsExample.scala new file mode 100644 index 00000000..4028b947 --- /dev/null +++ b/examples/direct/src/main/scala/sage/examples/direct/TransactionsExample.scala @@ -0,0 +1,21 @@ +package sage.examples.direct + +import sage.* +import sage.backend.* + +/** + * Runs a WATCH-guarded MULTI/EXEC transaction on one leased dedicated connection. It reads inside the transaction scope, decides what to do, + * then executes a Pipeline atomically. If a watched key changes before EXEC, the transaction returns `None` and the caller should retry. + */ +object TransactionsExample { + + def run(client: SageClient): Unit = { + client.set("tx:n", 1) + val result = client.transaction { tx => + tx.watch("tx:n") + tx.get[Int]("tx:n") + tx.exec((Commands.incr("tx:n"), Commands.incrBy("tx:n", 4))) + } + println(s"transaction result=$result") + } +} diff --git a/integration-tests/direct/src/test/scala/sage/integration/DirectSmokeSuite.scala b/integration-tests/direct/src/test/scala/sage/integration/DirectSmokeSuite.scala new file mode 100644 index 00000000..f866a2a6 --- /dev/null +++ b/integration-tests/direct/src/test/scala/sage/integration/DirectSmokeSuite.scala @@ -0,0 +1,261 @@ +package sage.integration + +import java.util.concurrent.{Executors, TimeUnit} +import java.util.concurrent.atomic.AtomicBoolean + +import scala.concurrent.duration.* +import scala.util.Using + +import sage.* +import sage.backend.* + +class DirectSmokeSuite extends ServerSuite(Images.redis) { + + private def withNativeClient(body: SageClient => Unit): Unit = + withContainers(server => SageClient.use(configOf(server))(body)) + + test("use keeps the caller's interrupt status when the body returns with it set") { + withContainers { server => + SageClient.use(configOf(server)) { client => + assertEquals(client.ping(), "PONG") + Thread.currentThread().interrupt() + } + assert(Thread.interrupted(), "closing the client should not clear the interrupt status") + } + } + + test("close releases the connections even when the caller's interrupt status is already set") { + withContainers { server => + SageClient.use(configOf(server)) { observer => + val client = SageClient.connect(configOf(server).copy(clientName = Some("interrupted-close"))) + assertEquals(client.ping(), "PONG") + Thread.currentThread().interrupt() + client.close + assert(Thread.interrupted(), "close should keep the interrupt status") + val deadline = System.nanoTime() + 5.seconds.toNanos + while (observer.clientList.contains("name=interrupted-close") && System.nanoTime() < deadline) Thread.sleep(50) + assert(!observer.clientList.contains("name=interrupted-close"), "the server should no longer see the closed client's connection") + } + } + } + + test("a distributed lock scopes blocking code and skips contended bodies") { + withNativeClient { client => + val locks = client.lock[String]() + var evaluated = false + val busy = locks.withLock("native-lock", 2.seconds) { + locks.tryWithLock("native-lock") { + evaluated = true + client.ping() + } + } + val acquired = locks.tryWithLock("native-lock")(client.ping()) + assertEquals(busy, None) + assertEquals(acquired, Some("PONG")) + assert(!evaluated) + } + } + + test("a lock body runs on the calling thread and sees its thread-local values") { + withNativeClient { client => + val context = new ThreadLocal[String] + context.set("request-42") + try { + val caller = Thread.currentThread() + val (thread, value) = client.lock[String]().withLock("thread-lock", 2.seconds)((Thread.currentThread(), context.get())) + assert(thread eq caller) + assertEquals(value, "request-42") + } finally context.remove() + } + } + + test("an end user connects and round-trips from many virtual threads") { + withNativeClient { client => + assertEquals(client.ping(), "PONG") + Using.resource(Executors.newVirtualThreadPerTaskExecutor()) { executor => + val futures = (1 to 50).toList.map { i => + executor.submit { () => + val _ = client.set(s"key-$i", s"value-$i") + client.get[String](s"key-$i") + } + } + assertEquals(futures.map(_.get(10, TimeUnit.SECONDS)), (1 to 50).toList.map(i => Some(s"value-$i"))) + } + } + } + + test("interrupting a thread blocked on a command cancels the wait") { + withNativeClient { client => + var failure: Throwable = null + val blocked = Thread.ofVirtual().start { () => + try client.blPop[String]("never-filled")(BlockTimeout.After(5.seconds)): Unit + catch { case t: Throwable => failure = t } + } + Thread.sleep(200) + blocked.interrupt() + blocked.join(2000) + assert(!blocked.isAlive, "the interrupted thread should stop waiting") + assert(failure.isInstanceOf[InterruptedException], String.valueOf(failure)) + assertEquals(client.ping(), "PONG") + } + } + + test("a pipeline returns a typed tuple, surfacing failures per position") { + withNativeClient { client => + val _ = client.set("pipe:a", "x") + val _ = client.set("pipe:n", 10) + val out = client.pipeline((Commands.get[String, String]("pipe:a"), Commands.incrBy[String]("pipe:n", 5))) + assertEquals(out, (Some("x"), 15L)) + val _ = client.set("pipe:str", "hello") + val attempt = client.pipelineAttempt((Commands.get[String, String]("pipe:str"), Commands.incr[String]("pipe:str"))) + assert(attempt._1 == Right(Some("hello")), attempt._1) + assert(attempt._2.isLeft, attempt._2) + } + } + + test("a transaction commits atomically, guarded by WATCH") { + withNativeClient { client => + val _ = client.set("tx:n", 1) + val out = client.transaction { tx => + tx.watch("tx:n") + val _ = tx.get[Int]("tx:n") + tx.exec((Commands.incr[String]("tx:n"), Commands.incrBy[String]("tx:n", 4))) + } + assertEquals(out, Some((2L, 6L))) + } + } + + test("scanAll iterates over every key") { + withNativeClient { client => + (1 to 50).foreach(i => client.set(s"scan-$i", "v")) + val keys = client.scanAll(pattern = Some("scan-*"), count = Some(10L)).toSet + assertEquals(keys, (1 to 50).map(i => s"scan-$i").toSet) + } + } + + test("subscribe delivers published messages through a blocking iterator") { + withNativeClient { client => + Using.resource(client.subscribe[String]("smoke")) { messages => + (1 to 3).foreach(i => client.publish("smoke", s"m$i")) + val received = messages.take(3).toList + assertEquals(received.map(_.channel).toSet, Set("smoke")) + assertEquals(received.map(_.payload), List("m1", "m2", "m3")) + } + } + } + + test("closing a subscription ends its iterator") { + withNativeClient { client => + val messages = client.subscribe[String]("closing") + val _ = client.publish("closing", "a") + assertEquals(messages.next().payload, "a") + messages.close() + assert(!messages.hasNext) + } + } + + test("onMessage delivers each message to the callback until it is closed") { + withNativeClient { client => + val received = new java.util.concurrent.LinkedBlockingQueue[String]() + val subscription = client.onMessage[String]("callback")(m => received.put(m.payload)) + (1 to 3).foreach(i => client.publish("callback", s"m$i")) + assertEquals((1 to 3).map(_ => received.poll(5, TimeUnit.SECONDS)).toList, List("m1", "m2", "m3")) + // the delivery thread is waiting for the next message, so close must wake it rather than wait for traffic + val started = System.nanoTime() + subscription.close() + assert((System.nanoTime() - started).nanos < 2.seconds, "close should not wait for another message") + assertEquals(client.publish("callback", "after"), 0L) + subscription.close() + } + } + + test("a failing onMessage callback does not stop later deliveries") { + withNativeClient { client => + val received = new java.util.concurrent.LinkedBlockingQueue[String]() + val subscription = client.onMessage[String]("failing") { m => + if (m.payload == "bad") throw new RuntimeException("expected test failure") + received.put(m.payload) + } + try { + val _ = client.publish("failing", "bad") + val _ = client.publish("failing", "good") + assertEquals(received.poll(5, TimeUnit.SECONDS), "good") + } finally subscription.close() + } + } + + test("an onMessage callback does not inherit the subscriber's inheritable thread-local values") { + withNativeClient { client => + val context = new InheritableThreadLocal[String] + val seen = new java.util.concurrent.LinkedBlockingQueue[String]() + context.set("request-42") + val subscription = + try client.onMessage[String]("context")(_ => seen.put(String.valueOf(context.get()))) + finally context.remove() + try { + val _ = client.publish("context", "x") + assertEquals(seen.poll(5, TimeUnit.SECONDS), "null") + } finally subscription.close() + } + } + + test("onPatternMessage and onShardMessage deliver pattern and shard messages") { + withNativeClient { client => + val patterns = new java.util.concurrent.LinkedBlockingQueue[String]() + val shards = new java.util.concurrent.LinkedBlockingQueue[String]() + Using.resources( + client.onPatternMessage[String]("news.*")(m => patterns.put(s"${m.pattern} ${m.channel} ${m.payload}")), + client.onShardMessage[String]("shard")(m => shards.put(m.payload)) + ) { (_, _) => + val _ = client.publish("news.sport", "goal") + val _ = client.sPublish("shard", "s1") + assertEquals(patterns.poll(5, TimeUnit.SECONDS), "news.* news.sport goal") + assertEquals(shards.poll(5, TimeUnit.SECONDS), "s1") + } + } + } + + test("hScanAll, sScanAll, and zScanAll iterate over every element") { + withNativeClient { client => + (1 to 50).foreach { i => + val _ = client.hSet("hscan", (s"f$i", s"v$i")) + val _ = client.sAdd("sscan", s"m$i") + client.zAdd("zscan")((s"m$i", i.toDouble)) + } + assertEquals(client.hScanAll[String, String]("hscan", count = Some(10L)).toMap, (1 to 50).map(i => s"f$i" -> s"v$i").toMap) + assertEquals(client.sScanAll[String]("sscan", count = Some(10L)).toSet, (1 to 50).map(i => s"m$i").toSet) + assertEquals(client.zScanAll[String]("zscan", count = Some(10L)).toMap, (1 to 50).map(i => s"m$i" -> i.toDouble).toMap) + } + } + + test("xConsume acknowledges entries and stops when its thread is interrupted") { + withNativeClient { client => + client.xGroupCreate("events", "workers", mkStream = true) + (1 to 3).foreach(i => client.xAdd("events")(("n", i.toString))) + val _ = client.xAdd("events")(("n", "4")) + val seen = new java.util.concurrent.LinkedBlockingQueue[String]() + val stopped = new AtomicBoolean(false) + val consumer = Thread.ofVirtual().start { () => + try + client.xConsume[String, String]("workers", "w1", "events", block = BlockTimeout.After(100.millis))(e => seen.put(e.fields.toMap.apply("n"))) + catch { case _: InterruptedException => stopped.set(true) } + } + val values = (1 to 4).map(_ => seen.poll(5, TimeUnit.SECONDS)) + consumer.interrupt() + consumer.join(2000) + assertEquals(values.toList, List("1", "2", "3", "4")) + assert(stopped.get()) + } + } + + test("client.rateLimiter admits up to capacity then denies") { + withNativeClient { client => + val rl = client.rateLimiter[String](RateLimit(capacity = 2, refillTokens = 1, refillPeriod = 1.hour)) + val first = rl.tryAcquire("smoke") + val second = rl.tryAcquire("smoke") + val denied = rl.tryAcquire("smoke") + assert(first.isAllowed && second.isAllowed, "the first two are admitted") + assert(!denied.isAllowed, "the third is denied once the bucket empties") + } + } +} diff --git a/sage-client/direct/src/main/scala/sage/backend/SageClient.scala b/sage-client/direct/src/main/scala/sage/backend/SageClient.scala new file mode 100644 index 00000000..1a633aff --- /dev/null +++ b/sage-client/direct/src/main/scala/sage/backend/SageClient.scala @@ -0,0 +1,268 @@ +package sage.backend + +import scala.annotation.unused +import scala.concurrent.duration.FiniteDuration +import scala.util.control.NonFatal + +import kyo.compat.* + +import sage.{Message, PatternMessage} +import sage.client.SageConfig +import sage.client.internal.{Client, LoweredClient, Paged, Subscription} +import sage.codec.{KeyCodec, ValueCodec} +import sage.commands.* + +/** + * A Sage client for plain blocking code. Each operation blocks the calling thread until the reply arrives and returns the result directly. + * Failures are thrown as exceptions. Interrupting the calling thread cancels the wait, except in `close`, which always finishes. Call it from virtual threads, which makes blocking + * cheap, for example from an HTTP handler in a framework that runs each request on a virtual thread. + */ +type SageClient = Client[[A] =>> A, String] + +extension [K](client: Client[[A] =>> A, K])(using @unused ev: KeyCodec[K]) { + + /** + * Scans the full keyspace. The returned iterator fetches pages as it is consumed, so each call to `hasNext` may send a command. An empty + * page does not end the scan; iteration stops when the server returns a zero cursor. Redis may return the same key more than once. In + * cluster mode, each master is scanned with its own cursor. + */ + def scanAll( + pattern: Option[String] = None, + count: Option[Long] = None, + ofType: Option[RedisType] = None + ): Iterator[K] = + paged(Paged.scanAll[K](client.runner, pattern, count, ofType)) + + /** + * Iterates over all HSCAN field/value pairs until the server returns a zero cursor. An empty page with a non-zero cursor continues the scan. + */ + def hScanAll[F: KeyCodec, V: ValueCodec]( + key: K, + pattern: Option[String] = None, + count: Option[Long] = None + ): Iterator[(F, V)] = + paged(Paged.scanKey(client.runner)(cursor => Hashes.hScan[K, F, V](key, cursor, pattern, count))) + + /** + * Iterates over all SSCAN members until the server returns a zero cursor. An empty page with a non-zero cursor continues the scan. + */ + def sScanAll[V: ValueCodec]( + key: K, + pattern: Option[String] = None, + count: Option[Long] = None + ): Iterator[V] = + paged(Paged.scanKey(client.runner)(cursor => Sets.sScan[K, V](key, cursor, pattern, count))) + + /** + * Iterates over all ZSCAN member/score pairs until the server returns a zero cursor. An empty page with a non-zero cursor continues the scan. + */ + def zScanAll[V: ValueCodec]( + key: K, + pattern: Option[String] = None, + count: Option[Long] = None + ): Iterator[(V, Double)] = + paged(Paged.scanKey(client.runner)(cursor => SortedSets.zScan[K, V](key, cursor, pattern, count))) + + // convert pages from the shared Paged helper into individual iterator elements + private def paged[S, A](pages: Paged.Pages[S, A]): Iterator[A] = + CStream.unfold[S, Vector[A]](pages.init)(pages.step).flatMap(items => CStream.init(items)).lower + + /** + * Lazily pages an entire stream by range, batching `XRANGE` and advancing past the last id each page. Stops when a page comes back empty. + */ + def xRangeAll[F: KeyCodec, V: ValueCodec]( + key: K, + start: StreamRangeId = StreamRangeId.Min, + end: StreamRangeId = StreamRangeId.Max, + batch: Long = 100L + ): Iterator[StreamEntry[F, V]] = + paged(Paged.xRangeAll[K, F, V](client.runner, key, start, end, batch)) + + /** + * Auto-claims idle pending entries for `consumer`, advancing the `XAUTOCLAIM` cursor until it returns to the start. Entries whose data + * has already been deleted are skipped. + */ + def xAutoClaimAll[F: KeyCodec, V: ValueCodec]( + key: K, + group: String, + consumer: String, + minIdle: FiniteDuration, + start: StreamId = StreamId.Zero, + count: Option[Long] = None + ): Iterator[StreamEntry[F, V]] = + paged(Paged.xAutoClaimAll[K, F, V](client.runner, key, group, consumer, minIdle, start, count)) + + /** + * Follows a stream without a consumer group. It first reads every entry after `from`, then waits for new entries, so the iterator does + * not end on its own. The explicit entry ID used for each blocking read avoids missing entries that arrive between reads. Unlike + * [[xConsume]], this method does not acknowledge entries. `from` defaults to the start of the stream. + */ + def xTail[F: KeyCodec, V: ValueCodec]( + key: K, + from: StreamId = StreamId.Zero, + count: Option[Long] = None, + block: BlockTimeout = Paged.defaultPoll + ): Iterator[StreamEntry[F, V]] = + paged(Paged.xTail[K, F, V](client.runner, key, from, count, block)) + + /** + * Follows a stream as part of a consumer group and blocks the calling thread until it is interrupted or `handle` throws. It processes this + * consumer's pending entries first, then waits for new entries. Each entry is acknowledged only after `handle` succeeds. If `handle` + * fails, the entry remains pending and can be recovered later. + */ + def xConsume[F: KeyCodec, V: ValueCodec]( + group: String, + consumer: String, + key: K, + count: Option[Long] = None, + block: BlockTimeout = Paged.defaultPoll + )(handle: StreamEntry[F, V] => Unit): Unit = + paged(Paged.xConsume[K, F, V](client.runner, group, consumer, key, count, block)).foreach { entry => + handle(entry) + client.run(Streams.xAck(key, group)(entry.id)): Unit + } + + /** + * Subscribes to one or more channels and returns an iterator over the received messages. The subscription is open when this method + * returns: standalone and master-replica clients wait for server confirmation, and cluster clients wait up to the connection timeout and + * may return before confirmation. `hasNext` blocks until a message arrives. Close the iterator to unsubscribe, for example with + * `scala.util.Using`. Sage resubscribes after reconnecting, but messages published while the connection is down are lost. + */ + def subscribe[V: ValueCodec](channel: String, rest: String*): Messages[Message[V]] = + new Messages(client.subscribeChannels[V](channel, rest*)) + + /** + * Like [[subscribe]], but subscribes to glob patterns. Each delivery names the matching pattern and the concrete channel. + */ + def pSubscribe[V: ValueCodec](pattern: String, rest: String*): Messages[PatternMessage[V]] = + new Messages(client.subscribePatterns[V](pattern, rest*)) + + /** + * Like [[subscribe]], but subscribes to shard channels. In a cluster, each subscription follows its slot to the current owning node after a + * migration or failover. Sharded deliveries use the ordinary [[Message]] type. + */ + def sSubscribe[V: ValueCodec](channel: String, rest: String*): Messages[Message[V]] = + new Messages(client.subscribeShardChannels[V](channel, rest*)) + + /** + * Subscribes to one or more channels and calls `handle` for each message, in order, on a virtual thread that Sage starts. Delivery + * continues until the returned handle is closed or the client is closed. The subscription is open when this method returns, as with + * [[subscribe]]. Use it to forward messages to code that runs in the background, for example `client.onMessage[String]("events")(publish)`. + * + * If `handle` throws, the failure is passed to the delivery thread's uncaught exception handler and delivery continues with the next + * message. A payload that cannot be decoded ends the subscription and is reported the same way. The delivery thread does not inherit the + * caller's inheritable thread-local values, so request or transaction context does not leak into it. + * + * Closing the returned handle unsubscribes and waits until the delivery thread has finished, so `handle` is not called after `close` + * returns. A call to `handle` that is in progress runs to completion first. Closing more than once, or after the client was closed, has no + * effect. Closing from inside `handle` unsubscribes without waiting. + */ + def onMessage[V: ValueCodec](channel: String, rest: String*)(handle: Message[V] => Unit): AutoCloseable = + deliver(subscribe[V](channel, rest*))(handle) + + /** + * Like [[onMessage]], but subscribes to glob patterns. Each delivery names the matching pattern and the concrete channel. + */ + def onPatternMessage[V: ValueCodec](pattern: String, rest: String*)(handle: PatternMessage[V] => Unit): AutoCloseable = + deliver(pSubscribe[V](pattern, rest*))(handle) + + /** + * Like [[onMessage]], but subscribes to shard channels, as [[sSubscribe]] does. + */ + def onShardMessage[V: ValueCodec](channel: String, rest: String*)(handle: Message[V] => Unit): AutoCloseable = + deliver(sSubscribe[V](channel, rest*))(handle) + + // Consumes `messages` on a virtual thread and calls `handle` for each message. Closing the returned handle closes `messages`. + private def deliver[A](messages: Messages[A])(handle: A => Unit): AutoCloseable = { + val worker = Thread + .ofVirtual() + .name("sage-subscription") + .inheritInheritableThreadLocals(false) + .unstarted { () => + try + messages.foreach { message => + try handle(message) + catch { + case NonFatal(e) => + val thread = Thread.currentThread() + thread.getUncaughtExceptionHandler.uncaughtException(thread, e) + } + } + finally messages.close() + } + worker.start() + () => { + messages.close() + // Keep waiting after an interrupt so that `handle` is never called after `close` returns, then restore the interrupt status. + if (Thread.currentThread() ne worker) { + var interrupted = false + while (worker.isAlive) + try worker.join() + catch { case _: InterruptedException => interrupted = true } + if (interrupted) Thread.currentThread().interrupt() + } + } + } +} + +/** + * An open subscription read as a blocking iterator. `hasNext` waits for the next message and returns `false` once the subscription has + * ended, either because it was closed or because the client was closed. `close` unsubscribes and can be called more than once. + */ +final class Messages[A] private[backend] (subscription: Subscription[[X] =>> X, A]) extends Iterator[A] with AutoCloseable { + + private var pending: Option[A] = None + private var ended = false + private val closed = new java.util.concurrent.atomic.AtomicBoolean(false) + + def hasNext: Boolean = { + if (pending.isEmpty && !ended) + subscription.next match { + case some @ Some(_) => pending = some + case None => ended = true + } + pending.isDefined + } + + def next(): A = + (if (hasNext) pending else None) match { + case Some(a) => + pending = None + a + case None => throw new NoSuchElementException("the subscription has ended") + } + + def close(): Unit = if (closed.compareAndSet(false, true)) subscription.close +} + +object SageClient { + + /** + * A client that uses `K` for keys, returned by `client.as[K]`. [[SageClient]] uses `String` keys by default. Calling `as` changes only + * the key type and continues to use the same connection. + */ + type Keyed[K] = Client[[A] =>> A, K] + + /** + * Connects with `config` and returns the client. Call `close` when the client is no longer needed, for example when the application shuts + * down. + */ + def connect(config: SageConfig): SageClient = new Lowered(Client.connect(config).lower) + + /** + * Connects with `config`, passes the client to `body`, and closes the client when `body` returns or throws. Failures while closing are + * ignored so they cannot hide the result of `body`. + */ + def use[A](config: SageConfig)(body: SageClient => A): A = { + val client = connect(config) + try body(client) + finally + try client.close + catch { case NonFatal(_) => () } + } + + final private[sage] class Lowered(underlying: Client[CIO, String]) extends LoweredClient[[X] =>> X](underlying) { + protected def lower[A](c: CIO[A]): A = c.lower + protected def lift[A](fa: A): CIO[A] = CIO.value(fa) + } +} diff --git a/sage-client/direct/src/test/scala/sage/client/internal/DirectBindingSpec.scala b/sage-client/direct/src/test/scala/sage/client/internal/DirectBindingSpec.scala new file mode 100644 index 00000000..2faed179 --- /dev/null +++ b/sage-client/direct/src/test/scala/sage/client/internal/DirectBindingSpec.scala @@ -0,0 +1,96 @@ +package sage.client.internal + +import java.util.concurrent.{CompletionException, TimeoutException} + +import scala.concurrent.Await +import scala.concurrent.duration.* +import scala.util.{Failure, Try} + +import kyo.compat.* + +class DirectBindingSpec extends munit.FunSuite { + override val munitTimeout = 30.seconds + + test("unsafeRun completes its future when the computation throws a fatal throwable") { + val future = CIO.defer[Unit](throw new InterruptedException("fatal")).unsafeRun + assert(Try(Await.ready(future, 2.seconds)).isSuccess, "the future should complete") + assert(future.value.exists(_.isFailure), String.valueOf(future.value)) + } + + test("timeout reports a computation's own TimeoutException as a failure") { + val failure = new TimeoutException("from the computation") + assertEquals(CIO.timeout(5.seconds)(CIO.fail(failure)).liftToTry.lower, Failure(failure)) + } + + test("the first failure in a concurrent collection interrupts every element that started") { + val boom = new RuntimeException("boom") + val outcome = (1 to 200).iterator + .map { _ => + CIO.timeout(2.seconds)(CIO.foreach(0 until 50)(i => if (i == 0) CIO.fail(boom) else CIO.never).liftToTry).lower + } + .find(_ != Some(Failure(boom))) + assertEquals(outcome, None) + } + + test("a recursive flatMap loop of a million steps completes without overflowing the stack") { + def loop(n: Int, sum: Long): CIO[Long] = + if (n == 0) CIO.value(sum) else CIO.defer(n).flatMap(i => loop(n - 1, sum + i)) + assertEquals(loop(1000000, 0L).lower, 500000500000L) + } + + test("a failure after a long flatMap chain has moved to a new thread reaches the caller unchanged") { + val failure = new CompletionException(new RuntimeException("cause")) + def loop(n: Int): CIO[Unit] = if (n == 0) CIO.fail(failure) else CIO.unit.flatMap(_ => loop(n - 1)) + assertEquals(loop(1000).liftToTry.lower, Failure(failure)) + } + + test("a promise and an async callback rethrow the exact failure they were completed with") { + val failure = new CompletionException(new RuntimeException("cause")) + val promise = CPromise.init[Unit].lower + assert(promise.fail(failure).lower) + assertEquals(promise.get.liftToTry.lower, Failure(failure)) + assertEquals(promise.poll.lower, Some(Failure(failure))) + assertEquals(CIO.async[Unit](_(Failure(failure))).liftToTry.lower, Failure(failure)) + } + + test("fromCompletionStage rethrows a failure thrown while creating the stage unchanged") { + val failure = new CompletionException(new RuntimeException("cause")) + assertEquals(CIO.fromCompletionStage[Unit](throw failure).liftToTry.lower, Failure(failure)) + } + + test("a recursive loop through every combinator that runs a nested CIO on the same thread completes without overflowing the stack") { + val local = CLocal.init(0).lower + val meter = CMeter.init(Int.MaxValue).lower + def loop(n: Int): CIO[Int] = + if (n == 0) local.get + else CIO.ensure(CIO.unit)(local.let(n)(meter.run(CIO.unit.flatMap(_ => loop(n - 1))))).recover(CIO.fail) + assertEquals(loop(100000).lower, 1) + } + + test("race runs its first computation on the calling thread at every depth around the flatMap limit") { + val context = new ThreadLocal[String] + context.set("request-42") + // the first computation is itself a chain of 100 steps, as a lock body wrapped by the lock executor would be + val first = (1 to 100).foldLeft(CIO.defer((Thread.currentThread(), context.get())))((acc, _) => CIO.unit.flatMap(_ => acc)) + def loop(n: Int): CIO[(Thread, String)] = if (n == 0) CIO.race(first, CIO.never) else CIO.unit.flatMap(_ => loop(n - 1)) + try + for (n <- List(0, 1, 155, 156, 157, 254, 255, 256, 257, 511, 512, 513, 1000)) { + val (thread, value) = loop(n).lower + assert(thread eq Thread.currentThread(), s"at depth $n the first computation ran on $thread") + assertEquals(value, "request-42", s"at depth $n") + assert(!Thread.interrupted(), s"at depth $n") + } + finally context.remove() + } + + test("nested races inside a moved chain all run on the calling thread") { + val caller = Thread.currentThread() + def deep[A](c: CIO[A]): CIO[A] = (1 to 1000).foldLeft(c)((acc, _) => CIO.unit.flatMap(_ => acc)) + val inner = deep(CIO.race(CIO.defer(Thread.currentThread()), CIO.never)) + val outer = deep(CIO.race(inner.map(t => (t, Thread.currentThread())), CIO.never)) + val (innerThread, outerThread) = outer.lower + assert(innerThread eq caller) + assert(outerThread eq caller) + } + +} diff --git a/sage-client/direct/src/test/scala/sage/client/internal/DirectLockCancellationSpec.scala b/sage-client/direct/src/test/scala/sage/client/internal/DirectLockCancellationSpec.scala new file mode 100644 index 00000000..753e9e71 --- /dev/null +++ b/sage-client/direct/src/test/scala/sage/client/internal/DirectLockCancellationSpec.scala @@ -0,0 +1,36 @@ +package sage.client.internal + +import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger} + +import scala.concurrent.duration.* +import scala.util.control.ControlThrowable + +import kyo.compat.* + +import sage.backend.SageClient + +class DirectLockCancellationSpec extends LockCancellationSpec { + + override protected def tryWithLock[A](commands: SharedRunner, lease: FiniteDuration)(body: CIO[A]): CIO[Option[A]] = + CIO.defer(new SageClient.Lowered(new LockTestClient(commands)).lock[String](lease).tryWithLock("key")(body.lower)) + + override protected def withLock[A](commands: SharedRunner, lease: FiniteDuration, wait: FiniteDuration)(body: CIO[A]): CIO[A] = + CIO.defer(new SageClient.Lowered(new LockTestClient(commands)).lock[String](lease).withLock("key", wait)(body.lower)) + + test("a body control exception ends the scope and releases ownership") { + val released = new AtomicBoolean(false) + val renewals = new AtomicInteger(0) + val failure = new ControlThrowable {} + CIO.defer { + val caught = + try { + tryWithLock(runner(released, renewals, false), 300.millis)(CIO.defer(throw failure)).lower + None + } catch { + case cause: ControlThrowable => Some(cause) + } + assertEquals(caught, Some(failure)) + assert(released.get()) + }.unsafeRun + } +} diff --git a/sage-client/shared/src/test/scala/sage/client/internal/MultiplexedConnectionSpec.scala b/sage-client/shared/src/test/scala/sage/client/internal/MultiplexedConnectionSpec.scala index e583fa25..96316353 100644 --- a/sage-client/shared/src/test/scala/sage/client/internal/MultiplexedConnectionSpec.scala +++ b/sage-client/shared/src/test/scala/sage/client/internal/MultiplexedConnectionSpec.scala @@ -688,6 +688,21 @@ class MultiplexedConnectionSpec extends munit.FunSuite { assertEquals(connection.currentState, MultiplexedConnection.State.Closed) } + test("close on an interrupted thread closes the connection, fails the straggler and keeps the interrupt") { + val (connection, _, transports) = make(autoWrite = false, closeTimeout = 2.seconds) + @volatile var result: Option[Try[String]] = None + connection.submit(Connection.ping(), r => result = Some(r)) + transports.head.writeNext() // written, no reply will arrive + Thread.currentThread().interrupt() + connection.close() + assert(Thread.interrupted(), "close should keep the interrupt status") + assertEquals(result, Some(Failure(ConnectionLost(mayHaveExecuted = true)))) + assertEquals(connection.currentState, MultiplexedConnection.State.Closed) + assertEquals(transports.head.closeCount, 1) + connection.close() + assertEquals(transports.head.closeCount, 1) + } + private def cachedConnection( events: Events = Events.disabled ): (MultiplexedConnection, ManualScheduler, mutable.ArrayBuffer[FakeTransport]) = { diff --git a/sage-client/shared/src/test/scala/sage/client/internal/SocketTransportSpec.scala b/sage-client/shared/src/test/scala/sage/client/internal/SocketTransportSpec.scala index c2afd256..8d90e1ce 100644 --- a/sage-client/shared/src/test/scala/sage/client/internal/SocketTransportSpec.scala +++ b/sage-client/shared/src/test/scala/sage/client/internal/SocketTransportSpec.scala @@ -100,6 +100,41 @@ class SocketTransportSpec extends munit.FunSuite { } } + test("close keeps waiting for a busy reader while the closing thread is interrupted repeatedly") { + @volatile var closedCount = 0 + val inFrame = new CountDownLatch(1) + val release = new Semaphore(0) + // the reader stays in onFrame, ignoring interrupts, until the test releases it + val onFrame: Frame => Unit = _ => { + inFrame.countDown() + release.acquireUninterruptibly() + } + withTransport(onClosed = () => closedCount += 1, onFrame = onFrame) { (transport, peer) => + peer.getOutputStream.write("+PONG\r\n".getBytes(StandardCharsets.UTF_8)) + peer.getOutputStream.flush() + assert(inFrame.await(5, TimeUnit.SECONDS), "the reader should reach onFrame") + @volatile var interruptedAfter = false + val closer = Thread.ofVirtual().start { () => + transport.close() + interruptedAfter = Thread.currentThread().isInterrupted + } + (1 to 5).foreach { _ => + closer.interrupt() + Thread.sleep(20) + } + assert(closer.isAlive, "close should still wait for the reader") + assertEquals(closedCount, 0) + release.release() + closer.join(5000) + assert(!closer.isAlive) + assert(interruptedAfter, "close should set the interrupt status again") + assertEquals(closedCount, 1) + assert(!transport.reader.isAlive) + assert(!transport.writer.isAlive) + assertEquals(peer.getInputStream.read(), -1, "the socket should be closed") + } + } + test("close on an interrupted thread still waits for the I/O threads, drops queued items and runs onClosed") { @volatile var closedCount = 0 val inWrite = new CountDownLatch(1) diff --git a/sage-compat-direct/README.md b/sage-compat-direct/README.md new file mode 100644 index 00000000..e6636480 --- /dev/null +++ b/sage-compat-direct/README.md @@ -0,0 +1,24 @@ +# sage-compat-direct + +This module implements the [kyo-compat](https://github.com/getkyo/kyo/tree/main/kyo-compat) API for plain blocking code. It depends only on the Scala library and JDK 21. The `-direct` cells of `sage-client` and downstream modules compile and run against it, so `sage-client-direct` returns results directly instead of wrapping them in an effect type. + +## How it works + +A `CIO[A]` is a function from the current `flatMap` nesting depth to the result, the same carrier as Kyo's Ox binding without the Ox capability. Running it with `lower` calls the function at depth 0 on the calling thread. `flatMap` passes the depth on and calls its functions directly. Once the depth reaches 256, the rest of the chain runs on a fresh virtual thread that the caller waits for, so long `flatMap` chains and recursive loops do not overflow the stack. That thread inherits the caller's `CLocal` values, its failures reach the caller unchanged, and interrupting the caller interrupts it and waits for it to exit. Failures are thrown exceptions and cancellation is thread interruption. + +`CIO.async` blocks the calling thread until the callback fires. This is cheap on a virtual thread, which is how the binding is meant to be used. + +Operations that need concurrency start JDK virtual threads and wait for every thread they start before they return: + +- `timeout` runs the computation on a virtual thread and interrupts it at the deadline. +- `race` runs its first argument on the calling thread and its second on a virtual thread. Sage's distributed locks pass the user's body as the first argument, so the body keeps the caller's thread-local context. This also holds when a long `flatMap` chain has already moved to a fresh virtual thread: the thread that called `lower` waits for that chain and runs the race when the chain reaches it. If the second argument wins, the caller is interrupted to stop the first, and that interrupt is cleared before `race` returns. +- `acquireReleaseWith` runs `release` on a separate virtual thread and waits for it without reacting to interrupts, so a pending interrupt cannot cut the release short. +- `zip`, `foreach`, and the other collection operations run each element on its own virtual thread. + +`CFiber.init` starts a virtual thread that is not tied to the computation that started it. `CLocal` uses `InheritableThreadLocal`, so virtual threads started by the binding see the values that were current when they started. + +The atomic types, `CPromise`, `CChannel`, `CLatch`, `CMeter`, and `CChunk` use the same JDK and Scala types as Kyo's Ox binding. + +## Conformance + +`kyo-compat-plugin` contains a conformance suite shared by every binding. The `directConformance` matrix runs it against this module, and CI includes it in `testUnit`. Run `sbt conformanceDirect` to execute only this suite. diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CAtomicBoolean.scala b/sage-compat-direct/src/main/scala/kyo/compat/CAtomicBoolean.scala new file mode 100644 index 00000000..22776135 --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CAtomicBoolean.scala @@ -0,0 +1,53 @@ +package kyo.compat + +import java.util.concurrent.atomic.AtomicBoolean + +/** + * Uses `java.util.concurrent.atomic.AtomicBoolean`. `lift` and `lower` return the existing atomic cell. + */ +opaque type CAtomicBoolean = AtomicBoolean + +object CAtomicBoolean { + + /** + * Allocates a fresh atomic boolean initialized to `v`. + */ + inline def init(inline v: Boolean): CIO[CAtomicBoolean] = + CIO.defer(new AtomicBoolean(v)) + + /** + * Wraps a native `AtomicBoolean` as a `CAtomicBoolean`. The conversion is the identity on the carrier. + */ + inline def lift(inline u: AtomicBoolean): CAtomicBoolean = u + + extension (inline self: CAtomicBoolean) { + + /** + * Unwraps to the native `AtomicBoolean`. The conversion is the identity on the carrier. + */ + inline def lower: AtomicBoolean = self + + /** + * Reads the current value. + */ + inline def get: CIO[Boolean] = CIO.defer(self.get()) + + /** + * Atomically sets the value to `v`. + */ + inline def set(inline v: Boolean): CIO[Unit] = CIO.defer(self.set(v)) + + /** + * Atomically sets the value to `v` and returns the previous value. + */ + inline def getAndSet(inline v: Boolean): CIO[Boolean] = CIO.defer(self.getAndSet(v)) + + /** + * Atomic compare-and-set: replaces `expected` with `updated` if the current value equals `expected`. + */ + inline def compareAndSet(inline expected: Boolean, inline updated: Boolean): CIO[Boolean] = + CIO.defer(self.compareAndSet(expected, updated)) + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CAtomicInt.scala b/sage-compat-direct/src/main/scala/kyo/compat/CAtomicInt.scala new file mode 100644 index 00000000..d3bd0501 --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CAtomicInt.scala @@ -0,0 +1,83 @@ +package kyo.compat + +import java.util.concurrent.atomic.AtomicInteger + +/** + * Uses `java.util.concurrent.atomic.AtomicInteger`. `lift` and `lower` return the existing atomic cell. + */ +opaque type CAtomicInt = AtomicInteger + +object CAtomicInt { + + /** + * Allocates a fresh atomic cell initialized to `v`. + */ + inline def init(inline v: Int): CIO[CAtomicInt] = + CIO.defer(new AtomicInteger(v)) + + /** + * Wraps a native `AtomicInteger` as a `CAtomicInt`. The conversion is the identity on the carrier. + */ + inline def lift(inline u: AtomicInteger): CAtomicInt = u + + extension (inline self: CAtomicInt) { + + /** + * Unwraps to the native `AtomicInteger`. The conversion is the identity on the carrier. + */ + inline def lower: AtomicInteger = self + + /** + * Reads the current value. + */ + inline def get: CIO[Int] = CIO.defer(self.get()) + + /** + * Atomically sets the value to `v`. + */ + inline def set(inline v: Int): CIO[Unit] = CIO.defer(self.set(v)) + + /** + * Atomically sets the value to `v` and returns the previous value. + */ + inline def getAndSet(inline v: Int): CIO[Int] = CIO.defer(self.getAndSet(v)) + + /** + * Atomically increments by 1 and returns the new value. + */ + inline def incrementAndGet: CIO[Int] = CIO.defer(self.incrementAndGet()) + + /** + * Atomically increments by 1 and returns the previous value. + */ + inline def getAndIncrement: CIO[Int] = CIO.defer(self.getAndIncrement()) + + /** + * Atomically decrements by 1 and returns the new value. + */ + inline def decrementAndGet: CIO[Int] = CIO.defer(self.decrementAndGet()) + + /** + * Atomically decrements by 1 and returns the previous value. + */ + inline def getAndDecrement: CIO[Int] = CIO.defer(self.getAndDecrement()) + + /** + * Atomically adds `delta` and returns the new value. + */ + inline def addAndGet(inline delta: Int): CIO[Int] = CIO.defer(self.addAndGet(delta)) + + /** + * Atomically adds `delta` and returns the previous value. + */ + inline def getAndAdd(inline delta: Int): CIO[Int] = CIO.defer(self.getAndAdd(delta)) + + /** + * Atomic compare-and-set: replaces `expected` with `updated` if the current value equals `expected`. + */ + inline def compareAndSet(inline expected: Int, inline updated: Int): CIO[Boolean] = + CIO.defer(self.compareAndSet(expected, updated)) + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CAtomicLong.scala b/sage-compat-direct/src/main/scala/kyo/compat/CAtomicLong.scala new file mode 100644 index 00000000..c8c1bb76 --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CAtomicLong.scala @@ -0,0 +1,83 @@ +package kyo.compat + +import java.util.concurrent.atomic.AtomicLong + +/** + * Uses `java.util.concurrent.atomic.AtomicLong`. `lift` and `lower` return the existing atomic cell. + */ +opaque type CAtomicLong = AtomicLong + +object CAtomicLong { + + /** + * Allocates a fresh atomic cell initialized to `v`. + */ + inline def init(inline v: Long): CIO[CAtomicLong] = + CIO.defer(new AtomicLong(v)) + + /** + * Wraps a native `AtomicLong` as a `CAtomicLong`. The conversion is the identity on the carrier. + */ + inline def lift(inline u: AtomicLong): CAtomicLong = u + + extension (inline self: CAtomicLong) { + + /** + * Unwraps to the native `AtomicLong`. The conversion is the identity on the carrier. + */ + inline def lower: AtomicLong = self + + /** + * Reads the current value. + */ + inline def get: CIO[Long] = CIO.defer(self.get()) + + /** + * Atomically sets the value to `v`. + */ + inline def set(inline v: Long): CIO[Unit] = CIO.defer(self.set(v)) + + /** + * Atomically sets the value to `v` and returns the previous value. + */ + inline def getAndSet(inline v: Long): CIO[Long] = CIO.defer(self.getAndSet(v)) + + /** + * Atomically increments by 1 and returns the new value. + */ + inline def incrementAndGet: CIO[Long] = CIO.defer(self.incrementAndGet()) + + /** + * Atomically increments by 1 and returns the previous value. + */ + inline def getAndIncrement: CIO[Long] = CIO.defer(self.getAndIncrement()) + + /** + * Atomically decrements by 1 and returns the new value. + */ + inline def decrementAndGet: CIO[Long] = CIO.defer(self.decrementAndGet()) + + /** + * Atomically decrements by 1 and returns the previous value. + */ + inline def getAndDecrement: CIO[Long] = CIO.defer(self.getAndDecrement()) + + /** + * Atomically adds `delta` and returns the new value. + */ + inline def addAndGet(inline delta: Long): CIO[Long] = CIO.defer(self.addAndGet(delta)) + + /** + * Atomically adds `delta` and returns the previous value. + */ + inline def getAndAdd(inline delta: Long): CIO[Long] = CIO.defer(self.getAndAdd(delta)) + + /** + * Atomic compare-and-set: replaces `expected` with `updated` if the current value equals `expected`. + */ + inline def compareAndSet(inline expected: Long, inline updated: Long): CIO[Boolean] = + CIO.defer(self.compareAndSet(expected, updated)) + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CAtomicRef.scala b/sage-compat-direct/src/main/scala/kyo/compat/CAtomicRef.scala new file mode 100644 index 00000000..09981ca3 --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CAtomicRef.scala @@ -0,0 +1,64 @@ +package kyo.compat + +import java.util.concurrent.atomic.AtomicReference + +/** + * Uses `java.util.concurrent.atomic.AtomicReference[A]`. `lift` and `lower` return the existing atomic reference. `compareAndSet` follows + * the JDK contract and compares references, which matches the Ox binding. + */ +opaque type CAtomicRef[A] = AtomicReference[A] + +object CAtomicRef { + + /** + * Allocates a fresh atomic reference initialized to `a`. + */ + inline def init[A](inline a: A): CIO[CAtomicRef[A]] = + CIO.defer(new AtomicReference[A](a)) + + /** + * Wraps a native `AtomicReference` as a `CAtomicRef`. The conversion is the identity on the carrier. + */ + inline def lift[A](inline u: AtomicReference[A]): CAtomicRef[A] = u + + extension [A](inline self: CAtomicRef[A]) { + + /** + * Unwraps to the native `AtomicReference`. The conversion is the identity on the carrier. + */ + inline def lower: AtomicReference[A] = self + + /** + * Reads the current value. + */ + inline def get: CIO[A] = CIO.defer(self.get()) + + /** + * Atomically sets the value to `a`. + */ + inline def set(inline a: A): CIO[Unit] = CIO.defer(self.set(a)) + + /** + * Atomically sets the value to `a` and returns the previous value. + */ + inline def getAndSet(inline a: A): CIO[A] = CIO.defer(self.getAndSet(a)) + + /** + * Atomically applies `f` to the current value and returns the new value. + */ + inline def updateAndGet(inline f: A => A): CIO[A] = CIO.defer(self.updateAndGet(a => f(a))) + + /** + * Atomically applies `f` to the current value and returns the previous value. + */ + inline def getAndUpdate(inline f: A => A): CIO[A] = CIO.defer(self.getAndUpdate(a => f(a))) + + /** + * Atomic compare-and-set: replaces `expected` with `updated` if the current value is the same reference as `expected`. + */ + inline def compareAndSet(inline expected: A, inline updated: A): CIO[Boolean] = + CIO.defer(self.compareAndSet(expected, updated)) + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CChannel.scala b/sage-compat-direct/src/main/scala/kyo/compat/CChannel.scala new file mode 100644 index 00000000..516e8708 --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CChannel.scala @@ -0,0 +1,48 @@ +package kyo.compat + +import java.util.concurrent.LinkedBlockingQueue + +/** + * Uses a bounded `java.util.concurrent.LinkedBlockingQueue[A]`. `lift` and `lower` return the existing queue. `put` blocks while the + * channel is full and `take` blocks while it is empty. `poll` does not block. + */ +opaque type CChannel[A] = LinkedBlockingQueue[A] + +object CChannel { + + /** + * Allocates a bounded FIFO channel with the given capacity. + */ + inline def init[A](inline capacity: Int): CIO[CChannel[A]] = + CIO.defer(new LinkedBlockingQueue[A](capacity)) + + /** + * Wraps a native `LinkedBlockingQueue` as a `CChannel`. The conversion is the identity on the carrier. + */ + inline def lift[A](inline u: LinkedBlockingQueue[A]): CChannel[A] = u + + extension [A](inline self: CChannel[A]) { + + /** + * Unwraps to the native `LinkedBlockingQueue`. The conversion is the identity on the carrier. + */ + inline def lower: LinkedBlockingQueue[A] = self + + /** + * Enqueues `v`, blocking while the channel is full. + */ + inline def put(inline v: A): CIO[Unit] = CIO.defer(self.put(v)) + + /** + * Dequeues the next element, blocking while the channel is empty. + */ + inline def take: CIO[A] = CIO.defer(self.take()) + + /** + * Returns the next element if one is available, without blocking. + */ + inline def poll: CIO[Option[A]] = CIO.defer(Option(self.poll())) + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CChunk.scala b/sage-compat-direct/src/main/scala/kyo/compat/CChunk.scala new file mode 100644 index 00000000..f264d862 --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CChunk.scala @@ -0,0 +1,55 @@ +package kyo.compat + +/** + * Uses Scala's `Vector[A]`. `lift` and `lower` return the existing `Vector`. Every backend provides `toSeq`, `toIndexedSeq`, `apply`, + * `size`, `iterator`, and `isEmpty`. + */ +opaque type CChunk[+A] = Vector[A] + +object CChunk { + + /** + * Wraps a `Vector` as a `CChunk`. The conversion is the identity on the carrier. + */ + inline def lift[A](inline c: Vector[A]): CChunk[A] = c + + extension [A](inline self: CChunk[A]) { + + /** + * Unwraps to the native `Vector`. The conversion is the identity on the carrier. + */ + inline def lower: Vector[A] = self + + /** + * Views the chunk as a `Seq[A]`. + */ + inline def toSeq: Seq[A] = self + + /** + * Views the chunk as an `IndexedSeq[A]`. + */ + inline def toIndexedSeq: IndexedSeq[A] = self + + /** + * Returns the element at index `i`. + */ + inline def apply(inline i: Int): A = self(i) + + /** + * Returns the number of elements. + */ + inline def size: Int = self.size + + /** + * Returns an iterator over the chunk's elements. + */ + inline def iterator: Iterator[A] = self.iterator + + /** + * Returns `true` if the chunk has no elements. + */ + inline def isEmpty: Boolean = self.isEmpty + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CFiber.scala b/sage-compat-direct/src/main/scala/kyo/compat/CFiber.scala new file mode 100644 index 00000000..211677e9 --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CFiber.scala @@ -0,0 +1,35 @@ +package kyo.compat + +/** + * Uses a fiber from the binding's runtime, which runs a computation on its own virtual thread. A fiber is not tied to the computation that started + * it. It runs until its body completes, fails, or its thread is interrupted. + */ +opaque type CFiber[+A] = DirectRuntime.Fiber[A] + +object CFiber { + + /** + * Starts `c` on a new virtual thread. The thread inherits the caller's `CLocal` values. + */ + def init[A](c: CIO[A]): CIO[CFiber[A]] = + CIO.defer(DirectRuntime.Fiber.start(() => c.lower)) + + extension [A](self: CFiber[A]) { + + /** + * Blocks until the fiber completes and returns its result or rethrows its failure. + */ + def get: CIO[A] = CIO.defer(self.await()) + + /** + * Registers `cb` to run on a new virtual thread when the fiber completes. A callback registered after completion runs immediately. A + * failure inside `cb` goes to the thread's uncaught exception handler. + */ + def onComplete(cb: scala.util.Try[A] => CIO[Unit]): CIO[Unit] = + CIO.defer { + self.result.whenComplete((_, _) => DirectRuntime.threads.newThread(() => cb(DirectRuntime.outcome(self.result)).lower).start()): Unit + } + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CIO.scala b/sage-compat-direct/src/main/scala/kyo/compat/CIO.scala new file mode 100644 index 00000000..3cf131b5 --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CIO.scala @@ -0,0 +1,382 @@ +package kyo.compat + +import java.util.concurrent.{CompletableFuture, CompletionException, CompletionStage} + +import scala.concurrent.{Await, Future as ScalaFuture} +import scala.concurrent.duration.{Duration as ScalaDuration, FiniteDuration, NANOSECONDS} +import scala.jdk.FutureConverters.* +import scala.util.{Failure, Success, Try} +import scala.util.control.NonFatal + +/** + * Uses a function from the current `flatMap` nesting depth to the result. `lift` wraps ordinary blocking code, and `lower` calls the + * function at depth 0 on the calling thread. Errors are thrown exceptions and cancellation is thread interruption. `flatMap` passes the + * depth on and calls its functions directly. Once the depth reaches 256, the rest of the chain runs on a fresh virtual thread that the + * caller waits for, so long chains and recursive loops do not overflow the stack. `race` still runs its first argument on the caller. `async` blocks the calling thread until the callback + * fires, which is cheap on a virtual thread. Concurrent operators such as `race`, `timeout`, `zip`, and `foreach` run work on virtual + * threads and wait for every thread they start before returning. + */ +opaque type CIO[+A] = Int => A + +object CIO { + + // The maximum `flatMap` nesting on one thread's stack. Deeper chains continue on a fresh virtual thread. + private inline val Limit = 256 + + /** + * Wraps blocking code as a `CIO`. The code runs each time the `CIO` runs. + */ + def lift[A](f: => A): CIO[A] = _ => f + + /** + * Returns a successful `CIO` with the given value. + */ + def value[A](a: A): CIO[A] = _ => a + + /** + * Returns a successful `CIO[Unit]`. + */ + def unit: CIO[Unit] = _ => () + + /** + * Suspends a side-effecting thunk until the `CIO` runs. + */ + def defer[A](thunk: => A): CIO[A] = _ => thunk + + /** + * Returns a `CIO` that fails with `e`. + */ + def fail(e: Throwable): CIO[Nothing] = _ => throw e + + /** + * Succeeds with the value of a `Success` and fails with the throwable of a `Failure`. + */ + def get[A](t: Try[A]): CIO[A] = _ => t.get + + /** + * Waits for a Scala `Future` by blocking the calling thread. + */ + def fromScalaFuture[A](f: => ScalaFuture[A]): CIO[A] = _ => Await.result(f, ScalaDuration.Inf) + + /** + * Waits for a `CompletionStage` by blocking the calling thread. A failure is rethrown without the `CompletionException` that dependent + * stages wrap it in. Interrupting the caller stops the wait but does not cancel the stage. + */ + def fromCompletionStage[A](cs: => CompletionStage[A]): CIO[A] = + _ => { + val future = cs.toCompletableFuture + try DirectRuntime.await(future) + catch { case e: CompletionException if e.getCause ne null => throw e.getCause } + } + + /** + * Runs `use` with the acquired value and runs `release` on every exit path, including interruption. A release failure is added as a + * suppressed exception when `use` also fails. + */ + def acquireReleaseWith[A, B]( + acquire: CIO[A] + )( + release: A => CIO[Unit] + )( + use: A => CIO[B] + ): CIO[B] = + depth => { + val resource = acquire(depth) + // Release runs on its own thread and the caller waits for it without reacting to interrupts, so a pending interrupt cannot cut it + // short. The caller's interrupt status is restored afterward. + def releaseResource(): Unit = DirectRuntime.uninterruptible(release(resource).lower) + val result = + try use(resource)(depth) + catch { + case useFailure: Throwable => + try releaseResource() + catch { case releaseFailure: Throwable if releaseFailure ne useFailure => useFailure.addSuppressed(releaseFailure) } + throw useFailure + } + releaseResource() + result + } + + // Runs `c` inside `wrap` on the calling thread at the depth of the enclosing `CIO`, so a recursion through `wrap` still reaches the + // `flatMap` limit. Calling `c.lower` there would restart the depth at 0 on every level and overflow the stack. + private[compat] def around[A, B](c: => CIO[A])(wrap: (() => A) => B): CIO[B] = depth => wrap(() => c(depth)) + + /** + * Runs `c` and then `cleanup` on every exit path. + */ + def ensure[A]( + cleanup: CIO[Unit] + )( + c: CIO[A] + ): CIO[A] = + acquireReleaseWith(CIO.unit)(_ => cleanup)(_ => c) + + /** + * Returns a `CIO` that blocks until the thread is interrupted. + */ + val never: CIO[Nothing] = CIO.async[Nothing](_ => ()) + + extension [A](self: CIO[A]) { + + /** + * Runs this `CIO` on the calling thread and returns its result. + */ + def lower: A = self(0) + + /** + * Captures a non-fatal failure as `Failure`. Interruption and other fatal throwables are not captured. + */ + def liftToTry: CIO[Try[A]] = + depth => + try Success(self(depth)) + catch { case NonFatal(t) => Failure(t) } + + /** + * Runs `handler` on failure to produce a recovery `CIO`. + */ + def recover[A2 >: A](handler: Throwable => CIO[A2]): CIO[A2] = self.fold(CIO.value(_), handler) + + /** + * Handles success and failure with the matching function. + */ + def fold[B]( + onSuccess: A => CIO[B], + onFail: Throwable => CIO[B] + ): CIO[B] = + self.liftToTry.flatMap { + case Success(a) => onSuccess(a) + case Failure(t) => onFail(t) + } + + /** + * Discards the success value. A failure still propagates. + */ + def unit: CIO[Unit] = self.map(_ => ()) + + /** + * Runs `that` if this `CIO` fails. + */ + def orElse[A2 >: A](that: CIO[A2]): CIO[A2] = self.recover(_ => that) + + /** + * Transforms a failure with `f`. + */ + def mapError(f: Throwable => Throwable): CIO[A] = self.recover(t => CIO.fail(f(t))) + + /** + * Transforms the success value with a pure function. + */ + def map[B](f: A => B): CIO[B] = self.flatMap(a => CIO.value(f(a))) + + /** + * Continues with a `CIO` built from the success value. When the depth reaches the limit, this step and the rest of the chain run at + * depth 0 on a new virtual thread. Its failures reach the caller unchanged, it inherits the caller's `CLocal` values, and interrupting + * the caller interrupts the thread and waits for it to exit before the interrupt is rethrown. While it waits, the caller runs the + * first argument of each `race` in the chain. + */ + def flatMap[B](f: A => CIO[B]): CIO[B] = + depth => + if (depth < Limit) f(self(depth + 1))(depth + 1) + else DirectRuntime.bounce(() => f(self.lower).lower) + + /** + * Runs this `CIO` on a new virtual thread and returns a `Future` with its outcome. + */ + def unsafeRun: ScalaFuture[A] = DirectRuntime.Fiber.start(() => self.lower).result.asScala + + } + + /** + * Blocks the calling thread for `d`. + */ + def sleep(d: FiniteDuration): CIO[Unit] = _ => if (d.toNanos > 0) Thread.sleep(java.time.Duration.ofNanos(d.toNanos)) + + /** + * Returns the wall-clock time. + */ + def now: CIO[java.time.Instant] = _ => java.time.Instant.now() + + /** + * Returns a monotonic timestamp. Use it to measure elapsed time between two points in the program. + */ + def nowMonotonic: CIO[FiniteDuration] = _ => FiniteDuration(System.nanoTime(), NANOSECONDS) + + /** + * Runs `c` with a deadline and returns `None` if `d` elapses first. The computation runs on a virtual thread that is interrupted at the + * deadline or when the caller is interrupted, and this `CIO` returns only after that thread has exited. + */ + def timeout[A](d: FiniteDuration)(c: CIO[A]): CIO[Option[A]] = + _ => DirectRuntime.Fiber.scoped(() => c.lower)(_.await(d)) + + /** + * Runs `c` with a deadline and fails with `e` if `d` elapses first. + */ + def timeoutWithError[A](d: FiniteDuration)(e: Throwable)(c: CIO[A]): CIO[A] = + _ => DirectRuntime.Fiber.scoped(() => c.lower)(_.await(d)).getOrElse(throw e) + + /** + * Sleeps for `d` and then runs `c`. + */ + def delay[A](d: FiniteDuration)(c: CIO[A]): CIO[A] = + CIO.sleep(d).flatMap(_ => c) + + /** + * Runs `a` and `b` concurrently and returns the first successful result. `a` runs on the thread that called `lower`, even when a long + * `flatMap` chain has moved to another thread before the race, and `b` runs on a virtual thread. The losing side is interrupted. + */ + def race[A]( + a: CIO[A], + b: CIO[A] + ): CIO[A] = + // `a` starts at depth 0 so that its first steps, such as a lock body, run before its chain can move to another thread. Each nested + // race can therefore add up to the depth limit to the stack of the thread that runs it. + _ => DirectRuntime.onHome(() => DirectRuntime.race(() => a.lower, () => b.lower)) + + /** + * Maps the elements concurrently, with at most `concurrency` computations running at once. + */ + def foreach[A, B]( + coll: Iterable[A], + concurrency: Int = Int.MaxValue + )(f: A => CIO[B]): CIO[CChunk[B]] = + CIO.defer(CChunk.lift(DirectRuntime.parallel(coll, concurrency)(a => f(a).lower))) + + /** + * Like [[foreach]], and passes each element's index to `f`. + */ + def foreachIndexed[A, B]( + coll: Iterable[A], + concurrency: Int = Int.MaxValue + )(f: (Int, A) => CIO[B]): CIO[CChunk[B]] = + foreach(coll.view.zipWithIndex, concurrency) { case (a, i) => f(i, a) } + + /** + * Runs `f` for each element concurrently and discards the results. + */ + def foreachDiscard[A]( + coll: Iterable[A], + concurrency: Int = Int.MaxValue + )(f: A => CIO[Any]): CIO[Unit] = + foreach(coll, concurrency)(f).unit + + /** + * Keeps the elements that satisfy `p`, evaluating the predicate concurrently. + */ + def filter[A]( + coll: Iterable[A], + concurrency: Int = Int.MaxValue + )(p: A => CIO[Boolean]): CIO[CChunk[A]] = + foreach(coll, concurrency)(a => p(a).map(Option.when(_)(a))).map(kept => CChunk.lift(kept.lower.flatten)) + + /** + * Runs the computations concurrently and returns their results in order. + */ + def collectAll[A]( + coll: Iterable[CIO[A]], + concurrency: Int = Int.MaxValue + ): CIO[CChunk[A]] = + foreach(coll, concurrency)(c => c) + + /** + * Runs the computations concurrently and discards the results. + */ + def collectAllDiscard( + coll: Iterable[CIO[Any]], + concurrency: Int = Int.MaxValue + ): CIO[Unit] = + foreachDiscard(coll, concurrency)(c => c) + + /** + * Bridges a one-shot completion callback. `register` receives the callback, and the calling thread blocks until it fires. Interrupting the + * caller stops the wait. A callback that fires afterward is ignored. + */ + def async[A](register: ((Try[A] => Unit) => Unit)): CIO[A] = + _ => { + val outcome = new CompletableFuture[Try[A]]() + register(t => outcome.complete(t): Unit) + outcome.get().get + } + + /** + * Yields the calling thread so other threads can run. + */ + def cede: CIO[Unit] = _ => Thread.`yield`() + + /** + * Runs blocking code. Every `CIO` already runs on the calling thread, so this is the same as [[defer]]. + */ + def blocking[A](thunk: => A): CIO[A] = _ => thunk + + private def zipAll[T <: Tuple](cios: Tuple.Map[T, CIO]): CIO[T] = + collectAll(cios.productIterator.map(_.asInstanceOf[CIO[Any]]).toVector).map(results => Tuple.fromArray(results.lower.toArray).asInstanceOf[T]) + + /** + * Runs two computations concurrently and returns both results. + */ + def zip[A1, A2]( + a1: CIO[A1], + a2: CIO[A2] + ): CIO[(A1, A2)] = + zipAll[(A1, A2)]((a1, a2)) + + /** + * Runs three computations concurrently and returns all results. + */ + def zip[A1, A2, A3]( + a1: CIO[A1], + a2: CIO[A2], + a3: CIO[A3] + ): CIO[(A1, A2, A3)] = + zipAll[(A1, A2, A3)]((a1, a2, a3)) + + /** + * Runs four computations concurrently and returns all results. + */ + def zip[A1, A2, A3, A4]( + a1: CIO[A1], + a2: CIO[A2], + a3: CIO[A3], + a4: CIO[A4] + ): CIO[(A1, A2, A3, A4)] = + zipAll[(A1, A2, A3, A4)]((a1, a2, a3, a4)) + + /** + * Runs five computations concurrently and returns all results. + */ + def zip[A1, A2, A3, A4, A5]( + a1: CIO[A1], + a2: CIO[A2], + a3: CIO[A3], + a4: CIO[A4], + a5: CIO[A5] + ): CIO[(A1, A2, A3, A4, A5)] = + zipAll[(A1, A2, A3, A4, A5)]((a1, a2, a3, a4, a5)) + + /** + * Runs six computations concurrently and returns all results. + */ + def zip[A1, A2, A3, A4, A5, A6]( + a1: CIO[A1], + a2: CIO[A2], + a3: CIO[A3], + a4: CIO[A4], + a5: CIO[A5], + a6: CIO[A6] + ): CIO[(A1, A2, A3, A4, A5, A6)] = + zipAll[(A1, A2, A3, A4, A5, A6)]((a1, a2, a3, a4, a5, a6)) + + /** + * Runs seven computations concurrently and returns all results. + */ + def zip[A1, A2, A3, A4, A5, A6, A7]( + a1: CIO[A1], + a2: CIO[A2], + a3: CIO[A3], + a4: CIO[A4], + a5: CIO[A5], + a6: CIO[A6], + a7: CIO[A7] + ): CIO[(A1, A2, A3, A4, A5, A6, A7)] = + zipAll[(A1, A2, A3, A4, A5, A6, A7)]((a1, a2, a3, a4, a5, a6, a7)) + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CLatch.scala b/sage-compat-direct/src/main/scala/kyo/compat/CLatch.scala new file mode 100644 index 00000000..5749092c --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CLatch.scala @@ -0,0 +1,43 @@ +package kyo.compat + +import java.util.concurrent.CountDownLatch + +/** + * Uses `java.util.concurrent.CountDownLatch`. `lift` and `lower` return the existing latch. A count of zero or less creates a latch that is + * already released. + */ +opaque type CLatch = CountDownLatch + +object CLatch { + + /** + * Allocates a latch with count `n`. + */ + inline def init(inline n: Int): CIO[CLatch] = + CIO.defer(new CountDownLatch(math.max(0, n))) + + /** + * Wraps a native `CountDownLatch` as a `CLatch`. The conversion is the identity on the carrier. + */ + inline def lift(inline u: CountDownLatch): CLatch = u + + extension (inline self: CLatch) { + + /** + * Unwraps to the native `CountDownLatch`. The conversion is the identity on the carrier. + */ + inline def lower: CountDownLatch = self + + /** + * Blocks until the count reaches zero. + */ + inline def await: CIO[Unit] = CIO.defer(self.await()) + + /** + * Decrements the count. Waiters are released when it reaches zero. + */ + inline def release: CIO[Unit] = CIO.defer(self.countDown()) + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CLocal.scala b/sage-compat-direct/src/main/scala/kyo/compat/CLocal.scala new file mode 100644 index 00000000..5ced4ed1 --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CLocal.scala @@ -0,0 +1,57 @@ +package kyo.compat + +/** + * Uses an `InheritableThreadLocal[A]`. `let` installs a value for the duration of a computation on the current thread and restores the + * previous value afterward. Virtual threads started by this binding inherit the value that is current when they start. + */ +opaque type CLocal[A] = InheritableThreadLocal[A] + +object CLocal { + + /** + * Allocates a fresh local whose value is `default` on every thread until `let` changes it. + */ + def init[A](default: => A): CIO[CLocal[A]] = + CIO.defer { + val initial = default + new InheritableThreadLocal[A] { + override def initialValue(): A = initial + } + } + + /** + * Wraps a native `InheritableThreadLocal` as a `CLocal`. The conversion is the identity on the carrier. + */ + inline def lift[A](inline u: InheritableThreadLocal[A]): CLocal[A] = u + + extension [A](self: CLocal[A]) { + + /** + * Unwraps to the native `InheritableThreadLocal`. The conversion is the identity on the carrier. + */ + inline def lower: InheritableThreadLocal[A] = self + + /** + * Reads the current thread's value. + */ + inline def get: CIO[A] = CIO.defer(self.get()) + + /** + * Installs `v` while `c` runs, then restores the previous value. + */ + def let[B](v: => A)(c: => CIO[B]): CIO[B] = self.update(_ => v)(c) + + /** + * Installs `f` applied to the current value while `c` runs, then restores the previous value. + */ + def update[B](f: A => A)(c: => CIO[B]): CIO[B] = + CIO.around(c) { run => + val previous = self.get() + self.set(f(previous)) + try run() + finally self.set(previous) + } + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CMeter.scala b/sage-compat-direct/src/main/scala/kyo/compat/CMeter.scala new file mode 100644 index 00000000..d662f20d --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CMeter.scala @@ -0,0 +1,60 @@ +package kyo.compat + +import java.util.concurrent.Semaphore + +/** + * Uses `java.util.concurrent.Semaphore`. `lift` and `lower` return the existing semaphore. `run` releases its permit when the computation + * succeeds, fails, or is interrupted. + */ +opaque type CMeter = Semaphore + +object CMeter { + + /** + * Allocates a semaphore with `permits` permits. + */ + inline def init(inline permits: Int): CIO[CMeter] = + CIO.defer(new Semaphore(permits)) + + /** + * Wraps a native `Semaphore` as a `CMeter`. The conversion is the identity on the carrier. + */ + inline def lift(inline u: Semaphore): CMeter = u + + extension (self: CMeter) { + + /** + * Unwraps to the native `Semaphore`. The conversion is the identity on the carrier. + */ + inline def lower: Semaphore = self + + /** + * Acquires a permit, blocking until one is available, runs `c`, and then releases the permit. + */ + def run[A](c: => CIO[A]): CIO[A] = + CIO.around(c) { run => + self.acquire() + try run() + finally self.release() + } + + /** + * Runs `c` if a permit is available immediately. Returns `None` without running `c` otherwise. + */ + def tryRun[A](c: => CIO[A]): CIO[Option[A]] = + CIO.around(c) { run => + if (!self.tryAcquire()) None + else + try Some(run()) + finally self.release() + } + + /** + * Returns the number of available permits. + */ + inline def availablePermits: CIO[Int] = + CIO.defer(self.availablePermits()) + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CPromise.scala b/sage-compat-direct/src/main/scala/kyo/compat/CPromise.scala new file mode 100644 index 00000000..1ec0825a --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CPromise.scala @@ -0,0 +1,62 @@ +package kyo.compat + +import java.util.concurrent.CompletableFuture + +/** + * Uses `java.util.concurrent.CompletableFuture[A]`. `lift` and `lower` return the existing future. `succeed` and `fail` return `true` only + * for the first completion. `get` blocks the calling thread and rethrows the original failure. `poll` reports a cancelled future as pending. + */ +opaque type CPromise[A] = CompletableFuture[A] + +object CPromise { + + /** + * Allocates a fresh single-shot promise. + */ + inline def init[A]: CIO[CPromise[A]] = + CIO.defer(new CompletableFuture[A]()) + + /** + * Wraps a native `CompletableFuture` as a `CPromise`. The conversion is the identity on the carrier. + */ + inline def lift[A](inline u: CompletableFuture[A]): CPromise[A] = u + + extension [A](inline self: CPromise[A]) { + + /** + * Unwraps to the native `CompletableFuture`. The conversion is the identity on the carrier. + */ + inline def lower: CompletableFuture[A] = self + + /** + * Attempts to complete the promise with `a`. Returns `true` if this is the first completion. + */ + inline def succeed(inline a: A): CIO[Boolean] = + CIO.defer(self.complete(a)) + + /** + * Attempts to complete the promise with failure `e`. Returns `true` if this is the first completion. + */ + inline def fail(inline e: Throwable): CIO[Boolean] = + CIO.defer(self.completeExceptionally(e)) + + /** + * Blocks until the promise is completed and returns its value or rethrows its failure. + */ + inline def get: CIO[A] = CIO.defer(DirectRuntime.await(self.lower)) + + /** + * Returns `None` while the promise is pending or cancelled, and `Some` with the outcome once it is completed. + */ + inline def poll: CIO[Option[scala.util.Try[A]]] = + CIO.defer(Option.when(self.isDone && !self.isCancelled)(DirectRuntime.outcome(self.lower))) + + /** + * Returns `true` if the promise has been completed. + */ + inline def done: CIO[Boolean] = + CIO.defer(self.isDone) + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/CStream.scala b/sage-compat-direct/src/main/scala/kyo/compat/CStream.scala new file mode 100644 index 00000000..24e83b2e --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/CStream.scala @@ -0,0 +1,131 @@ +package kyo.compat + +/** + * Uses a function that returns a fresh `Iterator[A]`, so a stream can run more than once. Elements are produced lazily on the thread that + * consumes the iterator, and effectful stages run their `CIO` when the element is pulled. A failure in a stage is thrown from the + * iterator's `hasNext` or `next`. + */ +opaque type CStream[+A] = () => Iterator[A] + +object CStream { + + /** + * Returns an empty stream. + */ + def empty[A]: CStream[A] = () => Iterator.empty + + /** + * Returns a stream of the elements produced by `c`. + */ + def init[A](c: CIO[Seq[A]]): CStream[A] = () => c.lower.iterator + + /** + * Returns a stream of the elements of `seq`. + */ + def init[A](seq: Seq[A]): CStream[A] = () => seq.iterator + + /** + * Returns the integers from `start` until `end`, excluding `end`. + */ + def range(start: Int, end: Int): CStream[Int] = () => Iterator.range(start, end) + + /** + * Unfolds a stream from `acc`. The stream ends when `f` returns `None`. + */ + def unfold[S, A](acc: S)(f: S => CIO[Option[(A, S)]]): CStream[A] = () => Iterator.unfold(acc)(s => f(s).lower) + + /** + * Wraps an iterator factory as a `CStream`. The factory runs each time the stream runs. + */ + def lift[A](native: => Iterator[A]): CStream[A] = () => native + + extension [A](self: CStream[A]) { + + /** + * Returns a new iterator over the stream. Each call runs the stream again. + */ + def lower: Iterator[A] = self() + + /** + * Appends `other` after this stream. + */ + def concat[A2 >: A](other: CStream[A2]): CStream[A2] = () => self() ++ other() + + /** + * Transforms each element with a pure function. + */ + def mapPure[B](f: A => B): CStream[B] = () => self().map(f) + + /** + * Transforms each element with an effectful function. + */ + def map[B](f: A => CIO[B]): CStream[B] = () => self().map(a => f(a).lower) + + /** + * Replaces each element with a stream and concatenates the results. + */ + def flatMap[B](f: A => CStream[B]): CStream[B] = () => self().flatMap(a => f(a)()) + + /** + * Runs `f` for each element and passes the element through unchanged. + */ + def tap(f: A => CIO[Any]): CStream[A] = + () => + self().map { a => + val _ = f(a).lower + a + } + + /** + * Keeps the first `n` elements. + */ + def take(n: Int): CStream[A] = () => self().take(n) + + /** + * Skips the first `n` elements. + */ + def drop(n: Int): CStream[A] = () => self().drop(n) + + /** + * Keeps elements while `p` holds. + */ + def takeWhilePure(p: A => Boolean): CStream[A] = () => self().takeWhile(p) + + /** + * Keeps the elements that satisfy the pure predicate. + */ + def filterPure(p: A => Boolean): CStream[A] = () => self().filter(p) + + /** + * Keeps the elements that satisfy the effectful predicate. + */ + def filter(p: A => CIO[Boolean]): CStream[A] = () => self().filter(a => p(a).lower) + + /** + * Transforms and filters the elements in one pass. + */ + def collectPure[B](f: A => Option[B]): CStream[B] = () => self().flatMap(f) + + /** + * Runs the stream and collects its elements. + */ + def run: CIO[CChunk[A]] = CIO.defer(CChunk.lift(self().toVector)) + + /** + * Runs the stream and folds its elements with a pure function. + */ + def foldPure[B](acc: B)(f: (B, A) => B): CIO[B] = CIO.defer(self().foldLeft(acc)(f)) + + /** + * Runs `f` for each element. + */ + def foreach(f: A => CIO[Unit]): CIO[Unit] = CIO.defer(self().foreach(a => f(a).lower)) + + /** + * Runs the stream and discards its elements. + */ + def discard: CIO[Unit] = CIO.defer(self().foreach(_ => ())) + + } + +} diff --git a/sage-compat-direct/src/main/scala/kyo/compat/DirectRuntime.scala b/sage-compat-direct/src/main/scala/kyo/compat/DirectRuntime.scala new file mode 100644 index 00000000..20ee393e --- /dev/null +++ b/sage-compat-direct/src/main/scala/kyo/compat/DirectRuntime.scala @@ -0,0 +1,190 @@ +package kyo.compat + +import java.util.concurrent.{CompletableFuture, ExecutionException, Executors, LinkedBlockingQueue, RejectedExecutionException, Semaphore, TimeUnit} +import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference} + +import scala.concurrent.duration.FiniteDuration +import scala.util.{Failure, Success, Try} +import scala.util.control.NonFatal + +// The thread helpers behind the direct binding. The doc comment of `CIO` describes the execution model. +private[compat] object DirectRuntime { + + // Creates every thread the binding starts for a computation. The factory is safe to share, and its threads inherit `CLocal` values. + private[compat] val threads: java.util.concurrent.ThreadFactory = Thread.ofVirtual().name("sage-direct").factory() + + private[compat] def uninterruptible[A](body: => A): A = { + val fiber = Fiber.start(() => body) + joinUninterruptibly(fiber.thread) + fiber.await() + } + + // Waits for `future` and rethrows the exact throwable it was completed with. + private[compat] def await[A](future: CompletableFuture[? <: A]): A = + try future.get() + catch { case _: ExecutionException => outcome(future).get } + + // Returns the outcome of a completed future. `get`, `join` and `exceptionNow` replace a `CompletionException` with its cause, so the + // outcome is read through `handle`, which passes the throwable unchanged. + private[compat] def outcome[A](future: CompletableFuture[? <: A]): Try[A] = + future.handle[Try[A]]((a, error) => if (error eq null) Success(a) else Failure(error)).join() + + /** + * Races `a` on the calling thread against `b` on a forked thread and returns the first successful result. Running `a` on the caller keeps + * its thread-local context, which matters when `a` is user code such as a lock body. If `b` succeeds first, the caller is interrupted to + * stop `a`, and the interrupt is cleared before `b`'s result is returned. If both fail, the first failure is thrown with the other one + * suppressed. + */ + private[compat] def race[A](a: () => A, b: () => A): A = { + val caller = Thread.currentThread() + // The side that sets `settled` first decides the race. When `b` succeeds first, it interrupts the caller to stop `a`. + val settled = new AtomicBoolean(false) + Fiber.scoped { () => + val value = b() + if (settled.compareAndSet(false, true)) caller.interrupt() + value + } { fiber => + // `b` won. Wait until its thread has exited so its interrupt has been delivered, then clear it. + def resultOfB(): A = { + joinUninterruptibly(fiber.thread) + Thread.interrupted(): Unit + fiber.await() + } + try { + val value = a() + if (settled.compareAndSet(false, true)) value else resultOfB() + } catch { + case failureA: Throwable if settled.compareAndSet(false, true) => + if (!NonFatal(failureA)) throw failureA + try fiber.await() + catch { + case NonFatal(failureB) => + failureA.addSuppressed(failureB) + throw failureA + } + case _: Throwable => resultOfB() + } + } + } + + /** + * Runs `run` for each item on its own virtual thread, at most `concurrency` at a time, and returns the results in input order. The first + * failure interrupts the remaining work and is thrown after every started thread has exited. + */ + private[compat] def parallel[A, B](items: Iterable[A], concurrency: Int)(run: A => B): Vector[B] = { + val permits = new Semaphore(math.max(1, concurrency)) + val failure = new AtomicReference[Throwable](null) + // `shutdownNow` interrupts every thread the executor has started and rejects later submissions, so the first failure stops all work. + val executor = Executors.newThreadPerTaskExecutor(threads) + val futures = Vector.newBuilder[java.util.concurrent.Future[B]] + try { + val remaining = items.iterator + while (remaining.hasNext && !executor.isShutdown) { + val item = remaining.next() + permits.acquire() + try + futures += executor.submit[B] { () => + try run(item) + catch { + case t: Throwable => + if (failure.compareAndSet(null, t)) executor.shutdownNow(): Unit + throw t + } finally permits.release() + }: Unit + catch { case _: RejectedExecutionException => permits.release() } + } + executor.shutdown() + executor.awaitTermination(Long.MaxValue, TimeUnit.NANOSECONDS): Unit + } catch { + case e: Throwable => + executor.shutdownNow(): Unit + executor.close() + throw e + } + val failed = failure.get() + if (failed ne null) throw failed + futures.result().map(_.get()) + } + + // The work queue of the thread that waits for the current bounced chain (see `CIO.flatMap`). It is set on the threads that continue the + // chain, and it is a plain `ThreadLocal`, so threads that they fork do not inherit it. + private val home = new ThreadLocal[LinkedBlockingQueue[Runnable]] + + /** + * Continues a chain on a fresh virtual thread and waits for it. The thread that started the chain runs the work that [[onHome]] hands + * back to it while it waits, so that work sees the thread-local values of the code that started the chain. + */ + private[compat] def bounce[A](body: () => A): A = { + val inherited = home.get + val inbox = if (inherited eq null) new LinkedBlockingQueue[Runnable] else inherited + Fiber.scoped { () => + home.set(inbox) + body() + } { fiber => + if (inherited eq null) { + // wake the loop once the chain has finished + fiber.result.whenComplete((_, _) => inbox.put(() => ())): Unit + while (!fiber.result.isDone) inbox.take().run() + } + fiber.await() + } + } + + // Runs `body` on the thread that started the current bounced chain, or on the calling thread outside a bounced chain. + private[compat] def onHome[A](body: () => A): A = { + val inbox = home.get + if (inbox eq null) body() + else { + val result = new CompletableFuture[A]() + inbox.put(() => completeWith(result)(body)) + await(result) + } + } + + private def completeWith[A](result: CompletableFuture[A])(body: () => A): Unit = + try result.complete(body()): Unit + catch { case t: Throwable => result.completeExceptionally(t): Unit } + + private[compat] def joinUninterruptibly(thread: Thread): Unit = { + var interrupted = false + while (thread.isAlive) + try thread.join() + catch { case _: InterruptedException => interrupted = true } + if (interrupted) Thread.currentThread().interrupt() + } + + // A computation running on its own virtual thread. `result` holds its outcome, including fatal throwables. + final private[compat] class Fiber[+A] private (val thread: Thread, val result: CompletableFuture[? <: A]) { + + def await(): A = DirectRuntime.await(result) + + // Returns `None` when `d` elapses before the thread exits. The thread completes `result` as its last step, so `await()` then returns at once. + def await(d: FiniteDuration): Option[A] = + if (thread.join(java.time.Duration.ofNanos(d.toNanos))) Some(await()) else None + } + + private[compat] object Fiber { + + def start[A](body: () => A): Fiber[A] = { + val result = new CompletableFuture[A]() + val thread = threads.newThread(() => completeWith(result)(body)) + val fiber = new Fiber(thread, result) + thread.start() + fiber + } + + /** + * Runs `body` on a new fiber and passes the fiber to `use`. If the fiber is still running when `use` returns or throws, it is interrupted + * and joined first, so a forked computation never outlives the operation that started it. + */ + def scoped[A, B](body: () => A)(use: Fiber[A] => B): B = { + val fiber = start(body) + try use(fiber) + finally + if (!fiber.result.isDone) { + fiber.thread.interrupt() + joinUninterruptibly(fiber.thread) + } + } + } +} From f70c919f02290ae868f984fb03ef4e7357dc425b Mon Sep 17 00:00:00 2001 From: Pierre Ricadat Date: Sun, 25 Oct 2026 08:52:34 +0900 Subject: [PATCH 2/2] Show the direct client first in the docs --- README.md | 2 +- docs/client-side-caching.md | 2 +- docs/commands.md | 2 +- docs/distributed-locks.md | 2 +- docs/error-handling.md | 6 +-- docs/faq.md | 4 +- docs/getting-started.md | 68 +++++++++++++++++----------------- docs/index.md | 4 +- docs/json.md | 4 +- docs/pipelines-transactions.md | 6 +-- docs/pubsub.md | 18 ++++----- docs/rate-limiting.md | 2 +- docs/streams.md | 12 +++--- examples/README.md | 12 +++--- 14 files changed, 72 insertions(+), 72 deletions(-) diff --git a/README.md b/README.md index 6908ce75..f0a5c70d 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ Sage is a Redis and Valkey client for Scala 3. It implements the Redis protocol directly in Scala instead of wrapping a Java client. -- Choose the artifact for [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), [Ox](https://ox.softwaremill.com), or [Pekko](https://pekko.apache.org). Each artifact uses its ecosystem's native types. The direct artifact has no effect library dependency and serves plain blocking code. +- Use the direct artifact for plain blocking code with no effect library dependency, or choose the artifact for [Ox](https://ox.softwaremill.com), [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), or [Pekko](https://pekko.apache.org), which uses that ecosystem's native types. - The core implements RESP3, commands, and codecs in Scala 3. - Redis 8+ and Valkey 8+ support includes auto-pipelining, transactions, cluster routing, sharded pub/sub, client-side caching, rate limiting, distributed locks, and TLS. diff --git a/docs/client-side-caching.md b/docs/client-side-caching.md index 50a30d4a..1a346b87 100644 --- a/docs/client-side-caching.md +++ b/docs/client-side-caching.md @@ -6,7 +6,7 @@ Caching is opt-in per call. An ordinary `get` always goes to the server; only `c ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.set("cached:key", "v1") // first fetches and caches; second is a local hit val v1 = client.cached(Commands.get[String, String]("cached:key"), 1.minute) diff --git a/docs/commands.md b/docs/commands.md index 9df72191..3012a853 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -8,7 +8,7 @@ A `Command[Out]` describes a server command without running it. Client methods c ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] // per-command sugar val greeting = client.get[String]("greeting") diff --git a/docs/distributed-locks.md b/docs/distributed-locks.md index dd422623..2b35bdf2 100644 --- a/docs/distributed-locks.md +++ b/docs/distributed-locks.md @@ -14,7 +14,7 @@ locks.withLock("account:42", waitTimeout = 2.seconds) { } ``` -The body returns your backend's native effect. With Ox, it is a direct-style block. Construct the operation inside the block: an already-started `Future` may run before acquisition. Keep all protected work inside the returned effect. Wait for every child task before the body completes because detached work can continue after Sage releases the lock. +The body returns your backend's native effect. With the direct client and Ox, it is a plain block of code. Construct the operation inside the block: an already-started `Future` may run before acquisition. Keep all protected work inside the returned effect. Wait for every child task before the body completes because detached work can continue after Sage releases the lock. ## Waiting for a lock diff --git a/docs/error-handling.md b/docs/error-handling.md index a3cdbf18..2f1fde37 100644 --- a/docs/error-handling.md +++ b/docs/error-handling.md @@ -1,6 +1,6 @@ # Error handling -Every Sage failure is a `SageException` in a sealed hierarchy that you can match exhaustively. ZIO and Kyo include `SageException` in the error type as `IO[SageException, *]` and `Abort[SageException]`. Cats Effect raises the exception in `IO`, Ox and the direct client throw it, and Pekko returns a failed `scala.concurrent.Future`. +Every Sage failure is a `SageException` in a sealed hierarchy that you can match exhaustively. The direct client and Ox throw it. ZIO and Kyo include `SageException` in the error type as `IO[SageException, *]` and `Abort[SageException]`. Cats Effect raises the exception in `IO`, and Pekko returns a failed `scala.concurrent.Future`. ## The hierarchy @@ -73,9 +73,9 @@ Sage does not retry `-CLUSTERDOWN` for [commands that run on every master](/conf Each ecosystem reports the same `SageException` through its normal failure channel. ZIO and Kyo also include it in the error type. In those backends, another exception is a defect, such as a ZIO die or Kyo `Panic`. +- The direct client throws the exception from the blocking call. Handle it with `try` and `catch`. +- Ox throws the exception in direct style. Handle it with `try` and `catch`. - ZIO returns a failed `IO[SageException, *]`. Recover with `catchAll` or `catchSome`. - Cats Effect raises the exception in `IO`. Recover with `handleErrorWith` or `recoverWith` and match on `SageException`. - Kyo returns an `Abort[SageException]`. Handle it with the `Abort` combinators. -- Ox throws the exception in direct style. Handle it with `try` and `catch`. -- The direct client throws the exception from the blocking call. Handle it with `try` and `catch`. - Pekko returns a failed `scala.concurrent.Future`. Recover with `recover` or `recoverWith`. diff --git a/docs/faq.md b/docs/faq.md index fec8bc3c..778f775b 100644 --- a/docs/faq.md +++ b/docs/faq.md @@ -10,12 +10,12 @@ Run [the benchmarks](https://github.com/ghostdogpr/sage/tree/main/benchmarks) ag One per Scala stack, all sharing the same runtime: +- Plain blocking code, with no effect library: `sage-client-direct` +- Ox: `sage-client-ox` - ZIO: `sage-client-zio` - Cats Effect: `sage-client-ce` - Kyo: `sage-client-kyo` -- Ox: `sage-client-ox` - Pekko: `sage-client-pekko` -- Plain blocking code, with no effect library: `sage-client-direct` `sage-core` comes in transitively, so you depend on the backend artifact only. See [Getting started](/getting-started). diff --git a/docs/getting-started.md b/docs/getting-started.md index ba19a0c7..f271aded 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -2,7 +2,7 @@ Sage is a [Redis](https://redis.io) and [Valkey](https://valkey.io) client for [Scala 3](https://www.scala-lang.org/). It implements RESP3, commands, and codecs directly in Scala. The core has no dependencies on an effect system. -Sage provides integrations for [Ox](https://ox.softwaremill.com), [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), and [Apache Pekko](https://pekko.apache.org). Each integration uses its ecosystem's native types. The direct client has no dependency on an effect library: each command blocks the calling thread and returns its result, which suits frameworks that run each request on a virtual thread. Sage targets Redis 8+ and Valkey 8+. It supports Scala 3.9.x LTS and later and requires JDK 21 or later. +Sage provides a direct client for plain blocking code and integrations for [Ox](https://ox.softwaremill.com), [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), and [Apache Pekko](https://pekko.apache.org). The direct client has no dependency on an effect library: each command blocks the calling thread and returns its result, which suits frameworks that run each request on a virtual thread. Each integration uses its ecosystem's native types. Sage targets Redis 8+ and Valkey 8+. It supports Scala 3.9.x LTS and later and requires JDK 21 or later. ## Installation @@ -10,6 +10,10 @@ Add the artifact for your Scala stack. The core is pulled in transitively, so yo ::: code-group +```scala [Direct] +"com.github.ghostdogpr" %% "sage-client-direct" % "@VERSION@" +``` + ```scala [Ox] "com.github.ghostdogpr" %% "sage-client-ox" % "@VERSION@" ``` @@ -30,52 +34,48 @@ Add the artifact for your Scala stack. The core is pulled in transitively, so yo "com.github.ghostdogpr" %% "sage-client-pekko" % "@VERSION@" ``` -```scala [Direct] -"com.github.ghostdogpr" %% "sage-client-direct" % "@VERSION@" -``` - ::: Import `sage.*` for commands and connection configuration. Import `sage.backend.*` for the client. Every backend uses these imports, but each one has a different dependency. ## Your first connection -A `SageClient` owns all connections to one server or cluster. Build it from a `SageConfig` with the resource type for your backend. Ox and Kyo use a scope, ZIO uses a `ZLayer`, Cats Effect uses a `Resource`, and Pekko and the direct client use a `use` block that closes the client when the program finishes. All six integrations provide the same commands. +A `SageClient` owns all connections to one server or cluster. Build it from a `SageConfig` with the resource type for your backend. The direct client and Pekko use a `use` block that closes the client when the program finishes, Ox and Kyo use a scope, ZIO uses a `ZLayer`, and Cats Effect uses a `Resource`. All six integrations provide the same commands. ::: code-group -```scala [Ox] -import ox.supervised - +```scala [Direct] import sage.* import sage.backend.* -@main def main(): Unit = - supervised { - val config = SageConfig( - topology = Topology.Standalone(Endpoint("localhost", 6379)) - ) - val client = SageClient.scoped(config) +@main def main(): Unit = { + val config = SageConfig( + topology = Topology.Standalone(Endpoint("localhost", 6379)) + ) + SageClient.use(config) { client => client.set("greeting", "hello") val greeting = client.get[String]("greeting") println(s"greeting=$greeting") // Some("hello") } +} ``` -```scala [Direct] +```scala [Ox] +import ox.supervised + import sage.* import sage.backend.* -@main def main(): Unit = { - val config = SageConfig( - topology = Topology.Standalone(Endpoint("localhost", 6379)) - ) - SageClient.use(config) { client => +@main def main(): Unit = + supervised { + val config = SageConfig( + topology = Topology.Standalone(Endpoint("localhost", 6379)) + ) + val client = SageClient.scoped(config) client.set("greeting", "hello") val greeting = client.get[String]("greeting") println(s"greeting=$greeting") // Some("hello") } -} ``` ```scala [ZIO] @@ -182,7 +182,7 @@ Transactions and blocking commands such as `WATCH`, `MULTI`, `EXEC`, and `BLPOP` ## A short tour -Choose your backend tab in the examples below. Ox returns values directly. The other backends use a for-comprehension over their effect type. Pekko uses `Future`, so each `for` needs an `ExecutionContext`. Every example assumes that `client` and the imports for the effect type are in scope. +Choose your backend tab in the examples below. The direct client and Ox return values directly. The other backends use a for-comprehension over their effect type. Pekko uses `Future`, so each `for` needs an `ExecutionContext`. Every example assumes that `client` and the imports for the effect type are in scope. ### Commands @@ -190,7 +190,7 @@ Method names match Redis commands and are grouped by family (strings, hashes, li ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.set("greeting", "hello") val greeting = client.get[String]("greeting") // Some("hello") client.incrBy("counter", 10) @@ -225,7 +225,7 @@ Put commands in a pipeline to send them in one round trip and receive a typed tu ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.set("pipe:a", "x") client.set("pipe:n", 10) val tuple = client.pipeline( @@ -255,7 +255,7 @@ A transaction runs a pipeline atomically with `MULTI` and `EXEC`. You can use `W ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.set("tx:n", 1) val result = client.transaction { tx => tx.watch("tx:n") @@ -290,18 +290,12 @@ The distinction is covered in [Pipelines & transactions](/pipelines-transactions ### Pub/Sub -Subscribing returns the backend's native stream type. Ox returns `Flow`, ZIO returns `ZStream`, Cats Effect returns an fs2 `Stream`, Kyo returns `Stream`, Pekko returns `Source`, and the direct client returns a blocking `Iterator` that you close to unsubscribe. Ending the stream or closing its scope unsubscribes. +Subscribing returns the backend's native stream type. The direct client returns a blocking `Iterator` that you close to unsubscribe, Ox returns `Flow`, ZIO returns `ZStream`, Cats Effect returns an fs2 `Stream`, Kyo returns `Stream`, and Pekko returns `Source`. Ending the stream or closing its scope unsubscribes. These examples publish immediately after subscribing, so they use the variant that waits for the server to confirm the subscription first. [Pub/Sub](/pubsub) explains when you need it. ::: code-group -```scala [Ox] -val news = client.subscribeScoped[String]("news") -(1 to 3).foreach(i => client.publish("news", s"item-$i")) -val messages = news.take(3).runToList() -``` - ```scala [Direct] val messages = Using.resource(client.subscribe[String]("news")) { news => (1 to 3).foreach(i => client.publish("news", s"item-$i")) @@ -309,6 +303,12 @@ val messages = Using.resource(client.subscribe[String]("news")) { news => } ``` +```scala [Ox] +val news = client.subscribeScoped[String]("news") +(1 to 3).foreach(i => client.publish("news", s"item-$i")) +val messages = news.take(3).runToList() +``` + ```scala [ZIO] ZIO.scoped { for { @@ -364,7 +364,7 @@ Opt a read into client-side caching per call. The first read fetches and caches; ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.set("cached:key", "v1") // first fetches and caches; second is a local hit val v1 = client.cached(Commands.get[String, String]("cached:key"), 1.minute) diff --git a/docs/index.md b/docs/index.md index c5cc99e8..b315ae37 100644 --- a/docs/index.md +++ b/docs/index.md @@ -4,7 +4,7 @@ layout: home hero: name: "Sage" text: "A Redis & Valkey client for Scala 3" - tagline: Native Scala clients for ZIO, Cats Effect, Kyo, Ox, Pekko, and plain blocking code. + tagline: Native Scala clients for plain blocking code, Ox, ZIO, Cats Effect, Kyo, and Pekko. image: light: /sage.svg dark: /sage-dark.svg @@ -19,7 +19,7 @@ hero: features: - title: Native backend types - details: Choose a ZIO, Cats Effect, Kyo, Ox, or Pekko artifact, each with its ecosystem's effect and stream types, or the direct artifact for plain blocking code. + details: Choose the direct artifact for plain blocking code, or an Ox, ZIO, Cats Effect, Kyo, or Pekko artifact with its ecosystem's effect and stream types. - title: Redis protocol in Scala details: Sage implements RESP3, commands, and codecs directly in Scala 3 instead of wrapping a Java client. - title: Redis and Valkey support diff --git a/docs/json.md b/docs/json.md index c41bb40b..d41f4723 100644 --- a/docs/json.md +++ b/docs/json.md @@ -25,7 +25,7 @@ A value you write is raw JSON text. The built-in `String` codec passes it throug ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.jsonSet("user:1", JsonPath.root, """{"name":"Ada","age":36,"tags":["a","b"]}""") val name = client.jsonGet[String]("user:1", JsonPath("$.name")) // Some("[\"Ada\"]") val age = client.jsonGet[String]("user:1", JsonPath("$.age")) // Some("[36]") @@ -71,7 +71,7 @@ Numbers, strings, booleans, arrays, and objects each have their commands. All of ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.jsonSet("doc", JsonPath.root, """{"n":1,"s":"ab","flag":false,"xs":[1,2,3]}""") client.jsonNumIncrBy("doc", JsonPath("$.n"), 5) // Vector(Some(6.0)) client.jsonStrAppend("doc", JsonPath("$.s"), "\"c\"") // Vector(Some(3)) diff --git a/docs/pipelines-transactions.md b/docs/pipelines-transactions.md index 023939bb..2d917751 100644 --- a/docs/pipelines-transactions.md +++ b/docs/pipelines-transactions.md @@ -8,7 +8,7 @@ A pipeline sends several `Command` values in one round trip and returns their re ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.set("pipe:a", "x") client.set("pipe:n", 10) val tuple = client.pipeline( @@ -46,7 +46,7 @@ By default, a pipeline fails as a whole if any command fails. Use `pipelineAttem ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.set("pipe:str", "hello") // INCR on a non-numeric string fails only at its own position; // the GET still succeeds @@ -84,7 +84,7 @@ A transaction runs a pipeline atomically with `MULTI`/`EXEC` on a temporary dedi ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.set("tx:n", 1) val result = client.transaction { tx => tx.watch("tx:n") diff --git a/docs/pubsub.md b/docs/pubsub.md index 53e773f3..b03c0fac 100644 --- a/docs/pubsub.md +++ b/docs/pubsub.md @@ -1,6 +1,6 @@ # Pub/Sub -Subscribing returns the backend's native stream type. Ox returns `Flow`, ZIO returns `ZStream`, Cats Effect returns an fs2 `Stream`, Kyo returns `Stream`, Pekko returns `Source`, and the direct client returns a blocking `Iterator` that you close to unsubscribe. Each message contains its channel and a payload decoded by a `ValueCodec`. Ending the stream or closing its scope unsubscribes. If the server refuses a subscribed name, for example with `NOPERM` from an ACL, the stream fails with that `ServerError`, including when a reconnect subscribes the name again. Temporary errors such as `BUSY` or `LOADING` do not end the stream; Sage subscribes again after a backoff. +Subscribing returns the backend's native stream type. The direct client returns a blocking `Iterator` that you close to unsubscribe, Ox returns `Flow`, ZIO returns `ZStream`, Cats Effect returns an fs2 `Stream`, Kyo returns `Stream`, and Pekko returns `Source`. Each message contains its channel and a payload decoded by a `ValueCodec`. Ending the stream or closing its scope unsubscribes. If the server refuses a subscribed name, for example with `NOPERM` from an ACL, the stream fails with that `ServerError`, including when a reconnect subscribes the name again. Temporary errors such as `BUSY` or `LOADING` do not end the stream; Sage subscribes again after a backoff. ## Classic channels @@ -8,12 +8,6 @@ Subscribing returns the backend's native stream type. Ox returns `Flow`, ZIO ret ::: code-group -```scala [Ox] -val news = client.subscribeScoped[String]("news") -(1 to 3).foreach(i => client.publish("news", s"item-$i")) -val messages = news.take(3).runToList() -``` - ```scala [Direct] val messages = Using.resource(client.subscribe[String]("news")) { news => (1 to 3).foreach(i => client.publish("news", s"item-$i")) @@ -21,6 +15,12 @@ val messages = Using.resource(client.subscribe[String]("news")) { news => } ``` +```scala [Ox] +val news = client.subscribeScoped[String]("news") +(1 to 3).foreach(i => client.publish("news", s"item-$i")) +val messages = news.take(3).runToList() +``` + ```scala [ZIO] ZIO.scoped { for { @@ -73,10 +73,10 @@ Pattern subscriptions return a pattern message that includes the matching glob. ::: tip Confirmed subscriptions The plain `subscribe` returns the stream immediately and registers the subscription when the stream first requests a message. A message published before registration finishes may be missed. Each backend provides a way to wait for the server's SUBSCRIBE confirmation first (also in `p`/`s` forms): +- **Direct**: `subscribe` waits for confirmation before it returns. Close the iterator to unsubscribe, for example with `scala.util.Using`. +- **Ox**: `subscribeScoped`, bound to the enclosing Ox scope. - **ZIO / Kyo**: `subscribeScoped`, a scoped effect. - **Cats Effect**: `subscribeResource`, a `Resource`. -- **Ox**: `subscribeScoped`, bound to the enclosing Ox scope. -- **Direct**: `subscribe` waits for confirmation before it returns. Close the iterator to unsubscribe, for example with `scala.util.Using`. - **Pekko**: plain `subscribe` returns a `Source` whose materialized `Future[Done]` completes once registered; await it before publishing. On a standalone or master-replica server, confirmation guarantees that the subscription is ready. In a cluster, this guarantee is best-effort. With the scoped and resource variants, the subscription remains active until the scope closes. On Pekko, ending the `Source` unsubscribes. diff --git a/docs/rate-limiting.md b/docs/rate-limiting.md index 4fb8d11e..8796ef4a 100644 --- a/docs/rate-limiting.md +++ b/docs/rate-limiting.md @@ -8,7 +8,7 @@ The limiter stores its state on the server rather than in process memory. The li ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] val limiter = client.rateLimiter[String](RateLimit.perSecond(100)) val decision = limiter.tryAcquire("user:42") val allowed = decision.isAllowed diff --git a/docs/streams.md b/docs/streams.md index f517a6d1..51a3a2b8 100644 --- a/docs/streams.md +++ b/docs/streams.md @@ -8,7 +8,7 @@ A stream is an append-only log. Each entry has an ID made from a millisecond tim ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] client.del("stream:orders") client.xAdd("stream:orders")(("item", "book"), ("qty", "2")) client.xAdd("stream:orders")(("item", "pen"), ("qty", "5")) @@ -37,7 +37,7 @@ A consumer group divides a stream's entries among several consumers without over ::: code-group -```scala [Ox · Direct] +```scala [Direct · Ox] // create the group reading from the start of the stream client.xGroupCreate( "stream:orders", @@ -84,15 +84,15 @@ The same entry can be delivered again after a crash or a failed handler, so make ::: code-group -```scala [Ox] -// runs inside a `supervised` scope; tails new entries forever +```scala [Direct] +// blocks the calling thread until it is interrupted client.xConsume[String, String]("workers", "w1", "stream:orders") { entry => println(s"got ${entry.id}: ${entry.fields}") } ``` -```scala [Direct] -// blocks the calling thread until it is interrupted +```scala [Ox] +// runs inside a `supervised` scope; tails new entries forever client.xConsume[String, String]("workers", "w1", "stream:orders") { entry => println(s"got ${entry.id}: ${entry.fields}") } diff --git a/examples/README.md b/examples/README.md index 10e9f4db..d85cbb9f 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,6 +1,6 @@ # Examples -These runnable examples cover every Sage backend. They use ZIO `Task`, Cats Effect `IO`, Ox direct style, Kyo computations, Pekko `Future` and Pekko Streams, or plain blocking calls with the direct client. The build does not publish this module, but CI compiles it. +These runnable examples cover every Sage backend. They use plain blocking calls with the direct client, Ox direct style, ZIO `Task`, Cats Effect `IO`, Kyo computations, or Pekko `Future` and Pekko Streams. The build does not publish this module, but CI compiles it. Import `sage.*` for commands and connection configuration. Import `sage..*` for the client. @@ -8,15 +8,15 @@ Import `sage.*` for commands and connection configuration. Import `sage.