同步机制

在并发编程中，如果缺少同步机制来保护多个线程共享的变量，很容易会出现数据竞争问题（data race）。

仓颉编程语言提供三种常见的同步机制来确保数据的线程安全：原子操作、互斥锁和条件变量。

原子操作 Atomic

仓颉提供整数类型、Bool 类型和引用类型的原子操作。

其中整数类型包括： Int8、Int16、Int32、Int64、UInt8、UInt16、UInt32、UInt64。

整数类型的原子操作支持基本的读写、交换以及算术运算操作：

操作功能
load读取
store写入
swap交换，返回交换前的值
compareAndSwap比较再交换，交换成功返回 true，否则返回 false
fetchAdd加法，返回执行加操作之前的值
fetchSub减法，返回执行减操作之前的值
fetchAnd与，返回执行与操作之前的值
fetchOr或，返回执行或操作之前的值
fetchXor异或，返回执行异或操作之前的值

需要注意的是：

  1. 交换操作和算术操作的返回值是修改前的值。
  2. compareAndSwap 是判断当前原子变量的值是否等于 old 值，如果等于，则使用 new 值替换；否则不替换。

以 Int8 类型为例，对应的原子操作类型声明如下：

class AtomicInt8 {
    public func load(): Int8
    public func store(val: Int8): Unit
    public func swap(val: Int8): Int8
    public func compareAndSwap(old: Int8, new: Int8): Bool
    public func fetchAdd(val: Int8): Int8
    public func fetchSub(val: Int8): Int8
    public func fetchAnd(val: Int8): Int8
    public func fetchOr(val: Int8): Int8
    public func fetchXor(val: Int8): Int8
}

上述每一种原子类型的方法都有一个对应的方法可以接收内存排序参数，目前内存排序参数仅支持顺序一致性。

类似的，其他整数类型对应的原子操作类型有：

class AtomicInt16 {...}
class AtomicInt32 {...}
class AtomicInt64 {...}
class AtomicUInt8 {...}
class AtomicUInt16 {...}
class AtomicUInt32 {...}
class AtomicUInt64 {...}

下方示例演示了如何在多线程程序中，使用原子操作实现计数：

import std.sync.*
import std.time.*
import std.collection.*

let count = AtomicInt64(0)

main(): Int64 {
    let list = ArrayList<Future<Int64>>()

    // create 1000 threads.
    for (i in 0..1000) {
        let fut = spawn {
            sleep(Duration.millisecond) // sleep for 1ms.
            count.fetchAdd(1)
        }
        list.append(fut)
    }

    // Wait for all threads finished.
    for (f in list) {
        f.get()
    }

    let val = count.load()
    println("count = ${val}")
    return 0
}

输出结果应为：

count = 1000

以下是使用整数类型原子操作的一些其他正确示例：

var obj: AtomicInt32 = AtomicInt32(1)
var x = obj.load() // x: 1, the type is Int32
x = obj.swap(2) // x: 1
x = obj.load() // x: 2
var y = obj.compareAndSwap(2, 3) // y: true, the type is Bool.
y = obj.compareAndSwap(2, 3) // y: false, the value in obj is no longer 2 but 3. Therefore, the CAS operation fails.
x = obj.fetchAdd(1) // x: 3
x = obj.load() // x: 4

Bool 类型和引用类型的原子操作只提供读写和交换操作：

操作功能
load读取
store写入
swap交换，返回交换前的值
compareAndSwap比较再交换，交换成功返回 true，否则返回 false

注意：

引用类型原子操作只对引用类型有效。

原子引用类型是 AtomicReference，以下是使用 Bool 类型、引用类型原子操作的一些正确示例：

import std.sync.*

class A {}

main() {
    var obj = AtomicBool(true)
    var x1 = obj.load() // x1: true, the type is Bool
    println(x1)
    var t1 = A()
    var obj2 = AtomicReference(t1)
    var x2 = obj2.load() // x2 and t1 are the same object
    var y1 = obj2.compareAndSwap(x2, t1) // x2 and t1 are the same object, y1: true
    println(y1)
    var t2 = A()
    var y2 = obj2.compareAndSwap(t2, A()) // x and t1 are not the same object, CAS fails, y2: false
    println(y2)
    y2 = obj2.compareAndSwap(t1, A()) // CAS successes, y2: true
    println(y2)
}

编译执行上述代码，输出结果为：

true
true
false
true

可重入互斥锁 ReentrantMutex

