仓颉原子操作封装：从底层原理到鸿蒙高并发实战

// 声明一个原子整数，底层自动适配硬件指令
var atomicCount: Atomic<Int> = Atomic(initialValue: 0)

// 合法：Int类型符合AtomicCompatible约束
var atomicInt: Atomic<Int> = Atomic(0)

// 编译错误：String类型不支持原子操作
var atomicStr: Atomic<String> = Atomic("") // 编译器提示："String does not conform to AtomicCompatible"

// 线程1执行：原子自增
atomicCount.increment()

// 线程2执行：原子读取
let current = atomicCount.load() // 确保读取到线程1的最新修改

let atomic = Atomic<Int>(initialValue: 10)

// 原子读取：确保获取最新值
let value = atomic.load() // value = 10

// 原子赋值：确保修改对其他线程可见
atomic.store(20)

// 再次读取：获取更新后的值
assert(atomic.load() == 20)

var requestCounter: Atomic<UInt> = Atomic(0)

// 每处理一个请求，原子自增
func handleRequest() {
    requestCounter.increment() // 等价于 requestCounter.add(1)
    // ... 处理逻辑
}

// 定期统计请求量
func reportRequestCount() {
    let count = requestCounter.load()
    print("当前请求总数：\(count)")
    // 重置计数器（原子赋值）
    requestCounter.store(0)
}

var atomicValue: Atomic<Int> = Atomic(5)

// 尝试将值从5改为10
let success = atomicValue.compareAndSwap(expected: 5, new: 10)
assert(success == true)
assert(atomicValue.load() == 10)

// 再次尝试（期望值不匹配，失败）
let failure = atomicValue.compareAndSwap(expected: 5, new: 15)
assert(failure == false)
assert(atomicValue.load() == 10)

// 无锁栈节点
class Node<T> {
    let value: T
    var next: Node<T>?
    
    init(value: T, next: Node<T>? = nil) {
        self.value = value
        self.next = next
    }
}

// 基于原子操作的无锁栈
class LockFreeStack<T> {
    // 用原子指针存储栈顶节点
    private var top: Atomic<Node<T>?> = Atomic(nil)
    
    // 入栈操作
    func push(value: T) {
        let newNode = Node(value: value)
        while true {
            // 读取当前栈顶
            let currentTop = top.load()
            // 新节点的next指向当前栈顶
            newNode.next = currentTop
            // CAS尝试更新栈顶为新节点
            if top.compareAndSwap(expected: currentTop, new: newNode) {
                // 成功入栈，退出循环
                return
            }
            // 失败则重试（其他线程已修改栈顶）
        }
    }
    
    // 出栈操作（简化版）
    func pop() -> T? {
        while true {
            let currentTop = top.load()
            guard let topNode = currentTop else {
                return nil // 栈为空
            }
            // 尝试将栈顶更新为下一个节点
            if top.compareAndSwap(expected: topNode, new: topNode.next) {
                return topNode.value
            }
            // 失败则重试
        }
    }
}

var atomic: Atomic<Int> = Atomic(10)

// 获取当前值（10），然后加5（最终值15）
let oldValue = atomic.fetchAdd(5)
assert(oldValue == 10)
assert(atomic.load() == 15)

// 获取当前值（15），然后与3按位与（15 & 3 = 3）
let old = atomic.fetchAnd(3)
assert(old == 15)
assert(atomic.load() == 3)

// 原子计数器：记录心率采样次数
let heartRateCounter: Atomic<UInt> = Atomic(0)

// 传感器采样线程
async func heartRateSamplingThread() {
    while isRunning {
        // 采样逻辑（模拟）
        let sample = await HeartRateSensor.sample()
        processSample(sample)
        
        // 原子自增计数器
        heartRateCounter.increment()
        
        // 1秒采样50次，每次间隔20ms
        await sleep(20)
    }
}

// UI刷新线程
async func uiRefreshThread() {
    while isRunning {
        // 原子读取当前计数
        let currentCount = heartRateCounter.load()
        // 更新UI
        HeartRateUI.updateCount(currentCount)
        // 100ms刷新一次
        await sleep(100)
    }
}

// 任务节点
class TaskNode {
    let task: () -> Void
    var next: TaskNode?
    // 原子标记：0=未处理，1=处理中，2=已完成
    var status: Atomic<UInt8> = Atomic(0)
    
    init(task: @escaping () -> Void) {
        self.task = task
    }
}

// 分布式无锁任务队列
class DistributedTaskQueue {
    // 队列头（哨兵节点，简化边界处理）
    private let head: TaskNode = TaskNode(task: {})
    // 原子指针：指向队尾节点
    private var tail: Atomic<TaskNode> = Atomic(TaskNode(task: {}))
    
    init() {
        // 初始化时头尾指向同一哨兵节点
        tail.store(head)
    }
    
    // 入队：多生产者安全
    func enqueue(task: @escaping () -> Void) {
        let newNode = TaskNode(task: task)
        while true {
            let currentTail = tail.load()
            // 尝试将当前队尾的next指向新节点
            if currentTail.next.compareAndSwap(expected: nil, new: newNode) {
                // 成功后将队尾更新为新节点
                tail.compareAndSwap(expected: currentTail, new: newNode)
                return
            } else {
                // 其他线程已修改队尾，协助推进队尾指针
                tail.compareAndSwap(expected: currentTail, new: currentTail.next.load()!)
            }
        }
    }
    
    // 出队：多消费者安全
    func dequeue() -> (() -> Void)? {
        while true {
            let currentHead = head.next.load()
            guard let node = currentHead else {
                return nil // 队列为空
            }
            // 尝试将节点状态从0（未处理）改为1（处理中）
            if node.status.compareAndSwap(expected: 0, new: 1) {
                // 更新队头指针
                head.next.compareAndSwap(expected: node, new: node.next.load())
                return node.task
            }
            // 状态已被修改（其他消费者处理），继续尝试
        }
    }
}

// 分布式调度示例
func distributedSchedule() {
    let queue = DistributedTaskQueue()
    
    // 多设备并发提交任务（生产者）
    for deviceId in ["device1", "device2", "device3"] {
        go {
            queue.enqueue {
                processTaskOnDevice(deviceId)
            }
        }
    }
    
    // 多设备并发执行任务（消费者）
    for _ in 0..<3 {
        go {
            if let task = queue.dequeue() {
                task()
            }
        }
    }
}

// 线程1读取值为A
let a = atomic.load()
// 线程2将值改为B，再改回A
// 线程1执行CAS(A→C)，成功但实际中间已变化

func adaptiveCas(atomic: Atomic<Int>, expected: Int, new: Int) -> Bool {
    var attempts = 0
    while attempts < 10 {
        if atomic.compareAndSwap(expected: expected, new: new) {
            return true
        }
        attempts += 1
    }
    // 重试多次失败，主动退让
    coroutine_yield()
    return atomic.compareAndSwap(expected: expected, new: new)
}