Queue.swift

//
//  Queue.swift
//
//  An implemntation of a thread safe FIFO queue in Swift 
//

// inspired by 
// https://medium.com/@dmytro.anokhin/concurrency-in-swift-reader-writer-lock-4f255ae73422
// https://www.mikeash.com/pyblog/friday-qa-2011-10-14-whats-new-in-gcd.html
// https://benoitpasquier.com/data-structure-implement-queue-swift/

import Foundation

struct Queue<T> {
    
    // concurrent dispatch queue
    private let dispatch_queue = DispatchQueue(label: "ThreadSafeFIFOQueue.queue", attributes: .concurrent)
    
    private var _elements: [T] = []
    
    @discardableResult
    mutating func enqueue(_ value: T) -> Int {
        // Write with .barrier
        return dispatch_queue.sync(flags: .barrier) {
            self._elements.append(value)
            return self._elements.count
        }
    }
    
    mutating func dequeue() -> T? {
        // write with barrier to ensure only one remove at a time (avoid double dequeuing)
        return dispatch_queue.sync(flags: .barrier) {
            return self._elements.isEmpty ? nil : self._elements.removeFirst()
        }
    }
    
    var head: T? {
        return self._elements.first
    }
    
    var tail: T? {
        return self._elements.last
    }
    
    var isEmpty: Bool {
        return self._elements.isEmpty
    }
    
    var count: Int {
        return self._elements.count
    }
    
    var description: String {
        if self._elements.count <= 20 {
            return "Queue content : \(self._elements)"
        } else {
            return "A queue with \(self._elements.count) elements"
        }
    }
    
}

QueueTest.swift

//
//  QueueTest.swift
//

import XCTest
@testable import BikeTracker

class QueueTest: XCTestCase {
    
    override func setUpWithError() throws {
        // Put setup code here. This method is called before the invocation of each test method in the class.
    }
    
    override func tearDownWithError() throws {
        // Put teardown code here. This method is called after the invocation of each test method in the class.
    }
    
    
    func testAdd1ToQueue() {
        var queue = Queue<String>()
        XCTAssertEqual(queue.count, 0)
        XCTAssert(queue.isEmpty)
        queue.enqueue("1")
        XCTAssertEqual(queue.count, 1)
    }
    
    func testAddSeveralToQueue() {
        var queue = Queue<String>()
        XCTAssert(queue.isEmpty)
        queue.enqueue("1")
        queue.enqueue("1")
        XCTAssertFalse(queue.isEmpty)
    }
    
    func testRemoveOne() {
        var queue = Queue<String>()
        queue.enqueue("1")
        queue.enqueue("")
        queue.enqueue("")
        queue.enqueue("")
        let thefirstone = queue.dequeue()
        
        XCTAssertNotNil(thefirstone)
        XCTAssertEqual(thefirstone!, "1")
    }
    
    func testRemoveAll() {
        var queue = Queue<String>()
        queue.enqueue("1")
        queue.enqueue("2")
        queue.enqueue("3")
        queue.enqueue("4")
        
        XCTAssertEqual(queue.dequeue()!, "1")
        XCTAssertEqual(queue.dequeue()!, "2")
        XCTAssertEqual(queue.dequeue()!, "3")
        XCTAssertEqual(queue.dequeue()!, "4")
        XCTAssert(queue.isEmpty)
        XCTAssertNil(queue.dequeue())
        XCTAssertNil(queue.dequeue())
        XCTAssert(queue.isEmpty)
    }
    
    func testGenerics() {
        var queue = Queue<Int>()
        queue.enqueue(1)
        queue.enqueue(2)
        queue.enqueue(3)
        queue.enqueue(4)
        
        XCTAssertEqual(queue.dequeue()!, 1)
        XCTAssertEqual(queue.dequeue()!, 2)
        XCTAssertEqual(queue.dequeue()!, 3)
        XCTAssertEqual(queue.dequeue()!, 4)
    }
    