可重入互斥锁的作用是对临界区加以保护，使得任意时刻最多只有一个线程能够执行临界区的代码。当一个线程试图获取一个已被其他线程持有的锁时，该线程会被阻塞，直到锁被释放，该线程才会被唤醒，可重入是指线程获取该锁后可再次获得该锁。

注意：

ReentrantMutex 是内置的互斥锁，开发者需要保证不继承它。

使用可重入互斥锁时，必须牢记两条规则：

  1. 在访问共享数据之前，必须尝试获取锁；
  2. 处理完共享数据后，必须进行解锁，以便其他线程可以获得锁。

ReentrantMutex 提供的主要成员函数如下：

public open class ReentrantMutex {
    // Create a ReentrantMutex.
    public init()

    // Locks the mutex, blocks if the mutex is not available.
    public func lock(): Unit

    // Unlocks the mutex. If there are other threads blocking on this
    // lock, then wake up one of them.
    public func unlock(): Unit

    // Tries to lock the mutex, returns false if the mutex is not
    // available, otherwise returns true.
    public func tryLock(): Bool
}

下方示例演示了如何使用 ReentrantMutex 来保护对全局共享变量 count 的访问，对 count 的操作即属于临界区：

import std.sync.*
import std.time.*
import std.collection.*

var count: Int64 = 0
let mtx = ReentrantMutex()

main(): Int64 {
    let list = ArrayList<Future<Unit>>()

    // creat 1000 threads.
    for (i in 0..1000) {
        let fut = spawn {
            sleep(Duration.millisecond) // sleep for 1ms.
            mtx.lock()
            count++
            mtx.unlock()
        }
        list.append(fut)
    }

    // Wait for all threads finished.
    for (f in list) {
        f.get()
    }

    println("count = ${count}")
    return 0
}

输出结果应为：

count = 1000

下方示例演示了如何使用 tryLock：

import std.sync.*

main(): Int64 {
    let mtx: ReentrantMutex = ReentrantMutex()
    var future: Future<Unit> = spawn {
        mtx.lock()
        while (true) {}
        mtx.unlock()
    }
    try {
        future.get(Duration.millisecond * 10)
    } catch (e: TimeoutException) {
        if (mtx.tryLock()) {
            mtx.unlock()
            return 1
        }
        return 0
    }
    return 2
}

输出结果应为空。

以下是互斥锁的一些错误示例：

错误示例 1：线程操作临界区后没有解锁，导致其他线程无法获得锁而阻塞。

import std.sync.*

var sum: Int64 = 0
let mutex = ReentrantMutex()

main() {
    let foo = spawn { =>
        mutex.lock()
        sum = sum + 1
    }
    let bar = spawn { =>
        mutex.lock()
        sum = sum + 1
    }
    foo.get()
    println("${sum}")
    bar.get() // Because the thread is not unlocked, other threads waiting to obtain the current mutex will be blocked.
}

错误示例 2：在本线程没有持有锁的情况下调用 unlock 将会抛出异常。

import std.sync.*

var sum: Int64 = 0
let mutex = ReentrantMutex()

main() {
    let foo = spawn { =>
        sum = sum + 1
        mutex.unlock() // Error, Unlock without obtaining the lock and throw an exception: IllegalSynchronizationStateException.
    }
    foo.get()
    0
}

错误示例 3：tryLock() 并不保证获取到锁，可能会造成不在锁的保护下操作临界区和在没有持有锁的情况下调用 unlock 抛出异常等行为。

var sum: Int64 = 0
let mutex = ReentrantMutex()

main() {
    for (i in 0..100) {
        spawn { =>
            mutex.tryLock() // Error, `tryLock()` just trying to acquire a lock, there is no guarantee that the lock will be acquired, and this can lead to abnormal behavior.
            sum = sum + 1
            mutex.unlock()
        }
    }
}

另外，ReentrantMutex 在设计上是一个可重入锁，也就是说：在某个线程已经持有一个 ReentrantMutex 锁的情况下，再次尝试获取同一个 ReentrantMutex 锁，永远可以立即获得该 ReentrantMutex 锁。

注意：

虽然 ReentrantMutex 是一个可重入锁，但是调用 unlock() 的次数必须和调用 lock() 的次数相同，才能成功释放该锁。

下方示例代码演示了 ReentrantMutex 可重入的特性：

import std.sync.*
import std.time.*

var count: Int64 = 0
let mtx = ReentrantMutex()

func foo() {
    mtx.lock()
    count += 10
    bar()
    mtx.unlock()
}

func bar() {
    mtx.lock()
    count += 100
    mtx.unlock()
}

