Users › 10 Concurrency

Concurrency

Run code on several threads that share one heap, and synchronize them with synchronized, volatile, Lock and Cond.

1. Threads

The standard library provides a Runnable interface and a Thread class:

%class com.example.todo.Worker implements Runnable
    #import Runnable

    void run() Runnable@run {
        // runs on its own OS thread
    }
Thread t = new Thread(new Worker())
t.start()
t.join()            // wait for it to finish
int id = t.id()     // thread number, for diagnostics

All threads share the same VM: the same heap, the same singletons, the same objects. A worker can read and write what the main thread created, and vice versa.

ThreadUtil.sleep(milliseconds) pauses the current thread (#import ThreadUtil). An uncaught exception ends only that thread — the runtime prints it and the process keeps running.

2. Data races are undefined behavior

Sharing is the default; synchronizing is your job. Concurrent unsynchronized access to the same mutable data is undefined behavior. The rest of this chapter shows the tools.

3. synchronized

synchronized (obj) {
    // mutual exclusion on obj
}

synchronized void bump() {
    // equivalent to wrapping the body in synchronized (this)
}

Reads count too.

If one thread updates a field under a lock, every other thread that reads it must take the same lock. In the task-list example, TaskList.markAllDone() and every reader of the list are synchronized for exactly this reason.

4. volatile fields

volatile int ready

A volatile write is a release and a volatile read is an acquire, so a value written by one thread is visible to another. Volatile does not make compound operations atomic: ready = ready + 1 from several threads still loses updates. Use it for flags published under a lock, or combine it with synchronized when the update itself must be atomic.

Volatile is a field modifier only, and cannot be combined with interface binding.

5. Lock and Cond

Lock is an explicit reentrant mutex; Cond is a condition variable that works with it:

#import Lock
#import Cond

Lock lock
Cond hasItem

int take() {
    this.lock.lock()
    while (!this.full) {
        this.hasItem.wait(this.lock)   // releases lock, waits, re-acquires
    }
    int v = this.item
    this.full = false
    this.hasItem.signal()
    this.lock.unlock()
    return v
}

6. Happens-before

These operations establish ordering between threads: entering and exiting a monitor, a volatile write followed by a volatile read of the same field, Lock.lock/unlock, a Cond signal followed by a successful wait, Thread.start (before) and Thread.join (after), and host-thread attach/leave. Anything else shared without one of these is a data race.

Not in this version: timed waits, thread interruption, thread priorities, a thread pool, and wait/notify on plain objects (there is no Object root).

7. Try it: a synchronized counter

%simple com.example.conc.Demo implements Main
    #import Std
    #import Runnable
    #import Thread
    int n

    void ___init() {
        this.n = 0
    }

    synchronized void bump() {
        this.n = this.n + 1
    }

    int run(int argc, String[] argv) Main@run {
        Thread t1 = new Thread(new $Adder())
        Thread t2 = new Thread(new $Adder())
        t1.start()
        t2.start()
        t1.join()
        t2.join()
        Std.print("n=", this.n, "\n")
        return this.n == 2000 ? 0 : 1
    }

%sub Adder implements Runnable
    #import Runnable

    void run() Runnable@run {
        int i = 0
        while (i < 1000) {
            Demo.bump()
            i = i + 1
        }
    }
n=2000

8. Try it: producer and consumer

%simple com.example.conc.QueueDemo implements Main
    #import Std
    #import Lock
    #import Cond
    #import Runnable
    #import Thread
    Lock lock
    Cond hasItem
    int item
    bool full
    pub int sum

    void ___init() {
        this.lock = new Lock()
        this.hasItem = new Cond()
        this.item = 0
        this.full = false
        this.sum = 0
    }

    void put(int v) {
        this.lock.lock()
        while (this.full) {
            this.hasItem.wait(this.lock)
        }
        this.item = v
        this.full = true
        this.hasItem.signal()
        this.lock.unlock()
    }

    int take() {
        this.lock.lock()
        while (!this.full) {
            this.hasItem.wait(this.lock)
        }
        int v = this.item
        this.full = false
        this.hasItem.signal()
        this.lock.unlock()
        return v
    }

    int run(int argc, String[] argv) Main@run {
        Thread p = new Thread(new $Producer())
        Thread c = new Thread(new $Consumer())
        p.start()
        c.start()
        p.join()
        c.join()
        Std.print("sum=", this.sum, "\n")
        return this.sum == 6 ? 0 : 1
    }

%sub Producer implements Runnable
    #import Runnable

    void run() Runnable@run {
        QueueDemo.put(1)
        QueueDemo.put(2)
        QueueDemo.put(3)
    }

%sub Consumer implements Runnable
    #import Runnable

    void run() Runnable@run {
        int s = 0
        s = s + QueueDemo.take()
        s = s + QueueDemo.take()
        s = s + QueueDemo.take()
        QueueDemo.sum = s
    }
sum=6

9. Where next