    func  testAddNil() {
        var queue = Queue<Int?>()
        queue.enqueue(nil)
        XCTAssertNil(queue.dequeue()!)
        
        queue.enqueue(2)
        queue.enqueue(nil)
        queue.enqueue(4)
        
        XCTAssertEqual(queue.dequeue()!!, 2)
        XCTAssertNil(queue.dequeue()!)
        XCTAssertEqual(queue.dequeue()!!, 4)
    }
    
    func testAddAfterEmpty() {
        var queue = Queue<String>()
        
        queue.enqueue("1")
        XCTAssertEqual(queue.dequeue()!, "1")
        XCTAssertNil(queue.dequeue())
        
        queue.enqueue("1")
        queue.enqueue("2")
        XCTAssertEqual(queue.dequeue()!, "1")
        XCTAssertEqual(queue.dequeue()!, "2")
        XCTAssert(queue.isEmpty)
        XCTAssertNil(queue.dequeue())
    }
    
    func testAddAndRemoveChaotically() {
        var queue = Queue<String>()
        
        queue.enqueue("1")
        XCTAssertFalse(queue.isEmpty)
        XCTAssertEqual(queue.dequeue()!, "1")
        XCTAssert(queue.isEmpty)
        XCTAssertNil(queue.dequeue())
        
        queue.enqueue("1")
        queue.enqueue("2")
        XCTAssertEqual(queue.dequeue()!, "1")
        XCTAssertEqual(queue.dequeue()!, "2")
        XCTAssert(queue.isEmpty)
        XCTAssertNil(queue.dequeue())
        
        queue.enqueue("1")
        queue.enqueue("2")
        XCTAssertEqual(queue.dequeue()!, "1")
        queue.enqueue("3")
        queue.enqueue("4")
        XCTAssertEqual(queue.dequeue()!, "2")
        XCTAssertEqual(queue.dequeue()!, "3")
        XCTAssertFalse(queue.isEmpty)
        XCTAssertEqual(queue.dequeue()!, "4")
        XCTAssertNil(queue.dequeue())
        XCTAssertNil(queue.dequeue())
    }
    
    // I want to test :
    // one thread adds n elements to the queue, another thread remove n elements from the queue
    // ASSERT all elements are removed from the queue one and only once
    // ASSERT all elements retrieved from the queue are in the same order as the elements added in the queue
    
    func testConcurrency() {
        var queue = Queue<Int>()
        
        // prepare source & destination recipients
        let number_of_elements = 100
        
        // serial dispatch queue (pool of threads) to execute "adding" tasks
        let add_dispatch_queue = DispatchQueue(label: "add dispatch queue")
        
        // let's launch n concurrent tasks to add n elements
        // tasks are executed sequentially on the thread, this guarantees the order in the queue
        let add_task_group = DispatchGroup() // semaphore to count adding tasks completion
        for i in 1 ... number_of_elements {
            
            add_task_group.enter()
            
            add_dispatch_queue.async {
                print("Adding \(i)")
                queue.enqueue(i)
                add_task_group.leave()
            }
        }
        
        // another serial dispatch queue (pool of threads) to execute "removing" tasks
        let delete_dispatch_queue = DispatchQueue(label: "delete dispatch queue")
        let delete_task_group = DispatchGroup() // semaphore to count removing tasks completion
        
        // let's launch n concurrent task to remove n elements
        // tasks are executed sequentially on the thread, this guarantees elements should be retrieved in order from the queue
        for i in 1 ... number_of_elements {
            
            delete_task_group.enter()
            
            delete_dispatch_queue.async {
                guard let element = queue.dequeue() else {
                    XCTFail("found nil element in the queue")
                    delete_task_group.leave()
                    return
                }
                print("Removing \(element)")
                XCTAssertEqual(i, element, "elements are not dequeued in the correct order : \(element) should be \(i)")
                delete_task_group.leave()
            }
        }
        
        // wait for the two sets of tasks to terminate. there should never be a timeout.
        print("Wait to finish adding and removing")
        add_task_group.wait()
        delete_task_group.wait()
        print("Finished")
        
        // are all element extracted ?
        XCTAssert(queue.isEmpty)
                
    }
    
    
}