Создать потокобезопасный массив в Swift

У меня проблема с многопоточностью в Swift. У меня есть массив с некоторыми объектами в нем. Через делегата класс получает новые объекты примерно каждую секунду. После этого я должен проверить, есть ли объекты уже в массиве, поэтому я должен обновить объект, в противном случае я должен удалить/добавить новый объект.

Если я добавляю новый объект, мне сначала нужно получить некоторые данные по сети. Это гендельт через блок.

Теперь моя проблема в том, как мне синхронизировать эту задачу?

Я пробовал dispatch_semaphore, но этот блокирует пользовательский интерфейс, пока блок не будет завершен.

Я также попробовал простую переменную bool, которая проверяет, выполняется ли блок в настоящий момент, и тем временем пропускает метод сравнения.

Но оба метода не идеальны.

Какой лучший способ управлять массивом, я не хочу дублировать данные в массиве.

Ответы

Ответ 1

Кирстейнс верен, но вам не всегда нужно использовать очередь отправки. Ты можешь использовать:

objc_sync_enter(array)
// manipulate the array
objc_sync_exit(array)

Это должно сработать. Для дополнительного бонуса вы можете создать функцию, которая будет использоваться всякий раз, когда вам нужна безопасность потоков:

func sync(lock: NSObject, closure: () -> Void) {
    objc_sync_enter(lock)
    closure()
    objc_sync_exit(lock)
}

...
var list = NSMutableArray()
sync (list) {
   list.addObject("something")
}

Обратите внимание на то, что я изменил AnyObject к NSObject. В Swift типы коллекций реализованы как struct и они передаются по значению, поэтому я предполагаю, что было бы безопаснее работать с изменяемыми классами коллекций, которые передаются по ссылке при использовании удобной функции sync.

Обновление для Swift

Рекомендуемый шаблон для многопоточного доступа использует barrier диспетчеризации:

let queue = DispatchQueue(label: "thread-safe-obj", attributes: .concurrent)

// write
queue.async(flags: .barrier) {
    // perform writes on data
}

// read
var value: ValueType!
queue.sync {
    // perform read and assign value
}
return value

Ответ 2

Мой подход к этой проблеме заключался в использовании очереди последовательной отправки, чтобы синхронизировать доступ к массиву в штучной упаковке. Он будет блокировать поток, когда вы попытаетесь получить значение в индексе, и очередь действительно занята, но проблема также связана с блокировками.

public class SynchronizedArray<T> {
    private var array: [T] = []
    private let accessQueue = dispatch_queue_create("SynchronizedArrayAccess", DISPATCH_QUEUE_SERIAL)

    public func append(newElement: T) {
        dispatch_async(self.accessQueue) {
            self.array.append(newElement)
        }
    }

    public subscript(index: Int) -> T {
        set {
            dispatch_async(self.accessQueue) {
                self.array[index] = newValue
            }
        }
        get {
            var element: T!

            dispatch_sync(self.accessQueue) {
                element = self.array[index]
            }

            return element
        }
    }
}

var a = SynchronizedArray<Int>()
a.append(1)
a.append(2)
a.append(3)

// can be empty as this is non-thread safe access
println(a.array)

// thread-safe synchonized access
println(a[0])
println(a[1])
println(a[2])

Ответ 3

Ответ Kirsteins верен, но для удобства я обновил этот ответ с предложениями Amol Chaudhari и Rob для использования параллельной очереди с асинхронным барьером, чтобы разрешить параллельные чтения, но блокировать записи.

Я также добавил некоторые другие функции массива, которые были полезны для меня.

public class SynchronizedArray<T> {
private var array: [T] = []
private let accessQueue = dispatch_queue_create("SynchronizedArrayAccess", DISPATCH_QUEUE_CONCURRENT)

public func append(newElement: T) {
    dispatch_barrier_async(self.accessQueue) {
        self.array.append(newElement)
    }
}

public func removeAtIndex(index: Int) {
    dispatch_barrier_async(self.accessQueue) {
        self.array.removeAtIndex(index)
    }
}

public var count: Int {
    var count = 0

    dispatch_sync(self.accessQueue) {
        count = self.array.count
    }

    return count
}

public func first() -> T? {
    var element: T?

    dispatch_sync(self.accessQueue) {
        if !self.array.isEmpty {
            element = self.array[0]
        }
    }

    return element
}

public subscript(index: Int) -> T {
    set {
        dispatch_barrier_async(self.accessQueue) {
            self.array[index] = newValue
        }
    }
    get {
        var element: T!

        dispatch_sync(self.accessQueue) {
            element = self.array[index]
        }

        return element
    }
}
}