main(): Int64 {
    let fut = spawn {
        sleep(Duration.millisecond) // sleep for 1ms.
        foo()
    }

    foo()

    fut.get()

    println("count = ${count}")
    return 0
}

输出结果应为：

count = 220

在上方示例中，无论是主线程还是新创建的线程，如果在 foo() 中已经获得了锁，那么继续调用 bar() 的话，在 bar() 函数中由于是对同一个 ReentrantMutex 进行加锁，因此也是能立即获得该锁的，不会出现死锁。

Monitor

Monitor 是一个内置的数据结构，它绑定了互斥锁和单个与之相关的条件变量（也就是等待队列）。Monitor 可以使线程阻塞并等待来自另一个线程的信号以恢复执行。这是一种利用共享变量进行线程同步的机制，主要提供如下方法：

public class Monitor <: ReentrantMutex {
    // Create a monitor.
    public init()

    // Wait for a signal, blocking the current thread.
    public func wait(timeout!: Duration = Duration.Max): Bool

    // Wake up one thread of those waiting on the monitor, if any.
    public func notify(): Unit

    // Wake up all threads waiting on the monitor, if any.
    public func notifyAll(): Unit
}

调用 Monitor 对象的 wait、notify 或 notifyAll 方法前，需要确保当前线程已经持有对应的 Monitor 锁。wait 方法包含如下动作：

  1. 添加当前线程到该 Monitor 对应的等待队列中;
  2. 阻塞当前线程，同时完全释放该 Monitor 锁，并记录锁的重入次数;
  3. 等待某个其它线程使用同一个 Monitor 实例的 notify 或 notifyAll 方法向该线程发出信号;
  4. 当前线程被唤醒后，会自动尝试重新获取 Monitor 锁，且持有锁的重入状态与第 2 步记录的重入次数相同；但是如果尝试获取 Monitor 锁失败，则当前线程会阻塞在该 Monitor 锁上。

wait 方法接受一个可选参数 timeout。需要注意的是，业界很多常用的常规操作系统不保证调度的实时性，因此无法保证一个线程会被阻塞“精确的 N 纳秒”——可能会观察到与系统相关的不精确情况。此外，当前语言规范明确允许实现产生虚假唤醒——在这种情况下，wait 返回值是由实现决定的——可能为 true 或 false。因此鼓励开发者始终将 wait 包在一个循环中：

synchronized (obj) {
  while (<condition is not true>) {
    obj.wait()
  }
}

以下是使用 Monitor 的一个正确示例：

import std.sync.*
import std.time.*

var mon = Monitor()
var flag: Bool = true

main(): Int64 {
    let fut = spawn {
        mon.lock()
        while (flag) {
            println("New thread: before wait")
            mon.wait()
            println("New thread: after wait")
        }
        mon.unlock()
    }

    // Sleep for 10ms, to make sure the new thread can be executed.
    sleep(10 * Duration.millisecond)

    mon.lock()
    println("Main thread: set flag")
    flag = false
    mon.unlock()

    mon.lock()
    println("Main thread: notify")
    mon.notifyAll()
    mon.unlock()

    // wait for the new thread finished.
    fut.get()
    return 0
}

输出结果应为：

New thread: before wait
Main thread: set flag
Main thread: notify
New thread: after wait

Monitor 对象执行 wait 时，必须在锁的保护下进行，否则 wait 中释放锁的操作会抛出异常。

以下是使用条件变量的一些错误示例：

import std.sync.*

var m1 = Monitor()
var m2 = ReentrantMutex()
var flag: Bool = true
var count: Int64 = 0

func foo1() {
    spawn {
        m2.lock()
        while (flag) {
            m1.wait() // Error：The lock used together with the condition variable must be the same lock and in the locked state. Otherwise, the unlock operation in `wait` throws an exception.
        }
        count = count + 1
        m2.unlock()
    }
    m1.lock()
    flag = false
    m1.notifyAll()
    m1.unlock()
}

func foo2() {
    spawn {
        while (flag) {
            m1.wait() // Error：The `wait` of a conditional variable must be called with a lock held.
        }
        count = count + 1
    }
    m1.lock()
    flag = false
    m1.notifyAll()
    m1.unlock()
}

main() {
    foo1()
    foo2()
    m1.wait()
    return 0
}

MultiConditionMonitor

MultiConditionMonitor 是一个内置的数据结构，它绑定了互斥锁和一组与之相关的动态创建的条件变量。该类应仅当在 Monitor 类不足以满足复杂的线程间同步的场景下使用。主要提供如下方法：

public class MultiConditionMonitor <: ReentrantMutex {
   // Constructor.
   init()

