具有数组的意外竞争条件

Question

我试图设计一个简单的咖啡机系统，它使用三种资源:牛奶、咖啡、水。

对于每个订单，我创建一个新线程，每个线程为每个资源创建另外三个线程，以便在顺序访问它们时不要等待空闲资源。

根据资源，我将它们存储在一个简单的数组中，并使用锁保护每个索引(它代表三个索引中的一个)，但不幸的是，当我打开TSAN时出现了一个竞争条件。

我确信没有两个线程会同时访问相同的索引，而且为了更有信心，我将资源从数组中的变量更改为独立的变量，并且没有发生竞争，那么为什么当通过锁保护每个索引时访问相同的数组会导致争用条件。

下面的代码具有资源数组(具有争用条件的代码)：

    import Foundation

    enum Resources: Int {
        case milk
        case coffe
        case water
    }

    struct Order {
        var id: Int
        var requiredMilk: Int
        var requiredCoffe: Int
        var requiredWater: Int
        
        init(requiredMilk: Int, requiredCoffe: Int, requiredWater: Int) {
            id = Int.random(in: 0...Int.max)
            self.requiredMilk = requiredMilk
            self.requiredCoffe = requiredCoffe
            self.requiredWater = requiredWater
        }
    }

    var resources: [Int] = []
    let dispatchGroup = DispatchGroup()
    let queue = DispatchQueue(label: "Ordering Queue", attributes: .concurrent)
    let locks = [NSLock(), NSLock(), NSLock()]

    func initiateResources(initialMilk: Int, initialCoffe: Int, initialWater: Int) {
        resources.append(contentsOf: [initialMilk, initialCoffe, initialWater])
    }

    func getAvailableMilk() -> Int {
        return resources[Resources.milk.rawValue]
    }

    func getAvailableCoffe() -> Int {
        return resources[Resources.coffe.rawValue]
    }

    func getAvailableWater() -> Int {
        return resources[Resources.water.rawValue]
    }

    func handleOrder(_ order: Order) {
        dispatchGroup.enter()
        // pushing the current order to the queue in order to be handled concurrenlty
        queue.async {
            let done = prepareOrder(order)
            if done {
                //print("Order with id \(order.id) prepared successfully")
            } else {
                //print("No sufficient resources for order with id \(order.id)")
            }
            dispatchGroup.leave()
        }
        
    }

    func prepareOrder(_ order: Order) -> Bool {
        var can = true
        let internalQueue = DispatchQueue(label: "internal queue", attributes: .concurrent)
        let internalDispatchGroup = DispatchGroup()
        let internalLock = NSLock()
        
        // opening a new thraed for each type of resources in order to not wait for a free resource
        if order.requiredMilk != 0 {
            internalDispatchGroup.enter()
            internalQueue.async {
                locks[Resources.milk.rawValue].lock()
                if getAvailableMilk() < order.requiredMilk {
                    locks[Resources.milk.rawValue].unlock()
                    internalLock.lock()
                    can = false
                    internalLock.unlock()
                } else {
                    //Thread.sleep(forTimeInterval: 3)
                    resources[Resources.milk.rawValue] -= order.requiredMilk
                    locks[Resources.milk.rawValue].unlock()
                }
                internalDispatchGroup.leave()
            }
        }
        
        if order.requiredCoffe != 0 {
            internalDispatchGroup.enter()
            internalQueue.async {
                locks[Resources.coffe.rawValue].lock()
                if getAvailableCoffe() < order.requiredCoffe {
                    locks[Resources.coffe.rawValue].unlock()
                    internalLock.lock()
                    can = false
                    internalLock.unlock()
                } else {
                    //Thread.sleep(forTimeInterval: 2)
                    resources[Resources.coffe.rawValue] -= order.requiredCoffe
                    locks[Resources.coffe.rawValue].unlock()
                }
                internalDispatchGroup.leave()
            }
        }
        
        if order.requiredWater != 0 {
            internalDispatchGroup.enter()
            internalQueue.async {
                locks[Resources.water.rawValue].lock()
                if getAvailableWater() < order.requiredWater {
                    locks[Resources.water.rawValue].unlock()
                    internalLock.lock()
                    can = false
                    internalLock.unlock()
                } else {
                    //Thread.sleep(forTimeInterval: 1)
                    resources[Resources.water.rawValue] -= order.requiredWater
                    locks[Resources.water.rawValue].unlock()
                }
                internalDispatchGroup.leave()
            }
        }
        // making sure that the three resources threads funished successfully
        internalDispatchGroup.wait()
        return can
    }

    func endWork() {
        dispatchGroup.wait()
    }

    // note that I commented the thread.sleep in order to run faster, but every resource should have different serving time

    initiateResources(initialMilk: 10000, initialCoffe: 10000, initialWater: 10000)
    for _ in 0..<10 {
        handleOrder(Order(requiredMilk: 5, requiredCoffe: 5, requiredWater: 5))
    }
    endWork()
    print(resources[0])
    print(resources[1])
    print(resources[2])

    /*
     running 90 threads ech need 5 units from each resource ==> total of 450 units
     So reaming resources should be 9550 from each type if no confclicts occured
     */