UPDATE Это тот же код, обновленный для Swift3.

public class SynchronizedArray<T> {
private var array: [T] = []
private let accessQueue = DispatchQueue(label: "SynchronizedArrayAccess", attributes: .concurrent)

public func append(newElement: T) {

    self.accessQueue.async(flags:.barrier) {
        self.array.append(newElement)
    }
}

public func removeAtIndex(index: Int) {

    self.accessQueue.async(flags:.barrier) {
        self.array.remove(at: index)
    }
}

public var count: Int {
    var count = 0

    self.accessQueue.sync {
        count = self.array.count
    }

    return count
}

public func first() -> T? {
    var element: T?

    self.accessQueue.sync {
        if !self.array.isEmpty {
            element = self.array[0]
        }
    }

    return element
}

public subscript(index: Int) -> T {
    set {
        self.accessQueue.async(flags:.barrier) {
            self.array[index] = newValue
        }
    }
    get {
        var element: T!
        self.accessQueue.sync {
            element = self.array[index]
        }

        return element
    }
}
}

Ответ 4

Небольшая деталь: в Swift 3 (по крайней мере, в Xcode 8 Beta 6) синтаксис для очередей значительно изменился. Важные изменения в ответе @Kirsteins:

private let accessQueue = DispatchQueue(label: "SynchronizedArrayAccess")

txAccessQueue.async() {
  // Your async code goes here...
}

txAccessQueue.sync() {
  // Your sync code goes here...
}

Ответ 5

подробности

Xcode 10.1 (10B61), Swift 4.2
Xcode 10.2.1 (10E1001), Swift 5

Решение

import Foundation

// https://developer.apple.com/documentation/swift/rangereplaceablecollection
struct AtomicArray<T>: RangeReplaceableCollection {

    typealias Element = T
    typealias Index = Int
    typealias SubSequence = AtomicArray<T>
    typealias Indices = Range<Int>
    fileprivate var array: Array<T>
    var startIndex: Int { return array.startIndex }
    var endIndex: Int { return array.endIndex }
    var indices: Range<Int> { return array.indices }

    func index(after i: Int) -> Int { return array.index(after: i) }

    private var semaphore = DispatchSemaphore(value: 1)
    fileprivate func _wait() { semaphore.wait() }
    fileprivate func _signal() { semaphore.signal() }
}

// MARK: - Instance Methods

extension AtomicArray {

    init<S>(_ elements: S) where S : Sequence, AtomicArray.Element == S.Element {
        array = Array<S.Element>(elements)
    }

    init() { self.init([]) }

    init(repeating repeatedValue: AtomicArray.Element, count: Int) {
        let array = Array(repeating: repeatedValue, count: count)
        self.init(array)
    }
}

// MARK: - Instance Methods

extension AtomicArray {

    public mutating func append(_ newElement: AtomicArray.Element) {
        _wait(); defer { _signal() }
        array.append(newElement)
    }

    public mutating func append<S>(contentsOf newElements: S) where S : Sequence, AtomicArray.Element == S.Element {
        _wait(); defer { _signal() }
        array.append(contentsOf: newElements)
    }

    func filter(_ isIncluded: (AtomicArray.Element) throws -> Bool) rethrows -> AtomicArray {
        _wait(); defer { _signal() }
        let subArray = try array.filter(isIncluded)
        return AtomicArray(subArray)
    }

    public mutating func insert(_ newElement: AtomicArray.Element, at i: AtomicArray.Index) {
        _wait(); defer { _signal() }
        array.insert(newElement, at: i)
    }

    mutating func insert<S>(contentsOf newElements: S, at i: AtomicArray.Index) where S : Collection, AtomicArray.Element == S.Element {
        _wait(); defer { _signal() }
        array.insert(contentsOf: newElements, at: i)
    }

    mutating func popLast() -> AtomicArray.Element? {
        _wait(); defer { _signal() }
        return array.popLast()
    }

    @discardableResult mutating func remove(at i: AtomicArray.Index) -> AtomicArray.Element {
        _wait(); defer { _signal() }
        return array.remove(at: i)
    }

    mutating func removeAll() {
        _wait(); defer { _signal() }
        array.removeAll()
    }

