Txn

Overview

Txn is a monad which describes transactions involving TVars. 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@4225a3fe
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$10480/0x0000000802e19840@73ad2361,
//       event = cats.effect.tracing.TracingEvent$StackTrace
//     ),
//     f = io.github.timwspence.cats.stm.STM$Make$$anon$1$$anon$2$$Lambda$10481/0x0000000802e1a840@4ddf2ca7,
//     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@56db23ac
import stm._

val to   = stm.commit(TVar.of(1)).unsafeRunSync()
// to: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@161564c9
val from = stm.commit(TVar.of(0)).unsafeRunSync()
// from: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@574d0e98

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$10480/0x0000000802e19840@6c91c727,
//     event = cats.effect.tracing.TracingEvent$StackTrace
//   ),
//   f = io.github.timwspence.cats.stm.STM$Make$$anon$1$$anon$2$$Lambda$10481/0x0000000802e1a840@106a657a,
//   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 TVars 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@620190a6
import stm._

val to   = stm.commit(TVar.of(1)).unsafeRunSync()
// to: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@6a2519b0
val from = stm.commit(TVar.of(0)).unsafeRunSync()
// from: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@5ae1b96b

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@5ae1b96b),
//   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@5ae1b96b),
//   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@5ae1b96b),
//       f = <function1>
//     ),
//     fallback = Bind(
//       txn = Get(tvar = io.github.timwspence.cats.stm.STMLike$TVar@5ae1b96b),
//       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@3e2fbda0
import stm._

val to   = stm.commit(TVar.of(1)).unsafeRunSync()
// to: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@17b7cc0
val from = stm.commit(TVar.of(0)).unsafeRunSync()
// from: TVar[Int] = io.github.timwspence.cats.stm.STMLike$TVar@4f2d9f6f

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@4f2d9f6f),
//   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 TVars involved.