How to limit gcd queue buffer size

Question

I am trying to process series of UIImages using CoreImage & Metal and also display them. My problem is I want to drop the incoming image if my gcd block is busy. How do I achieve this GCD queues, how do I define the maximum buffer size of queue?

Answer 1

There’s no native mechanism for this, but you can achieve what you want algorithmically with semaphores.

Before I dive into the “process 4 at a time, but discard any that come if we’re busy” scenario, let me first consider the simpler “process all, but not more than 4 at any given time” pattern. (I’m going to answer your question below, but building on this simpler situation.)

For example, let’s imagine that you had some preexisting array of objects and you want to process them concurrently, but not more than four at any given time (perhaps to minimize peak memory usage):

DispatchQueue.global().async {
    let semaphore = DispatchSemaphore(value: 4)

    for object in objects {
        semaphore.wait()

        processQueue.async {
            self.process(object)
            semaphore.signal()
        }
    }
}

Basically, the wait function will, as the documentation says, “Decrement the counting semaphore. If the resulting value is less than zero, this function waits for a signal to occur before returning.”

So, we’re starting our semaphore with a count of 4. So if objects had 10 items in it, the first four would start immediately, but the fifth wouldn’t start until one of the earlier ones finished and sent a signal (which increments the semaphore counter back up by 1), and so on, achieving a “run concurrently, but a max of 4 at any given time” behavior.

---

So, let’s return to your question. Let’s say you wanted to process no more than four images at a time and drop any incoming image if there were already four images currently being processed. You can accomplish that by telling wait to not really wait at all, i.e., check right .now() whether the semaphore counter has hit zero already, i.e., something like:

let semaphore = DispatchSemaphore(value: 4)
let processQueue = DispatchQueue(label: "com.domain.app.process", attributes: .concurrent)

func submit(_ image: UIImage) {
    if semaphore.wait(timeout: .now()) == .timedOut { return }

    processQueue.async {
        self.process(image)
        self.semaphore.signal()
    }
}

Note, we generally want to avoid blocking the main thread (like wait can do), but because I’m using a timeout of .now(), it will never block, we’re just use the semaphore to keep track of where we are in a nice, thread-safe manner.

---

One final approach is to consider operation queues:

// create queue that will run no more than four at a time (no semaphores needed; lol)

let processQueue: OperationQueue = {
    let queue = OperationQueue()
    queue.maxConcurrentOperationCount = 4
    return queue
}()

func submit(_ image: UIImage) {
    // cancel all but the last three unstarted operations

    processQueue.operations
        .filter { $0.isReady && !$0.isFinished && !$0.isExecuting && !$0.isCancelled }
        .dropLast(3)
        .forEach { $0.cancel() }

    // now add new operation to the queue

    processQueue.addOperation(BlockOperation {
        self.process(image)
    })
}

The behavior is slightly different (keeping the most recent four images queued up, ready to go), but is something to consider.

Answer 2

GCD queues don't have a maximum queue size.

You can use a semaphore for this. Initialize it with the maximum queue length you want to support. Use dispatch_semaphore_wait() with DISPATCH_TIME_NOW as the timeout to try to reserve a spot before submitting a task to the queue. If it times out, don't enqueue the task (discard it, or whatever). Have the task signal the semaphore when it's complete to release the spot you reserved for it to be used for another task, later.