    mutating func removeAll(keepingCapacity keepCapacity: Bool) {
        _wait(); defer { _signal() }
        array.removeAll()
    }

    mutating func removeAll(where shouldBeRemoved: (AtomicArray.Element) throws -> Bool) rethrows {
        _wait(); defer { _signal() }
        try array.removeAll(where: shouldBeRemoved)
    }

    @discardableResult mutating func removeFirst() -> AtomicArray.Element {
        _wait(); defer { _signal() }
        return array.removeFirst()
    }

    mutating func removeFirst(_ k: Int) {
        _wait(); defer { _signal() }
        array.removeFirst(k)
    }

    @discardableResult mutating func removeLast() -> AtomicArray.Element {
        _wait(); defer { _signal() }
        return array.removeLast()
    }

    mutating func removeLast(_ k: Int) {
        _wait(); defer { _signal() }
        array.removeLast(k)
    }

    @inlinable public func forEach(_ body: (Element) throws -> Void) rethrows {
        _wait(); defer { _signal() }
        try array.forEach(body)
    }

    mutating func removeFirstIfExist(where shouldBeRemoved: (AtomicArray.Element) throws -> Bool) {
        _wait(); defer { _signal() }
        guard let index = try? array.firstIndex(where: shouldBeRemoved) else { return }
        array.remove(at: index)
    }

    mutating func removeSubrange(_ bounds: Range<Int>) {
        _wait(); defer { _signal() }
        array.removeSubrange(bounds)
    }

    mutating func replaceSubrange<C, R>(_ subrange: R, with newElements: C) where C : Collection, R : RangeExpression, T == C.Element, AtomicArray<Element>.Index == R.Bound {
        _wait(); defer { _signal() }
        array.replaceSubrange(subrange, with: newElements)
    }

    mutating func reserveCapacity(_ n: Int) {
        _wait(); defer { _signal() }
        array.reserveCapacity(n)
    }

    public var count: Int {
        _wait(); defer { _signal() }
        return array.count
    }

    public var isEmpty: Bool {
        _wait(); defer { _signal() }
        return array.isEmpty
    }
}

// MARK: - Get/Set

extension AtomicArray {

    // Single  action

    func get() -> [T] {
        _wait(); defer { _signal() }
        return array
    }

    mutating func set(array: [T]) {
        _wait(); defer { _signal() }
        self.array = array
    }

    // Multy actions

    mutating func get(closure: ([T])->()) {
        _wait(); defer { _signal() }
        closure(array)
    }

    mutating func set(closure: ([T]) -> ([T])) {
        _wait(); defer { _signal() }
        array = closure(array)
    }
}

// MARK: - Subscripts

extension AtomicArray {

    subscript(bounds: Range<AtomicArray.Index>) -> AtomicArray.SubSequence {
        get {
            _wait(); defer { _signal() }
            return AtomicArray(array[bounds])
        }
    }

    subscript(bounds: AtomicArray.Index) -> AtomicArray.Element {
        get {
            _wait(); defer { _signal() }
            return array[bounds]
        }
        set(value) {
            _wait(); defer { _signal() }
            array[bounds] = value
        }
    }
}

// MARK: - Operator Functions

extension AtomicArray {

    static func + <Other>(lhs: Other, rhs: AtomicArray) -> AtomicArray where Other : Sequence, AtomicArray.Element == Other.Element {
        return AtomicArray(lhs + rhs.get())
    }

    static func + <Other>(lhs: AtomicArray, rhs: Other) -> AtomicArray where Other : Sequence, AtomicArray.Element == Other.Element {
        return AtomicArray(lhs.get() + rhs)
    }

    static func + <Other>(lhs: AtomicArray, rhs: Other) -> AtomicArray where Other : RangeReplaceableCollection, AtomicArray.Element == Other.Element {
        return AtomicArray(lhs.get() + rhs)
    }

    static func + (lhs: AtomicArray<Element>, rhs: AtomicArray<Element>) -> AtomicArray {
        return AtomicArray(lhs.get() + rhs.get())
    }

    static func += <Other>(lhs: inout AtomicArray, rhs: Other) where Other : Sequence, AtomicArray.Element == Other.Element {
        lhs._wait(); defer { lhs._signal() }
        lhs.array += rhs
    }
}

// MARK: - CustomStringConvertible

extension AtomicArray: CustomStringConvertible {
    var description: String {
        _wait(); defer { _signal() }
        return "\(array)"
    }
}

