Concurrency

Spawning Concurrent Tasks

V's model of concurrency is similar to Go's.

go foo() runs foo() concurrently in a lightweight thread managed by the V runtime.

spawn foo() runs foo() concurrently in a different thread:

import math fn p(a f64, b f64) { // ordinary function without return value c := math.sqrt(a * a + b * b) println(c) } fn main() { spawn p(3, 4) // p will be run in parallel thread // It can also be written as follows // spawn fn (a f64, b f64) { // c := math.sqrt(a * a + b * b) // println(c) // }(3, 4) }

If a spawned thread's handle is discarded, including inside a discarded array or struct, V detaches the thread. Keep its handle and call wait() when the result or completion matters. The detached thread releases an owned return value after its function finishes. If the return value is a thread handle, it joins that thread; a returned closure releases its captured context.

[!NOTE] Threads rely on the machine's CPU (number of cores/threads). Be aware that OS threads spawned with spawn have limitations in regard to concurrency, including resource overhead and scalability issues, and might affect performance in cases of high thread count.

Sometimes it is necessary to wait until a parallel thread has finished. This can be done by assigning a handle to the started thread and calling the wait() method to this handle later:

import math fn p(a f64, b f64) { // ordinary function without return value c := math.sqrt(a * a + b * b) println(c) // prints `5` } fn main() { h := spawn p(3, 4) // p() runs in parallel thread h.wait() // p() has definitely finished }

This approach can also be used to get a return value from a function that is run in a parallel thread. There is no need to modify the function itself to be able to call it concurrently.

import math { sqrt } fn get_hypot(a f64, b f64) f64 { // ordinary function returning a value c := sqrt(a * a + b * b) return c } fn main() { g := spawn get_hypot(54.06, 2.08) // spawn thread and get handle to it h1 := get_hypot(2.32, 16.74) // do some other calculation here h2 := g.wait() // get result from spawned thread println('Results: ${h1}, ${h2}') // prints `Results: 16.9, 54.1` }

Discarding a spawned thread's handle, including through dump(spawn ...), detaches the thread. Keep the handle when you need to call .wait().

If there is a large number of tasks, it might be easier to manage them using an array of threads.

import time fn task(id int, duration int) { println('task ${id} begin') time.sleep(duration * time.millisecond) println('task ${id} end') } fn main() { mut threads := []thread{} threads << spawn task(1, 500) threads << spawn task(2, 900) threads << spawn task(3, 100) threads.wait() println('done') } // Output: // task 1 begin // task 2 begin // task 3 begin // task 3 end // task 1 end // task 2 end // done

Additionally for threads that return the same type, calling wait() on the thread array will return all computed values. Arrays whose elements are aliases of thread handles support wait() as well, including aliases of thread and thread T. The elements must be handles themselves; arrays of pointers to handles cannot be joined.

fn expensive_computing(i int) int { return i * i } fn main() { mut threads := []thread int{} for i in 1 .. 10 { threads << spawn expensive_computing(i) } // Join all tasks r := threads.wait() println('All jobs finished: ${r}') } // Output: All jobs finished: [1, 4, 9, 16, 25, 36, 49, 64, 81]

When the spawned function returns ?T or !T, waiting on the thread array returns ?[]T or ![]T. That means the batch can be handled with or, ?, or ! the same way as a single thread handle.

fn maybe_square(i int) !int { if i < 0 { return error('negative input') } return i * i } fn main() { mut threads := []thread !int{} for i in 1 .. 4 { threads << spawn maybe_square(i) } values := threads.wait() or { panic(err) } println(values) }

Channels

Channels are the preferred way to communicate between threads. They allow threads to exchange data safely without requiring explicit locking. V's channels are similar to those in Go, enabling you to push objects into a channel on one end and pop objects from the other. Channels can be buffered or unbuffered, and you can use the select statement to monitor multiple channels simultaneously.

Syntax and Usage

Channels are declared with the type chan objtype. You can optionally specify a buffer length using the cap field:

ch := chan int{} // unbuffered - "synchronous" ch2 := chan f64{cap: 100} // buffered with a capacity of 100

Channels do not have to be declared as mut. The buffer length is not part of the type but a field of the individual channel object. Channels can be passed to threads like normal variables:

import time fn worker(ch chan int) { for i in 0 .. 5 { ch <- i // push values into the channel } } fn clock(ch chan int) { for i in 0 .. 5 { time.sleep(1 * time.second) println('Clock tick') ch <- (i + 1000) // push a value into the channel } ch.close() // close the channel when done } fn main() { ch := chan int{cap: 5} spawn worker(ch) spawn clock(ch) for { value := <-ch or { // receive/pop values from the channel println('Channel closed') break } println('Received: ${value}') } }

Buffered Channels

Buffered channels allow you to push multiple items without blocking, as long as the buffer is not full:

ch := chan string{cap: 2} ch <- 'hello' ch <- 'world' // ch <- '!' // This would block because the buffer is full println(<-ch) // "hello" println(<-ch) // "world"

Closing Channels

A channel can be closed to indicate that no further objects can be pushed. Any attempt to do so will then result in a runtime panic (with the exception of select and try_push() - see below). Attempts to pop will return immediately if the associated channel has been closed and the buffer is empty. This situation can be handled using an or {} block (see Handling options/results). An optional IError can be passed to close(err) so receive operations that use or {} or ? can distinguish an error termination from a normal close.

ch := chan int{} ch2 := chan f64{} // ... ch.close() // ... m := <-ch or { println('channel has been closed') } // propagate error y := <-ch2 ? ch2.close(error('worker failed')) z := <-ch2 or { println(err.msg()) 0.0 }

Note: buffered channels can be closed while they have unread values in them. The buffered values can be retrieved, even after the closing:

ich := chan int{cap: 5} for i in 0 .. 5 { ich <- i } for _ in 0 .. 2 { x := <-ich or { break } eprintln('>> loop 0..2 | x: ${x} | ich.closed: ${ich.closed}') } ich.close() for { x := <-ich or { break } eprintln('>> final loop | x: ${x} | ich.closed: ${ich.closed}') }

... will produce:

>>  loop 0..2 | x: 0 | ich.closed: false
>>  loop 0..2 | x: 1 | ich.closed: false
>> final loop | x: 2 | ich.closed: true
>> final loop | x: 3 | ich.closed: true
>> final loop | x: 4 | ich.closed: true

Note: reading from the .closed field of the channel in the example, is done just for clarity of illustration. The recommended way to pop values from the channel is with: x := <-ich or { break } in a for loop, which will cleanly break out of the loop, when the channel is closed and empty. Avoid manually checking, whether the channel was closed or not, because that can introduce data races, if you are not careful.

Channel Select

Assertion messages inside a select branch can use values declared or received in that branch, including arrays and structs converted to strings.

The select command allows monitoring several channels at the same time without noticeable CPU load. It consists of a list of possible transfers and associated branches of statements - similar to the match command:

import time fn main() { ch := chan f64{} ch2 := chan f64{} ch3 := chan f64{} mut b := 0.0 c := 1.0 // ... setup spawn threads that will send on ch/ch2 spawn fn (the_channel chan f64) { time.sleep(5 * time.millisecond) the_channel <- 1.0 }(ch) spawn fn (the_channel chan f64) { time.sleep(1 * time.millisecond) the_channel <- 1.0 }(ch2) spawn fn (the_channel chan f64) { _ := <-the_channel }(ch3) select { a := <-ch { // do something with `a` eprintln('> a: ${a}') } b = <-ch2 { // do something with predeclared variable `b` eprintln('> b: ${b}') } ch3 <- c { // do something if `c` was sent time.sleep(5 * time.millisecond) eprintln('> c: ${c} was send on channel ch3') } 500 * time.millisecond { // do something if no channel has become ready within 0.5s eprintln('> more than 0.5s passed without a channel being ready') } } eprintln('> done') }

select receive branches support fixed-array payloads, including fixed-array aliases, with GCC and Clang.

The timeout branch is optional. If it is absent select waits for an unlimited amount of time. It is also possible to proceed immediately if no channel is ready in the moment select is called by adding an else { ... } branch. else and <timeout> are mutually exclusive. In statement form, else also runs when every channel operation is unavailable because the channels are already closed. To drain buffered values from a closed channel, use a plain receive with or {} instead of select ... else.

The select command can be used as an expression of type bool that becomes false if all channels are closed:

if select { ch <- a { // ... } } { // channel was open } else { // channel is closed }

Special Channel Features

For special purposes there are some builtin fields and methods:

ch := chan int{cap: 2} println(ch.try_push(42)) // `.success` if pushed, `.not_ready` if full, `.closed` if closed println(ch.len) // Number of items in the buffer println(ch.cap) // Buffer capacity println(ch.closed) // Whether the channel is closed
struct Abc { x int } a := 2.13 ch := chan f64{cap: 1} // buffered so `try_push()` can succeed without a waiting receiver res := ch.try_push(a) // try to perform `ch <- a` println(res) l := ch.len // number of elements in queue c := ch.cap // maximum queue length is_closed := ch.closed // bool flag - has `ch` been closed println(l) println(c) mut b := Abc{} ch2 := chan Abc{} res2 := ch2.try_pop(mut b) // try to perform `b = <-ch2`

The try_push/pop() methods will return immediately with one of the results .success, .not_ready or .closed - dependent on whether the object has been transferred or the reason why not. On an unbuffered channel, try_push() will return .not_ready unless a receiver is ready at the same time. Usage of these methods and fields in production is not recommended - algorithms based on them are often subject to race conditions. Especially .len and .closed should not be used to make decisions. Use or branches, error propagation or select instead (see Syntax and Usage and Channel Select above).

Shared Objects

Data can be exchanged between a thread and the calling thread via a shared variable. Such variables should be created as shared and passed to the thread as such, too. The underlying struct contains a hidden mutex that allows locking concurrent access using rlock for read-only and lock for read/write access.

Note: Shared variables must be structs, arrays or maps.

Example of Shared Objects

struct Counter { mut: value int } fn (shared counter Counter) increment() { lock counter { counter.value += 1 println('Incremented to: ${counter.value}') } } fn main() { shared counter := Counter{} spawn counter.increment() spawn counter.increment() rlock counter { println('Final value: ${counter.value}') } }

Thread-safe maps

A single operation on a shared map does not need a lock/rlock block: it locks the map by itself, for the duration of that one operation. So a shared map can be used as a thread-safe map:

import sync fn count(shared words map[string]int, text string, mut wg sync.WaitGroup) { for word in text.split(' ') { words[word]++ // locks `words` while it is updated } wg.done() } fn main() { shared words := map[string]int{} mut wg := sync.new_waitgroup() wg.add(2) spawn count(shared words, 'a b a', mut wg) spawn count(shared words, 'b c', mut wg) wg.wait() println(words['a']) // 2, read with an `rlock` println('c' in words) // true println(words.len) // 3 }

Statements that change the map (m[k] = v, m[k] += v, m[k]++, m[k] << v, m.delete(k) and m.clear()) take a lock. The whole statement runs with the map locked, so m[k] = m[k] * 2 is atomic too. Reads (m[k], m[k] or { ... }, k in m, m.len, m.keys(), m.values() and m.clone()) take an rlock. This works for shared variables, parameters, globals and struct fields.

Operations that must happen together still need an explicit block, and so do iterating over the map, if v := m[k] { and statements that use two shared maps:

shared m := map[string]int{} // check and insert in one step lock m { if 'a' !in m { m['a'] = 0 } } rlock m { for k, v in m { println('${k}: ${v}') } }

Difference Between Channels and Shared Objects

Purpose:

  • Channels: Used for message passing between threads, ensuring safe communication.
  • Shared objects: Used for direct data sharing and modification between threads.

Synchronization:

  • Channels: Implicit (via channel operations)
  • Shared objects: Explicit (via rlock/lock blocks)

Race Detector

A data race happens when two threads access the same memory at the same time, and at least one of the accesses is a write, without a channel, a lock/rlock block, a sync primitive or an atomic operation ordering them. Data races are hard to find: they depend on timing, and the program usually works until it does not.

Like Go, V has a race detector built in. Build or run a program, or its tests, with -race:

v -race run main.v
v -race test .

Consider this program, where two threads increment one counter without synchronization:

struct Counter { mut: n int } fn inc(mut c Counter) { for _ in 0 .. 1000 { c.n++ } } fn main() { mut c := &Counter{} t1 := spawn inc(mut c) t2 := spawn inc(mut c) t1.wait() t2.wait() println(c.n) }

v -race run main.v reports the race with the V source positions of both accesses, of the allocation, and of the spawn calls that started the threads:

==================
WARNING: ThreadSanitizer: data race (pid=55780)
  Write of size 8 at 0x000109800380 by thread T2:
    #0 inc main.v:8
    #1 inc_args_thread_wrapper src.c:2438

  Previous write of size 8 at 0x000109800380 by thread T1:
    #0 inc main.v:8
    #1 inc_args_thread_wrapper src.c:2438

  Location is heap block of size 8 at 0x000109800380 allocated by main thread:
    #0 malloc <null>
    #1 v_malloc allocation.c.v:96
    #2 memdup allocation.c.v:503
    #3 main main.v:13

  Thread T2 (tid=19154470, running) created by main thread at:
    #0 pthread_create <null>
    #1 __v_thread_spawn src.c:672
    #2 main main.v:15
  ...
SUMMARY: ThreadSanitizer: data race main.v:8 in inc
==================
2000
ThreadSanitizer: reported 1 warnings

Making n an atomic int, putting it in a shared object, or guarding it with a sync.Mutex removes the race.

The race detector finds the races that happen while the program runs; it cannot find races in code that does not run. So it is most useful with tests and realistic workloads. A program that reported races exits with status 66, so a test that races fails.

How it works: -race compiles the program with ThreadSanitizer (-fsanitize=thread), the race detection runtime that Go's race detector uses too. Like Go's runtime, V's channels, sync types and closure allocator tell it the happens-before relations that the language guarantees, instead of those of their implementation. V passes the race detector test suite of Go (vlib/v/slow_tests/race). -race needs clang or gcc with the ThreadSanitizer runtime (on some Linux distributions, the libtsan package for gcc), and uses clang when it is installed: gcc does not instrument copies of whole struct values, like strings and arrays passed to functions, so it misses races on them. It is supported on linux (amd64, arm64, ppc64le, s390x, loongarch64, riscv64), macos (amd64, arm64), freebsd/amd64 and netbsd/amd64. Race builds:

  • do not use a garbage collector, like -gc none, and cannot use -prealloc: ThreadSanitizer has to see every allocation and free of heap memory, which a garbage collector or an arena allocator hides from it;
  • typically run 2-20x slower and use 5-10x more memory, like in Go (with -prod, a function that was inlined into its caller is reported as the caller);
  • define race, so code can check for them with $if race ? {}, and _d_race.v files are compiled in them. That define is reserved: -d race without -race is an error.

The VRACE environment variable passes options to the race detector, in the same format as Go's GORACE. For example, VRACE="halt_on_error=1" stops the program at the first race, VRACE="log_path=/tmp/race" writes the reports to /tmp/race.<pid> instead of stderr, and VRACE="exitcode=1" changes the exit status. TSAN_OPTIONS takes the same options, and overrides VRACE.

On some Linux kernels, older ThreadSanitizer runtimes stop with FATAL: ThreadSanitizer: unexpected memory mapping. Run the program with address space randomization reduced (setarch $(uname -m) -R ./program, or sudo sysctl vm.mmap_rnd_bits=28), or use a newer compiler.

On this page