以下是相同的变量，但每个资源都有独立的变量(没有竞争条件)：

    import Foundation

    enum Resources: Int {
        case milk
        case coffe
        case water
    }

    struct Order {
        var id: Int
        var requiredMilk: Int
        var requiredCoffe: Int
        var requiredWater: Int
        
        init(requiredMilk: Int, requiredCoffe: Int, requiredWater: Int) {
            id = Int.random(in: 0...Int.max)
            self.requiredMilk = requiredMilk
            self.requiredCoffe = requiredCoffe
            self.requiredWater = requiredWater
        }
    }

    var milkRsource = 0, coffeResource = 0, waterResource = 0
    let dispatchGroup = DispatchGroup()
    let queue = DispatchQueue(label: "Ordering Queue", attributes: .concurrent)
    let milkLock = NSLock()
    let coffeLock = NSLock()
    let waterLock = NSLock()

    func initiateResources(initialMilk: Int, initialCoffe: Int, initialWater: Int) {
        milkRsource =  initialMilk
        coffeResource = initialCoffe
        waterResource = initialWater
    }

    func getAvailableMilk() -> Int {
        return milkRsource
    }

    func getAvailableCoffe() -> Int {
        return coffeResource
    }

    func getAvailableWater() -> Int {
        return waterResource
    }

    func handleOrder(_ order: Order) {
        dispatchGroup.enter()
        // pushing the current order to the queue in order to be handled concurrenlty
        queue.async {
            let done = prepareOrder(order)
            if done {
                //print("Order with id \(order.id) prepared successfully")
            } else {
                //print("No sufficient resources for order with id \(order.id)")
            }
            dispatchGroup.leave()
        }
        
    }

    func prepareOrder(_ order: Order) -> Bool {
        var can = true
        let internalQueue = DispatchQueue(label: "internal queue", attributes: .concurrent)
        let internalDispatchGroup = DispatchGroup()
        let internalLock = NSLock()
        
        // opening a new thraed for each type of resources in order to not wait for a free resource
        if order.requiredMilk != 0 {
            internalDispatchGroup.enter()
            internalQueue.async {
                milkLock.lock()
                if getAvailableMilk() < order.requiredMilk {
                    milkLock.unlock()
                    internalLock.lock()
                    can = false
                    internalLock.unlock()
                } else {
                    //Thread.sleep(forTimeInterval: 3)
                    milkRsource -= order.requiredMilk
                    milkLock.unlock()
                }
                internalDispatchGroup.leave()
            }
        }
        
        if order.requiredCoffe != 0 {
            internalDispatchGroup.enter()
            internalQueue.async {
                coffeLock.lock()
                if getAvailableCoffe() < order.requiredCoffe {
                    coffeLock.unlock()
                    internalLock.lock()
                    can = false
                    internalLock.unlock()
                } else {
                    //Thread.sleep(forTimeInterval: 2)
                    coffeResource -= order.requiredCoffe
                    coffeLock.unlock()
                }
                internalDispatchGroup.leave()
            }
        }
        
        if order.requiredWater != 0 {
            internalDispatchGroup.enter()
            internalQueue.async {
                waterLock.lock()
                if getAvailableWater() < order.requiredWater {
                    waterLock.unlock()
                    internalLock.lock()
                    can = false
                    internalLock.unlock()
                } else {
                    //Thread.sleep(forTimeInterval: 1)
                    waterResource -= order.requiredWater
                    waterLock.unlock()
                }
                internalDispatchGroup.leave()
            }
        }
        // making sure that the three resources threads funished successfully
        internalDispatchGroup.wait()
        return can
    }

    func endWork() {
        dispatchGroup.wait()
    }

    // note that I commented the thread.sleep in order to run faster, but every resource should have different serving time

    initiateResources(initialMilk: 10000, initialCoffe: 10000, initialWater: 10000)
    for _ in 0..<10 {
        handleOrder(Order(requiredMilk: 5, requiredCoffe: 5, requiredWater: 5))
    }
    endWork()
    print(milkRsource)
    print(coffeResource)
    print(waterResource)

    /*
     running 90 threads ech need 5 units from each resource ==> total of 450 units
     So reaming resources should be 9550 from each type if no confclicts occured
     */

Answer 1

在prepareOrder中，您正在变异resources，只与本地锁同步(即在当前调用中同步)。但是，如果有人多次调用prepareOrder (您是这样的)，就没有什么可以阻止与resources的并发交互。您的同步机制应该保持在它正在同步的变量的级别上。例如，局部变量可以与本地锁或队列(或其他同步机制)同步。属性应该与也是类的属性(但不是本地属性)的锁或队列(或其他什么)同步。