// MARK: - Equatable

extension AtomicArray where Element : Equatable {

    func split(separator: Element, maxSplits: Int, omittingEmptySubsequences: Bool) -> [ArraySlice<Element>] {
        _wait(); defer { _signal() }
        return array.split(separator: separator, maxSplits: maxSplits, omittingEmptySubsequences: omittingEmptySubsequences)
    }

    func firstIndex(of element: Element) -> Int? {
        _wait(); defer { _signal() }
        return array.firstIndex(of: element)
    }

    func lastIndex(of element: Element) -> Int? {
        _wait(); defer { _signal() }
        return array.lastIndex(of: element)
    }

    func starts<PossiblePrefix>(with possiblePrefix: PossiblePrefix) -> Bool where PossiblePrefix : Sequence, Element == PossiblePrefix.Element {
        _wait(); defer { _signal() }
        return array.starts(with: possiblePrefix)
    }

    func elementsEqual<OtherSequence>(_ other: OtherSequence) -> Bool where OtherSequence : Sequence, Element == OtherSequence.Element {
        _wait(); defer { _signal() }
        return array.elementsEqual(other)
    }

    func contains(_ element: Element) -> Bool {
        _wait(); defer { _signal() }
        return array.contains(element)
    }

    static func != (lhs: AtomicArray<Element>, rhs: AtomicArray<Element>) -> Bool {
        lhs._wait(); defer { lhs._signal() }
        rhs._wait(); defer { rhs._signal() }
        return lhs.array != rhs.array
    }

    static func == (lhs: AtomicArray<Element>, rhs: AtomicArray<Element>) -> Bool {
        lhs._wait(); defer { lhs._signal() }
        rhs._wait(); defer { rhs._signal() }
        return lhs.array == rhs.array
    }
}

Пример использования 1

import Foundation

// init
var array = AtomicArray<Int>()
print(array)
array = AtomicArray(repeating: 0, count: 5)
print(array)
array = AtomicArray([1,2,3,4,5,6,7,8,9])
print(array)

// add
array.append(0)
print(array)
array.append(contentsOf: [5,5,5])
print(array)

// filter
array = array.filter { $0 < 7 }
print(array)

// map
let strings = array.map { "\($0)" }
print(strings)

// insert
array.insert(99, at: 5)
print(array)
array.insert(contentsOf: [2, 2, 2], at: 0)
print(array)

// pop
_ = array.popLast()
print(array)
_ = array.popFirst()
print(array)

// remove
array.removeFirst()
print(array)
array.removeFirst(3)
print(array)
array.remove(at: 2)
print(array)
array.removeLast()
print(array)
array.removeLast(5)
print(array)
array.removeAll { $0%2 == 0 }
print(array)
array = AtomicArray([1,2,3,4,5,6,7,8,9,0])
array.removeSubrange(0...2)
print(array)
array.replaceSubrange(0...2, with: [0,0,0])
print(array)
array.removeAll()
print(array)

array.set(array: [1,2,3,4,5,6,7,8,9,0])
print(array)

// subscript
print(array[0])
array[0] = 100
print(array)
print(array[1...4])

// operator functions
array = [1,2,3] + AtomicArray([4,5,6])
print(array)
array = AtomicArray([4,5,6]) + [1,2,3]
print(array)
array = AtomicArray([1,2,3]) + AtomicArray([4,5,6])
print(array)

Пример использования 2

import Foundation

var arr = AtomicArray([0,1,2,3,4,5])
for i in 0...1000 {
    // Single actions
    DispatchQueue.global(qos: .background).async {
        usleep(useconds_t(Int.random(in: 100...10000)))
        let num = i*i
        arr.append(num)
        print("arr.append(\(num)), background queue")
    }
    DispatchQueue.global(qos: .default).async {
        usleep(useconds_t(Int.random(in: 100...10000)))
        arr.append(arr.count)
        print("arr.append(\(arr.count)), default queue")
    }

    // multy actions
    DispatchQueue.global(qos: .utility).async {
        arr.set { array -> [Int] in
            var newArray = array
            newArray.sort()
            print("sort(), .utility queue")
            return newArray
        }
    }
}

Ответ 6

Я думаю, что dispatch_barriers стоит посмотреть. Использование gcd для синхронности более интуитивно для меня, чем использование ключевого слова synchronize, чтобы избежать мутации состояния из нескольких потоков.