   // Returns a new ConditionID associated with this monitor. May be used to implement
   // "single mutex -- multiple wait queues" concurrent primitives.
   // Throws IllegalSynchronizationStateException("Mutex is not locked by the current thread") if the current thread does not hold this mutex.
   func newCondition(): ConditionID

   // Blocks until either a paired `notify` is invoked or `timeout` nanoseconds pass.
   // Returns `true` if the specified condition was signalled by another thread or `false` on timeout.
   // Spurious wakeups are allowed.
   // Throws IllegalSynchronizationStateException("Mutex is not locked by the current thread") if the current thread does not hold this mutex.
   // Throws IllegalSynchronizationStateException("Invalid condition") if `id` was not returned by `newCondition` of this MultiConditionMonitor instance.
   func wait(id: ConditionID, timeout!: Duration = Duration.Max): Bool

   // Wakes up a single thread waiting on the specified condition, if any (no particular admission policy implied).
   // Throws IllegalSynchronizationStateException("Mutex is not locked by the current thread") if the current thread does not hold this mutex.
   // Throws IllegalSynchronizationStateException("Invalid condition") if `id` was not returned by `newCondition` of this MultiConditionMonitor instance.
   func notify(id: ConditionID): Unit

   // Wakes up all threads waiting on the specified condition, if any (no particular admission policy implied).
   // Throws IllegalSynchronizationStateException("Mutex is not locked by the current thread") if the current thread does not hold this mutex.
   // Throws IllegalSynchronizationStateException("Invalid condition") if `id` was not returned by `newCondition` of this MultiConditionMonitor instance.
   func notifyAll(id: ConditionID): Unit
}
  1. newCondition(): ConditionID：创建一个新的条件变量并与当前对象关联，返回一个特定的 ConditionID 标识符
  2. wait(id: ConditionID, timeout!: Duration = Duration.Max): Bool：等待信号，阻塞当前线程
  3. notify(id: ConditionID): Unit：唤醒一个在 Monitor 上等待的线程（如果有）
  4. notifyAll(id: ConditionID): Unit：唤醒所有在 Monitor 上等待的线程（如果有）

初始化时，MultiConditionMonitor 没有与之相关的 ConditionID 实例。每次调用 newCondition 都会将创建一个新的条件变量并与当前对象关联，并返回如下类型作为唯一标识符：

public struct ConditionID {
   private init() { ... } // constructor is intentionally private to prevent
                          // creation of such structs outside of MultiConditionMonitor
}

请注意使用者不可以将一个 MultiConditionMonitor 实例返回的 ConditionID 传给其它实例，或者手动创建 ConditionID（例如使用 unsafe）。由于 ConditionID 所包含的数据（例如内部数组的索引，内部队列的直接地址，或任何其他类型数据等）和创建它的 MultiConditionMonitor 相关，所以将“外部” conditonID 传入 MultiConditionMonitor 中会导致 IllegalSynchronizationStateException。

以下是使用 MultiConditionMonitor 去实现一个长度固定的有界 FIFO 队列，当队列为空，get() 会被阻塞；当队列满了时，put() 会被阻塞。

import std.sync.*

class BoundedQueue {
    // Create a MultiConditionMonitor, two Conditions.
    let m: MultiConditionMonitor = MultiConditionMonitor()
    var notFull: ConditionID
    var notEmpty: ConditionID

    var count: Int64 // Object count in buffer.
    var head: Int64  // Write index.
    var tail: Int64  // Read index.

    // Queue's length is 100.
    let items: Array<Object> = Array<Object>(100, {i => Object()})

    init() {
        count = 0
        head = 0
        tail = 0

        synchronized(m) {
          notFull  = m.newCondition()
          notEmpty = m.newCondition()
        }
    }

    // Insert an object, if the queue is full, block the current thread.
    public func put(x: Object) {
        // Acquire the mutex.
        synchronized(m) {
          while (count == 100) {
            // If the queue is full, wait for the "queue notFull" event.
            m.wait(notFull)
          }
          items[head] = x
          head++
          if (head == 100) {
            head = 0
          }
          count++

          // An object has been inserted and the current queue is no longer
          // empty, so wake up the thread previously blocked on get()
          // because the queue was empty.
          m.notify(notEmpty)
        } // Release the mutex.
    }

    // Pop an object, if the queue is empty, block the current thread.
    public func get(): Object {
        // Acquire the mutex.
        synchronized(m) {
          while (count == 0) {
            // If the queue is empty, wait for the "queue notEmpty" event.
            m.wait(notEmpty)
          }
          let x: Object = items[tail]
          tail++
          if (tail == 100) {
            tail = 0
          }
          count--

          // An object has been popped and the current queue is no longer
          // full, so wake up the thread previously blocked on put()
          // because the queue was full.
          m.notify(notFull)

          return x
        } // Release the mutex.
    }
}

