Txn
Overview
Txn
is a monad which describes transactions involving TVar
s. It is executed via
STM#commit
:
import cats.implicits._
import cats.effect.IO
import cats.effect.unsafe.implicits.global
import io.github.timwspence.cats.stm.STM
val stm = STM.runtime[IO].unsafeRunSync()
// stm: STM[IO] = io.github.timwspence.cats.stm.STM$Make$$anon$1$$anon$2@53d4118e
import stm._
val prog: IO[(Int, Int)] = for {
to <- stm.commit(TVar.of(0))
from <- stm.commit(TVar.of(100))
_ <- stm.commit {
for {
balance <- from.get
_ <- from.modify(_ - balance)
_ <- to.modify(_ + balance)
} yield ()
}
v <- stm.commit((to.get, from.get).tupled)
} yield v
// prog: IO[(Int, Int)] = FlatMap(
// ioe = FlatMap(
// ioe = Uncancelable(
// body = cats.effect.IO$$$Lambda$11386/948436343@5848bd71,
// event = cats.effect.tracing.TracingEvent$StackTrace
// ),
// f = io.github.timwspence.cats.stm.STM$Make$$anon$1$$anon$2$$Lambda$11387/1682669306@2d463234,
// event = cats.effect.tracing.TracingEvent$StackTrace
// ),
// f = <function1>,
// event = cats.effect.tracing.TracingEvent$StackTrace
// )
val result = prog.unsafeRunSync()
// result: (Int, Int) = (100, 0)
Retries
STM#commit
supports the concept of retries, which can be introduced via
STM.check
:
import cats.effect.IO
import cats.effect.unsafe.implicits.global
import io.github.timwspence.cats.stm.STM
val stm = STM.runtime[IO].unsafeRunSync()
// stm: STM[IO] = io.github.timwspence.cats.stm.STM$Make$$anon$1$$anon$2@40df0df1
import stm._
val to = stm.commit(TVar.of(1)).unsafeRunSync()
// to: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@33742c05
val from = stm.commit(TVar.of(0)).unsafeRunSync()
// from: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@2ecabc2f
val txn: IO[Unit] = stm.commit {
for {
balance <- from.get
_ <- stm.check(balance > 100)
_ <- from.modify(_ - 100)
_ <- to.modify(_ + 100)
} yield ()
}
// txn: IO[Unit] = FlatMap(
// ioe = Uncancelable(
// body = cats.effect.IO$$$Lambda$11386/948436343@3281fc50,
// event = cats.effect.tracing.TracingEvent$StackTrace
// ),
// f = io.github.timwspence.cats.stm.STM$Make$$anon$1$$anon$2$$Lambda$11387/1682669306@a09df74,
// event = cats.effect.tracing.TracingEvent$StackTrace
// )
txn.unsafeRunSync()
will block until the transaction succeeds (or throws an
exception!). Internally, this is implemented by keeping track of which TVar
s are
involved in a transaction and retrying any pending transactions every time a TVar
is committed.
OrElse
STM.orElse
is built on top of the retry logic and allows you to attempt an
alternative action if the first retries:
import cats.effect.IO
import cats.effect.unsafe.implicits.global
import io.github.timwspence.cats.stm.STM
val stm = STM.runtime[IO].unsafeRunSync()
// stm: STM[IO] = io.github.timwspence.cats.stm.STM$Make$$anon$1$$anon$2@188fde34
import stm._
val to = stm.commit(TVar.of(1)).unsafeRunSync()
// to: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@63df32ce
val from = stm.commit(TVar.of(0)).unsafeRunSync()
// from: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@5e012199
val transferHundred: Txn[Unit] = for {
b <- from.get
_ <- stm.check(b > 100)
_ <- from.modify(_ - 100)
_ <- to.modify(_ + 100)
} yield ()
// transferHundred: Txn[Unit] = Bind(
// txn = Get(tvar = io.github.timwspence.cats.stm.STMLike$TVar@5e012199),
// f = <function1>
// )
val transferRemaining: Txn[Unit] = for {
balance <- from.get
_ <- from.modify(_ - balance)
_ <- to.modify(_ + balance)
} yield ()
// transferRemaining: Txn[Unit] = Bind(
// txn = Get(tvar = io.github.timwspence.cats.stm.STMLike$TVar@5e012199),
// f = <function1>
// )
val txn = for {
_ <- transferHundred.orElse(transferRemaining)
f <- from.get
t <- to.get
} yield f -> t
// txn: Txn[(Int, Int)] = Bind(
// txn = OrElse(
// txn = Bind(
// txn = Get(tvar = io.github.timwspence.cats.stm.STMLike$TVar@5e012199),
// f = <function1>
// ),
// fallback = Bind(
// txn = Get(tvar = io.github.timwspence.cats.stm.STMLike$TVar@5e012199),
// f = <function1>
// )
// ),
// f = <function1>
// )
val result = stm.commit(txn).unsafeRunSync()
// result: (Int, Int) = (0, 1)
Aborting
Transactions can be aborted via STM.abort
:
import cats.effect.IO
import cats.effect.unsafe.implicits.global
import io.github.timwspence.cats.stm.STM
val stm = STM.runtime[IO].unsafeRunSync()
// stm: STM[IO] = io.github.timwspence.cats.stm.STM$Make$$anon$1$$anon$2@1a37b752
import stm._
val to = stm.commit(TVar.of(1)).unsafeRunSync()
// to: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@7c2fb45e
val from = stm.commit(TVar.of(0)).unsafeRunSync()
// from: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@66fcb057
val txn = for {
balance <- from.get
_ <- if (balance < 100)
stm.abort(new RuntimeException("Balance must be at least 100"))
else
stm.unit
_ <- from.modify(_ - 100)
_ <- to.modify(_ + 100)
} yield ()
// txn: Txn[Unit] = Bind(
// txn = Get(tvar = io.github.timwspence.cats.stm.STMLike$TVar@66fcb057),
// f = <function1>
// )
val result = stm.commit(txn).attempt.unsafeRunSync()
// result: Either[Throwable, Unit] = Left(
// value = java.lang.RuntimeException: Balance must be at least 100
// )
Note that aborting a transaction will not modify any of the TVar
s involved.