https://mikeash.com/pyblog/friday-qa-2011-10-14-whats-new-in-gcd.html

Ответ 7

Здесь есть отличный ответ, который является потокобезопасным и не блокирует одновременные чтения: fooobar.com/questions/21948/...

Это написано в Objective C, но перенос на Swift тривиален.

@property (nonatomic, readwrite, strong) dispatch_queue_t thingQueue;
@property (nonatomic, strong) NSObject *thing;

- (id)init {
  ...
    _thingQueue = dispatch_queue_create("...", DISPATCH_QUEUE_CONCURRENT);
  ...
}

- (NSObject *)thing {
  __block NSObject *thing;
  dispatch_sync(self.thingQueue, ^{
    thing = _thing;
  });
  return thing;
}

- (void)setThing:(NSObject *)thing {
  dispatch_barrier_async(self.thingQueue, ^{
    _thing = thing;
  });
}

Кредит https://stackoverflow.com/users/97337/rob-napier

Ответ 8

Подход:

Используйте DispatchQueue для синхронизации

См:

http://basememara.com/creating-thread-safe-arrays-in-swift/

Код:

Ниже приведена грубая реализация поточно-безопасного массива, вы можете настроить его.

public class ThreadSafeArray<Element> {

    private var elements    : [Element]
    private let syncQueue   = DispatchQueue(label: "Sync Queue",
                                            qos: .default,
                                            attributes: .concurrent,
                                            autoreleaseFrequency: .inherit,
                                            target: nil)

    public init() {
        elements = []
    }

    public init(_ newElements: [Element]) {
        elements = newElements
    }

    //MARK: Non-mutating

    public var first : Element? {
        return syncQueue.sync {
            elements.first
        }
    }

    public var last : Element? {
        return syncQueue.sync {
            elements.last
        }
    }

    public var count : Int {

        return syncQueue.sync {
            elements.count
        }
    }

    public subscript(index: Int) -> Element {

        get {
            return syncQueue.sync {
                elements[index]
            }
        }

        set {
            syncQueue.sync(flags: .barrier) {
                elements[index] = newValue
            }
        }
    }

    public func reversed() -> [Element] {

        return syncQueue.sync {

            elements.reversed()
        }
    }

    public func flatMap<T>(_ transform: (Element) throws -> T?) rethrows -> [T]  {

        return try syncQueue.sync {

           try elements.flatMap(transform)
        }
    }

    public func filter(_ isIncluded: (Element) -> Bool) -> [Element] {

        return syncQueue.sync {

            elements.filter(isIncluded)
        }
    }

    //MARK: Mutating

    public func append(_ element: Element) {

        syncQueue.sync(flags: .barrier) {

            elements.append(element)
        }
    }

    public func append<S>(contentsOf newElements: S) where Element == S.Element, S : Sequence {

        syncQueue.sync(flags: .barrier) {

            elements.append(contentsOf: newElements)
        }
    }

    public func remove(at index: Int) -> Element? {

        var element : Element?

        syncQueue.sync(flags: .barrier) {

            if elements.startIndex ..< elements.endIndex ~= index {
                element = elements.remove(at: index)
            }
            else {
                element = nil
            }
        }

        return element
    }
}

extension ThreadSafeArray where Element : Equatable {

    public func index(of element: Element) -> Int? {

        return syncQueue.sync {
            elements.index(of: element)
        }
    }
}

Ответ 9

во-первых, objc_sync_enter не работает

objc_sync_enter(array)
defer {
   objc_sync_exit(array)
}

причина, по которой objc_sync_enter/objc_sync_exit не работает с DISPATCH_QUEUE_PRIORITY_LOW

objc_sync_enter является примитивом крайне низкого уровня и не предназначен для непосредственного использования. Это деталь реализации старой @synchronized системы в ObjC.

для swift следует использовать вот так, как сказал @Kirsteins, и я предлагаю синхронизацию вместо async:

private let syncQueue = DispatchQueue(label:"com.test.LockQueue") 
func test(){
    self.syncQueue.sync{
        // thread safe code here
    }
}

Ответ 10

Чтобы улучшить принятый ответ, я бы предложил использовать defer:

objc_sync_enter(array)
defer {
   objc_sync_exit(array)
}
// manipulate the array

и второй

func sync(lock: NSObject, closure: () -> Void) {
    objc_sync_enter(lock)
    defer {
        objc_sync_exit(lock)
    }
    closure()
}