Sharing data between threads

Atomic values, locks, and concurrent maps.

On this page

Data that several threads use must be marked as such. The compiler then checks that every change is safe: atomic T for data that is mostly read, Lock[T] for data that threads change.

Atomic values#

Converting a value to atomic T makes a copy that threads can share. The original stays a normal object:

Valk

use valk.thread

class Config {
    name: String
    limit: uint
}

fn main() {
    let config = Config { name: "api", limit: 10 }
    let shared_config: atomic Config = config   // a copy
    config.name = "changed"

    let worker = thread.Thread[uint].start(fn() uint {
        shared_config.limit = 20                // numbers can be assigned
        return shared_config.limit
    }) ! panic("thread")
    println(worker.await())     // 20
    println(shared_config.name) // api
}

Through an atomic view you can:

read every property;
assign properties that hold a number, bool, enum or object (an object is copied in);
assign elements of an array of numbers, like values[i] = x.

Other changes, such as appending to an array or changing a string property, need a Lock. clone x turns atomic data back into an ordinary copy. A literal written in place, like let c: atomic Config = Config { ... }, is not copied.

Locks#

Lock[T] holds an object that threads change. The value is only reachable inside a lock block, which holds the lock until the block ends:

Valk

use valk.thread

class Stats {
    count: uint (0)
    names: Array[String] (.{})
}

fn main() {
    let stats: Lock[Stats] = .new(Stats {}) !!
    let group = thread.ThreadGroup[void].new()
    each 0..4 as i {
        group.start(fn() {
            lock stats as s {
                s.count++
                s.names.append("worker")
            }
        }) ! panic("thread")
    }
    while group.has_pending() : group.await_next() ! break

    lock stats as s {
        println(s.count)        // 4
        println(s.names.length) // 4
    }
}

Inside the block s is a locked Stats. It cannot be stored, returned or captured outside the block.
Values stored into locked data from outside are copied in.
Use clone to take data out of the block.
The lock is released on every exit, including return and errors. break and continue cannot leave a lock block.
Locking the same Lock again inside its own block deadlocks. Keep blocks short.

Lock, Channel, CancelToken, WaitGroup, Semaphore and ConcurrentMap are already shareable, so they need no atomic in front.

Shared globals#

A shared global is one value for all threads, read as atomic T. Put mutable state in a Lock:

Valk

use valk.thread

class Registry {
    names: Array[String] (.{})
}

shared registry: Lock[Registry] (Lock[Registry].new(Registry{}) !!)

fn main() {
    let worker = thread.start(fn() {
        lock registry as r : r.names.append("from thread")
    }) ! panic("thread")
    worker.await()
    lock registry as r : println(r.names.length) // 1
}

Concurrent maps#

sync.ConcurrentMap[K, V] is a map that threads change at the same time without one big lock. Keys and values are numbers, bools, strings or atomic objects:

Valk

use valk.sync

fn main() {
    let hits = sync.ConcurrentMap[String, uint].new() !!
    hits.set("home", 1)
    hits.update("home", fn(old: ?uint) uint { return (old ?? 0) + 1 })
    println(hits.get("home") ?? 0)  // 2
    println(hits.has("about"))      // false

    each hits.snapshot() as count, page {
        println("%page: %count")    // home: 2
    }
}

Method	Does
`get(key)`	the value, or `null`
`set(key, value)` / `remove(key)`	store or remove one entry
`update(key, fn)`	replace a value based on the old one, atomically
`has(key)`, `length`, `clear()`	as on `HashMap`
`snapshot()`	an ordinary `HashMap` copy of all entries

Limit concurrent work#

A Semaphore lets at most a set number of tasks in at once. acquire() waits for a free permit and release() gives it back:

Valk

use valk.sync
use valk.time

fn main() {
    let slots = sync.Semaphore.new(2) ! panic("semaphore")
    let tasks = Array[co()]{}
    each 0..5 as i {
        tasks.append(co fn() {
            slots.acquire() ! return
            defer slots.release()
            time.sleep_ms(5)
        }())
    }
    each tasks as task : await task
    println(slots.available) // 2
}

try_acquire() takes a permit only when one is free.

Mutex#

core.Mutex is a plain lock without data. Waiting for it only pauses the current coroutine. Prefer Lock[T], which ties the data to its lock:

Valk

use valk.core

fn main() {
    let mutex = core.Mutex.new() !!
    mutex.lock()
    defer mutex.unlock()
    println("in the critical section")
}

Atomic numbers#

For a single counter, an atomic operation is cheaper than a lock. See Atomics.

View page source ↗