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 kotlinx.coroutines.runBlocking
import kotlinx.coroutines.async
import kotlinx.coroutines.delay

//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(): Unit = runBlocking {
val acc1 = TVar.new(0)
val acc2 = TVar.new(300)
println("Balance account 1: ${acc1.unsafeRead()}")
println("Balance account 2: ${acc2.unsafeRead()}")
async {
println("Sending money - Searching")
delay(2000)
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 kotlinx.coroutines.runBlocking
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(): Unit = runBlocking {
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:

Functions

Link copied to clipboard
open fun TSemaphore.acquire()

Acquire 1 permit from a TSemaphore.

open fun TSemaphore.acquire(n: Int)

Acquire n permit from a TSemaphore.

Link copied to clipboard
open fun TSemaphore.available(): Int

Returns the currently available number of permits in a TSemaphore.

Link copied to clipboard
abstract fun <A> catch(f: STM.() -> A, onError: STM.(Throwable) -> A): A

Run f and handle any exception thrown with onError.

Link copied to clipboard
open fun <A> TQueue<A>.flush(): List<A>

Drains all entries of a TQueue into a single list.

Link copied to clipboard
open fun <A, B> TArray<A>.fold(init: B, f: (B, A) -> B): B

Fold a TArray to a single value.

Link copied to clipboard
open operator fun <A> TArray<A>.get(i: Int): A

Read a variable from the TArray.

open operator fun <K, V> TMap<K, V>.get(k: K): V?

Alias of STM.lookup

Link copied to clipboard
open fun <A> TSet<A>.insert(a: A)

Adds an element to the set.

open fun <K, V> TMap<K, V>.insert(k: K, v: V)

Add a key value pair to the map

Link copied to clipboard
open fun <A> TMVar<A>.isEmpty(): Boolean

Check if a TMVar is empty. This function never retries.

open fun <A> TQueue<A>.isEmpty(): Boolean

Check if a TQueue is empty.

Link copied to clipboard
open fun <A> TMVar<A>.isNotEmpty(): Boolean

Check if a TMVar is not empty. This function never retries.

open fun <A> TQueue<A>.isNotEmpty(): Boolean

Check if a TQueue is not empty.

Link copied to clipboard
open fun <K, V> TMap<K, V>.lookup(k: K): V?

Lookup a value at the specific key k

Link copied to clipboard
open fun <K, V> TMap<K, V>.member(k: K): Boolean

Check if a key k is in the map

open fun <A> TSet<A>.member(a: A): Boolean

Check if an element is already in the set

Link copied to clipboard
open fun <A> TVar<A>.modify(f: (A) -> A)

Modify the value of a TVar

Link copied to clipboard
open fun <A> newTVar(a: A): TVar<A>

Create a new TVar inside a transaction, because TVar.new is not possible inside STM transactions.

Link copied to clipboard
abstract infix fun <A> STM.() -> A.orElse(other: STM.() -> A): A

Run the given transaction and fallback to the other one if the first one calls retry.

Link copied to clipboard
open fun <A> TQueue<A>.peek(): A

Read the front element of a TQueue without removing it.

Link copied to clipboard
open operator fun <K, V> TMap<K, V>.plusAssign(kv: Pair<K, V>)

Add a key value pair to the map

open operator fun <A> TQueue<A>.plusAssign(a: A)

Append an element to the TQueue. Alias for STM.write.

open operator fun <A> TSet<A>.plusAssign(a: A)

Adds an element to the set. Alias of STM.insert.

Link copied to clipboard
open fun <A> TMVar<A>.put(a: A)

Put a value into an empty TMVar.

Link copied to clipboard
open fun <A> TMVar<A>.read(): A

Read a value from a TMVar without removing it.

open fun <A> TQueue<A>.read(): A

Remove the front element from the TQueue or retry if the TQueue is empty.

abstract fun <A> TVar<A>.read(): A

Read the value from a TVar.

Link copied to clipboard
open fun TSemaphore.release()

Release a permit back to the TSemaphore.

open fun TSemaphore.release(n: Int)

Release n permits back to the TSemaphore.

Link copied to clipboard
open fun <K, V> TMap<K, V>.remove(k: K)

Remove a key value pair from a map

open fun <A> TSet<A>.remove(a: A)

Remove an element from the set.

Link copied to clipboard
open fun <A> TQueue<A>.removeAll(pred: (A) -> Boolean)

Filter a TQueue, removing all elements for which pred returns false.

Link copied to clipboard
abstract fun retry(): Nothing

Abort and retry the current transaction.

Link copied to clipboard
open operator fun <A> TArray<A>.set(i: Int, a: A)

Set a variable in the TArray.

open operator fun <K, V> TMap<K, V>.set(k: K, v: V)

Alias for STM.insert

Link copied to clipboard
open fun <A> TQueue<A>.size(): Int

Return the current number of elements in a TQueue

Link copied to clipboard
open fun <A> TMVar<A>.swap(a: A): A

Swap the content of a TMVar or retry if it is empty.

open fun <A> TVar<A>.swap(a: A): A

Swap the content of the TVar

Link copied to clipboard
open fun <A> TMVar<A>.take(): A

Read the value from a TMVar and empty it.

Link copied to clipboard
open fun <A> TArray<A>.transform(f: (A) -> A)

Modify each element in a TArray by applying f.

Link copied to clipboard
open fun TSemaphore.tryAcquire(): Boolean
open fun TSemaphore.tryAcquire(n: Int): Boolean

Like TSemaphore.acquire except that it returns whether or not acquisition was successful.

Link copied to clipboard
open fun <A> TQueue<A>.tryPeek(): A?

Same as TQueue.peek except it returns null if the TQueue is empty.

Link copied to clipboard
open fun <A> TMVar<A>.tryPut(a: A): Boolean

Same as TMVar.put except that it returns true or false if was successful or it retried.

Link copied to clipboard
open fun <A> TMVar<A>.tryRead(): A?

Same as TMVar.read except that it returns null if the TMVar is empty and thus never retries.

open fun <A> TQueue<A>.tryRead(): A?

Same as TQueue.read except it returns null if the TQueue is empty.

Link copied to clipboard
open fun <A> TMVar<A>.tryTake(): A?

Same as TMVar.take except it returns null if the TMVar is empty and thus never retries.

Link copied to clipboard
open fun <K, V> TMap<K, V>.update(k: K, fn: (V) -> V)

Update a value at a key if it exists.

Link copied to clipboard
open fun <A> TQueue<A>.write(a: A)

Append an element to the TQueue.

abstract fun <A> TVar<A>.write(a: A)

Set the value of a TVar.

Link copied to clipboard
open fun <A> TQueue<A>.writeFront(a: A)

Prepend an element to the TQueue.

Extensions

Link copied to clipboard
inline fun <A> STM.alterHamtWithHash(    hamt: Hamt<A>,     hash: Int,     test: (A) -> Boolean,     fn: (A?) -> A?): Boolean
Link copied to clipboard
fun STM.check(b: Boolean)

Retry if b is false otherwise does nothing.

Link copied to clipboard
fun <A> STM.clearHamt(hamt: Hamt<A>)
Link copied to clipboard
inline fun <A> STM.lookupHamtWithHash(    hmt: Hamt<A>,     hash: Int,     test: (A) -> Boolean): A?
Link copied to clipboard
fun <A> STM.newEmptyTMVar(): TMVar<A>
Link copied to clipboard
fun <A> STM.newHamt(): Hamt<A>
Link copied to clipboard
fun <A> STM.newTArray(size: Int, f: (Int) -> A): TArray<A>
fun <A> STM.newTArray(size: Int, a: A): TArray<A>
fun <A> STM.newTArray(vararg arr: A): TArray<A>
fun <A> STM.newTArray(xs: Iterable<A>): TArray<A>
Link copied to clipboard
fun <K, V> STM.newTMap(fn: (K) -> Int): TMap<K, V>
fun <K, V> STM.newTMap(): TMap<K, V>
Link copied to clipboard
fun <A> STM.newTMVar(a: A): TMVar<A>
Link copied to clipboard
fun <A> STM.newTQueue(): TQueue<A>
Link copied to clipboard
fun STM.newTSem(initial: Int): TSemaphore
Link copied to clipboard
fun <A> STM.newTSet(fn: (A) -> Int): TSet<A>
fun <A> STM.newTSet(): TSet<A>
Link copied to clipboard
fun <A> STM.pair(    depth: Int,     hash1: Int,     branch1: Branch<A>,     hash2: Int,     branch2: Branch<A>): Hamt<A>