arrow-fx-stm / arrow.fx.stm / STM

STM

interface STM

Consistent and safe concurrent state updates

Software transactional memory, or STM, is an abstraction for concurrent state modification. With STM one can write code that concurrently accesses state and that can easily be composed without exposing details of how it ensures safety guarantees. Programs running within an STM transaction will neither deadlock nor have race-conditions.

The api of STM is based on the haskell package stm and the implementation is based on the GHC implementation for fine-grained locks.

The base building blocks of STM are TVar’s and the primitives retry, orElse and catch.

STM Datastructures

There are several datastructures built on top of TVar’s already provided out of the box:

• TQueue: A transactional mutable queue
• TMVar: A mutable transactional variable that may be empty
• TSet, TMap: Transactional Set and Map
• TArray: Array of TVar’s
• TSemaphore: Transactional semaphore
• TVar: A transactional mutable variable

All of these structures (excluding TVar) are built upon TVar’s and the STM primitives and implementing other datastructures with STM can be done by composing the existing structures.

Reading and writing to concurrent state:

In order to modify transactional datastructures we have to be inside the STM context. This is achieved either by defining our functions with STM as the receiver or using stm to create lambda functions with STM as the receiver.

Running a transaction is then done using atomically:

import arrow.fx.stm.atomically
import arrow.fx.stm.TVar
import arrow.fx.stm.STM

//sampleStart
fun STM.transfer(from: TVar<Int>, to: TVar<Int>, amount: Int): Unit {
  withdraw(from, amount)
  deposit(to, amount)
}

fun STM.deposit(acc: TVar<Int>, amount: Int): Unit {
  val current = acc.read()
  acc.write(current + amount)
  // or the shorthand acc.modify { it + amount }
}

fun STM.withdraw(acc: TVar<Int>, amount: Int): Unit {
  val current = acc.read()
  if (current - amount >= 0) acc.write(current + amount)
  else throw IllegalStateException("Not enough money in the account!")
}
//sampleEnd

suspend fun main() {
  val acc1 = TVar.new(500)
  val acc2 = TVar.new(300)
  println("Balance account 1: ${acc1.unsafeRead()}")
  println("Balance account 2: ${acc2.unsafeRead()}")
  println("Performing transaction")
  atomically { transfer(acc1, acc2, 50) }
  println("Balance account 1: ${acc1.unsafeRead()}")
  println("Balance account 2: ${acc2.unsafeRead()}")
}

This example shows a banking service moving money from one account to the other with STM. Should the first account not have enough money we throw an exception. This code is guaranteed to never deadlock and to never produce an invalid state by committing after the read state has changed concurrently.

Note: A transaction that sees an invalid state (a TVar that was read has been changed concurrently) will restart and try again. This usually means we rerun the function entirely, therefore it is recommended to keep transactions small and to never use code that has side-effects inside. However no kotlin interface can actually keep you from doing side effects inside STM. Using side-effects such as access to resources, logging or network access comes with severe disadvantages:

• Transactions may be aborted at any time so accessing resources may never trigger finalizers
• Transactions may rerun an arbitrary amount of times before finishing and thus all effects will rerun.

Retrying manually

It is sometimes beneficial to manually abort the current transaction if, for example, an invalid state has been read. E.g. a TQueue had no elements to read. The aborted transaction will automatically restart once any previously accessed variable has changed.

This is achieved by the primitive retry:

import arrow.fx.stm.atomically
import arrow.fx.stm.TVar
import arrow.fx.stm.STM
import arrow.fx.coroutines.Environment
import arrow.fx.coroutines.ForkConnected
import arrow.fx.coroutines.seconds
import arrow.fx.coroutines.sleep

//sampleStart
fun STM.transfer(from: TVar<Int>, to: TVar<Int>, amount: Int): Unit {
  withdraw(from, amount)
  deposit(to, amount)
}

fun STM.deposit(acc: TVar<Int>, amount: Int): Unit {
  val current = acc.read()
  acc.write(current + amount)
  // or the shorthand acc.modify { it + amount }
}

fun STM.withdraw(acc: TVar<Int>, amount: Int): Unit {
  val current = acc.read()
  if (current - amount >= 0) acc.write(current + amount)
  else retry() // we now retry if there is not enough money in the account
  // this can also be achieved by using `check(current - amount >= 0); acc.write(it + amount)`
}
//sampleEnd

fun main() {
  Environment().unsafeRunSync {
    val acc1 = TVar.new(0)
    val acc2 = TVar.new(300)
    println("Balance account 1: ${acc1.unsafeRead()}")
    println("Balance account 2: ${acc2.unsafeRead()}")
    ForkConnected {
      println("Sending money - Searching")
      sleep(2.seconds)
      println("Sending money - Found some")
      atomically { acc1.write(100_000_000) }
    }
    println("Performing transaction")
    atomically {
      println("Trying to transfer")
      transfer(acc1, acc2, 50)
    }
    println("Balance account 1: ${acc1.unsafeRead()}")
    println("Balance account 2: ${acc2.unsafeRead()}")
  }
}

Here in this (silly) example we changed withdraw to use retry and thus wait until enough money is in the account, which after a few seconds just happens to be the case.

retry can be used to implement a lot of complex transactions and many datastructures like TMVar or TQueue use to to great effect.

Branching with orElse

orElse is another important primitive which allows a user to detect if a branch called retry and then use a fallback instead. If the fallback retries as well the whole transaction retries.

import arrow.fx.coroutines.Environment
import arrow.fx.stm.atomically
import arrow.fx.stm.TVar
import arrow.fx.stm.STM
import arrow.fx.stm.stm

//sampleStart
fun STM.transaction(v: TVar<Int>): Int? =
  stm {
    val result = v.read()
    check(result in 0..10)
    result
  } orElse { null }
//sampleEnd

fun main() {
  Environment().unsafeRunSync {
    val v = TVar.new(100)
    println("Value is ${v.unsafeRead()}")
    atomically { transaction(v) }
      .also { println("Transaction returned $it") }
    println("Set value to 5")
    println("Value is ${v.unsafeRead()}")
    atomically { v.write(5) }
    atomically { transaction(v) }
      .also { println("Transaction returned $it") }
  }
}

This example uses stm which is a helper just like the stdlib function suspend to ease use of an infix function like orElse. In this transaction, when the value inside the variable is not in the correct range, the transaction retries (due to check calling retry). If it is in the correct range it simply returns the value. orElse here intercepts a call to retry and executes the alternative which simply returns null.

Exceptions

Throwing inside STM will let the exception bubble up to either a catch handler or to atomically which will rethrow it.

Note: Using try {...} catch (e: Exception) {...} is not encouraged because any state change inside try will not be undone when an exception occurs! The recommended way of catching exceptions is to use catch which properly rolls back the transaction!

Further reading:

• Composable memory transactions, by Tim Harris, Simon Marlow, Simon Peyton Jones, and Maurice Herlihy, in ACM Conference on Principles and Practice of Parallel Programming 2005.

Functions

Extension Functions