synchronized 关键字

互斥锁 ReentrantMutex 提供了一种便利灵活的加锁的方式，同时因为它的灵活性，也可能引起忘了解锁，或者在持有互斥锁的情况下抛出异常不能自动释放持有的锁的问题。因此，仓颉编程语言提供一个 synchronized 关键字，搭配 ReentrantMutex 一起使用，可以在其后跟随的作用域内自动进行加锁解锁操作，用来解决类似的问题。

下方示例代码演示了如何使用 synchronized 关键字来保护共享数据：

import std.sync.*
import std.time.*
import std.collection.*

var count: Int64 = 0
let mtx = ReentrantMutex()

main(): Int64 {
    let list = ArrayList<Future<Unit>>()

    // creat 1000 threads.
    for (i in 0..1000) {
        let fut = spawn {
            sleep(Duration.millisecond) // sleep for 1ms.
            // Use synchronized(mtx), instead of mtx.lock() and mtx.unlock().
            synchronized(mtx) {
                count++
            }
        }
        list.append(fut)
    }

    // Wait for all threads finished.
    for (f in list) {
        f.get()
    }

    println("count = ${count}")
    return 0
}

输出结果应为：

count = 1000

通过在 synchronized 后面加上一个 ReentrantMutex 实例，对其后面修饰的代码块进行保护，可以使得任意时刻最多只有一个线程可以执行被保护的代码：

  1. 一个线程在进入 synchronized 修饰的代码块之前，会自动获取 ReentrantMutex 实例对应的锁，如果无法获取锁，则当前线程被阻塞；
  2. 一个线程在退出 synchronized 修饰的代码块之前，会自动释放该 ReentrantMutex 实例的锁。

对于控制转移表达式（如 break、continue、return、throw），在导致程序的执行跳出 synchronized 代码块时，也符合上面第 2 条的说明，也就说也会自动释放 synchronized 表达式对应的锁。

下方示例演示了在 synchronized 代码块中出现 break 语句的情况：

import std.sync.*
import std.collection.*

var count: Int64 = 0
var mtx: ReentrantMutex = ReentrantMutex()

main(): Int64 {
    let list = ArrayList<Future<Unit>>()
    for (i in 0..10) {
        let fut = spawn {
            while (true) {
                synchronized(mtx) {
                    count = count + 1
                    break
                    println("in thread")
                }
            }
        }
        list.append(fut)
    }

    // Wait for all threads finished.
    for (f in list) {
        f.get()
    }

    synchronized(mtx) {
        println("in main, count = ${count}")
    }
    return 0
}

输出结果应为：

in main, count = 10

实际上 in thread 这行不会被打印，因为 break 语句实际上会让程序执行跳出 while 循环（当然，在跳出 while 循环之前，是先跳出 synchronized 代码块）。

线程局部变量 ThreadLocal

使用 core 包中的 ThreadLocal 可以创建并使用线程局部变量，每一个线程都有它独立的一个存储空间来保存这些线程局部变量，因此，在每个线程可以安全地访问他们各自的线程局部变量，而不受其他线程的影响。

public class ThreadLocal<T> {
    /*
     * 构造一个携带空值的仓颉线程局部变量
     */
    public init()

    /*
     * 获得仓颉线程局部变量的值，如果值不存在，则返回 Option<T>.None
     * 返回值 Option<T> - 仓颉线程局部变量的值
     */
    public func get(): Option<T>

    /*
     * 通过 value 设置仓颉线程局部变量的值
     * 如果传入 Option<T>.None，该局部变量的值将被删除，在线程后续操作中将无法获取
     * 参数 value - 需要设置的局部变量的值
     */
    public func set(value: Option<T>): Unit
}

下方示例代码演示了如何通过 ThreadLocal类来创建并使用各自线程的局部变量：


main(): Int64 {
    let tl = ThreadLocal<Int64>()
    let fut1 = spawn {
        tl.set(123)
        println("tl in spawn1 = ${tl.get().getOrThrow()}")
    }
    let fut2 = spawn {
        tl.set(456)
        println("tl in spawn2 = ${tl.get().getOrThrow()}")
    }
    fut1.get()
    fut2.get()
    0
}

可能的输出结果如下：

tl in spawn1 = 123
tl in spawn2 = 456

或者

tl in spawn2 = 456
tl in spawn1 = 123